// Copyright 2020 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package filter

import (
	"sync"

	"github.com/gogo/protobuf/proto"
	"github.com/pingcap/log"
	"github.com/pingcap/ticdc/eventpb"
	bf "github.com/pingcap/ticdc/pkg/binlog-filter"
	"github.com/pingcap/ticdc/pkg/common"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/util"
	timodel "github.com/pingcap/tidb/pkg/meta/model"
	"github.com/pingcap/tidb/pkg/util/chunk"
	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
	"go.uber.org/zap"
)

const (
	// SyncPointTable is the table name use to write ts-map when sync-point is enable.
	SyncPointTable = "syncpoint_v1"
	// DDLTsTable is the table name use to write ddl commitTs for each table when downstream is mysql-class
	DDLTsTable = "ddl_ts_v1"

	// TiCDCSystemSchema is the schema only use by TiCDC.
	TiCDCSystemSchema = "tidb_cdc"
	// LightningTaskInfoSchema is the schema only generated by Lightning
	LightningTaskInfoSchema = "lightning_task_info"
	// TiDBWorkloadSchema is the schema that needs to be skipped by TiCDC.
	TiDBWorkloadSchema = "workload_schema"
)

// Filter are safe for concurrent use.
type Filter interface {
	// ShouldIgnoreDML returns true if the DML event should not be handled.
	ShouldIgnoreDML(dmlType common.RowType, preRow, row chunk.Row, tableInfo *common.TableInfo, startTs uint64) (bool, error)
	// ShouldDiscardDDL returns true if the DDL event should not be handled.
	ShouldDiscardDDL(schema, table string, ddlType timodel.ActionType, tableInfo *common.TableInfo) bool
	// ShouldIgnoreDDL returns true if the DDL event should not be sent to downstream.
	//
	// If a ddl is ignored, it will be sent to table trigger dispatcher to update the schema or table info,
	// but will not be sent to downstream.
	// Note that a ignored ddl is different from a discarded ddl. For example, suppose
	// we have a changefeed-test with the following config:
	//   - table filter: rules = ['test.*']
	//   - event-filters: matcher = ["test.worker"] ignore-event = ["create table"]
	//
	// Then, for the following DDLs:
	//  1. `CREATE TABLE test.worker` will be ignored, but the table will be replicated by changefeed-test.
	//  2. `CREATE TABLE other.worker` will be discarded, and the table will not be replicated by changefeed-test.
	ShouldIgnoreDDL(schema, table, query string, ddlType timodel.ActionType, startTs uint64) (bool, error)
	// ShouldIgnoreTable returns true if the table should be ignored.
	ShouldIgnoreTable(schema, table string) bool
	// ShouldIgnoreSchema returns true if the schema should be ignored.
	ShouldIgnoreSchema(schema string) bool
	// IsEligibleTable returns true if the table is eligible to be replicated.
	IsEligibleTable(tableInfo *common.TableInfo) bool
	// Verify should only be called by create changefeed OpenAPI.
	// Its purpose is to verify the expression filter config.
	Verify(tableInfos []*common.TableInfo) error
}

// filter implements Filter.
type filter struct {
	// tableFilter is used to filter in dml/ddl event by table name.
	tableFilter tfilter.Filter
	// dmlExprFilter is used to filter out dml event by its columns value.
	dmlExprFilter *dmlExprFilter
	// sqlEventFilter is used to filter out dml/ddl event by its type or query.
	sqlEventFilter *sqlEventFilter
	// ignoreTxnStartTs is used to filter out dml/ddl event by its starsTs.
	ignoreTxnStartTs []uint64
	forceReplicate   bool
}

// NewFilter creates a filter.
func NewFilter(cfg *config.FilterConfig, tz string, caseSensitive bool, forceReplicate bool) (Filter, error) {
	f, err := VerifyTableRules(cfg)
	if err != nil {
		return nil, err
	}

	if !caseSensitive {
		f = tfilter.CaseInsensitive(f)
	}

	dmlExprFilter, err := newExprFilter(tz, cfg)
	if err != nil {
		return nil, err
	}
	sqlEventFilter, err := newSQLEventFilter(cfg, caseSensitive)
	if err != nil {
		return nil, err
	}
	return &filter{
		tableFilter:      f,
		dmlExprFilter:    dmlExprFilter,
		sqlEventFilter:   sqlEventFilter,
		ignoreTxnStartTs: cfg.IgnoreTxnStartTs,
		forceReplicate:   forceReplicate,
	}, nil
}

// ShouldIgnoreDML checks if a DML event should be ignore by conditions below:
// 1. By startTs.
// 2. By table name.
// 3. By type.
func (f *filter) ShouldIgnoreDML(dmlType common.RowType, preRow, row chunk.Row, tableInfo *common.TableInfo, startTs uint64) (bool, error) {
	if f.shouldIgnoreStartTs(startTs) {
		return true, nil
	}

	if f.ShouldIgnoreTable(tableInfo.GetSchemaName(), tableInfo.GetTableName()) {
		return true, nil
	}

	ignoreByEventType, err := f.sqlEventFilter.shouldSkipDML(tableInfo.GetSchemaName(), tableInfo.GetTableName(), dmlType)
	if err != nil {
		return false, err
	}
	if ignoreByEventType {
		return true, nil
	}
	return f.dmlExprFilter.shouldSkipDML(dmlType, preRow, row, tableInfo)
}

// ShouldDiscardDDL checks if a DDL event should be discarded by conditions below:
// 1. By allow list.
// 2. By startTs.
// 3. By schema name.
// 4. By table name.
// 5. By eligibility of the table.
func (f *filter) ShouldDiscardDDL(schema, table string, ddlType timodel.ActionType, tableInfo *common.TableInfo) bool {
	if !isAllowedDDL(ddlType) {
		return true
	}

	// If the DDL is a schema DDL, we should ignore it if the schema is not allowed.
	if IsSchemaDDL(ddlType) {
		return f.ShouldIgnoreSchema(schema)
	}

	if f.ShouldIgnoreTable(schema, table) {
		return true
	}

	if !f.IsEligibleTable(tableInfo) {
		log.Info("table is not eligible, should discard this ddl", zap.String("schema", tableInfo.GetSchemaName()), zap.String("table", tableInfo.GetTableName()),
			zap.Bool("forceReplicate", f.forceReplicate), zap.Bool("hasPKOrNotNullUK", tableInfo.HasPKOrNotNullUK))
		return true
	}
	return false
}

// ShouldIgnoreDDL checks if a DDL event should be ignore by conditions below:
// 1. By ddl type.
// 2. By ddl query.
func (f *filter) ShouldIgnoreDDL(schema, table, query string, ddlType timodel.ActionType, startTs uint64) (bool, error) {
	if f.shouldIgnoreStartTs(startTs) {
		return true, nil
	}
	return f.sqlEventFilter.shouldSkipDDL(schema, table, query, ddlType)
}

// ShouldIgnoreTable returns true if the specified table should be ignored by this changefeed.
// NOTICE: Set `tbl` to an empty string to test against the whole database.
func (f *filter) ShouldIgnoreTable(db, tbl string) bool {
	if IsSysSchema(db) {
		return true
	}

	return !f.tableFilter.MatchTable(db, tbl)
}

// ShouldIgnoreSchema returns true if the specified schema should be ignored by this changefeed.
func (f *filter) ShouldIgnoreSchema(schema string) bool {
	return IsSysSchema(schema) || !f.tableFilter.MatchSchema(schema)
}

// TODO
func (f *filter) Verify(tableInfos []*common.TableInfo) error {
	return nil
	// return f.dmlExprFilter.verify(tableInfos)
}

func (f *filter) IsEligibleTable(tableInfo *common.TableInfo) bool {
	if tableInfo == nil {
		// This should only happen in test
		return true
	}
	return tableInfo.IsEligible(f.forceReplicate)
}

func (f *filter) shouldIgnoreStartTs(ts uint64) bool {
	for _, ignoreTs := range f.ignoreTxnStartTs {
		if ignoreTs == ts {
			return true
		}
	}
	return false
}

func isAllowedDDL(actionType timodel.ActionType) bool {
	_, ok := ddlWhiteListMap[actionType]
	return ok
}

// IsSchemaDDL returns true if the action type is a schema DDL.
func IsSchemaDDL(actionType timodel.ActionType) bool {
	switch actionType {
	case timodel.ActionCreateSchema, timodel.ActionDropSchema,
		timodel.ActionModifySchemaCharsetAndCollate:
		return true
	default:
		return false
	}
}

var (
	once    sync.Once
	storage *SharedFilterStorage
)

type FilterWithConfig struct {
	Filter
	config   *eventpb.FilterConfig
	timeZone string
}

type SharedFilterStorage struct {
	// Each dispatcher in the same changefeed will share the same filter storage.
	m     map[common.ChangeFeedID]FilterWithConfig
	mutex sync.Mutex
}

func GetSharedFilterStorage() *SharedFilterStorage {
	once.Do(func() {
		storage = &SharedFilterStorage{
			m: make(map[common.ChangeFeedID]FilterWithConfig),
		}
	})
	return storage
}

func (s *SharedFilterStorage) GetOrSetFilter(
	changeFeedID common.ChangeFeedID,
	cfg *eventpb.FilterConfig,
	timeZone string,
) (Filter, error) {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	if f, ok := s.m[changeFeedID]; ok {
		if !isFilterConfigEqual(f.config, cfg) || f.timeZone != timeZone {
			log.Info("filter config changed, need to rebuild filter", zap.Any("preConfig", f.config), zap.Any("newConfig", cfg), zap.String("preTimeZone", f.timeZone), zap.String("newTimeZone", timeZone))
		} else {
			return f.Filter, nil
		}
	}
	// convert eventpb.FilterConfig to config.FilterConfig
	filterCfg := &config.FilterConfig{
		Rules:            cfg.FilterConfig.Rules,
		IgnoreTxnStartTs: cfg.FilterConfig.IgnoreTxnStartTs,
	}
	for _, rule := range cfg.FilterConfig.EventFilters {
		f := &config.EventFilterRule{
			Matcher:                  rule.Matcher,
			IgnoreSQL:                rule.IgnoreSql,
			IgnoreInsertValueExpr:    util.AddressOf(rule.IgnoreInsertValueExpr),
			IgnoreUpdateNewValueExpr: util.AddressOf(rule.IgnoreUpdateNewValueExpr),
			IgnoreUpdateOldValueExpr: util.AddressOf(rule.IgnoreUpdateOldValueExpr),
			IgnoreDeleteValueExpr:    util.AddressOf(rule.IgnoreDeleteValueExpr),
		}
		for _, e := range rule.IgnoreEvent {
			f.IgnoreEvent = append(f.IgnoreEvent, bf.EventType(e))
		}

		filterCfg.EventFilters = append(filterCfg.EventFilters, f)
	}
	// generate table filter
	f, err := NewFilter(filterCfg, timeZone, cfg.CaseSensitive, cfg.ForceReplicate)
	if err != nil {
		return nil, err
	}
	s.m[changeFeedID] = FilterWithConfig{
		Filter:   f,
		config:   cfg,
		timeZone: timeZone,
	}
	return f, nil
}

// isFilterConfigEqual compares two FilterConfig for equality by content
func isFilterConfigEqual(cfg1, cfg2 *eventpb.FilterConfig) bool {
	// Fast path: if pointers are equal, configs are definitely equal
	if cfg1 == cfg2 {
		return true
	}

	// Handle nil cases
	if cfg1 == nil || cfg2 == nil {
		return false
	}

	// Use protobuf's built-in Equal method for content comparison
	return proto.Equal(cfg1, cfg2)
}