package service

import (
	"context"
	"encoding/base64"
	"fmt"

	"github.com/pingcap/errors"
	filterutil "github.com/pingcap/tidb/pkg/util/filter"
	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy/api"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
)

const (
	// QueryEligibleSchemaTableClaus is the query clause to get tables that can be replicated by TiCDC
	QueryEligibleSchemaTableClaus = `
SELECT table_schema, table_name
FROM information_schema.tables
WHERE (table_schema, table_name) IN
      (SELECT table_schema, table_name
       FROM information_schema.columns
       WHERE COLUMN_KEY = 'PRI'
          OR COLUMN_KEY = 'UNI'
       union all
       SELECT table_schema, table_name
       FROM information_schema.tidb_indexes
       WHERE NON_UNIQUE = 0)
  AND table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT',
                           'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA',
                           'METRICS_SCHEMA', 'INSPECTION_SCHEMA', 'sys', 'mysql', 'heartbeat',
                           'dm_heartbeat', 'information_schema', 'performance_schema',
                           'metrics_schema', 'inspection_schema');
`
	// QueryIneligibleSchemaTableClaus is the query clause to get tables that can not be replicated by TiCDC
	QueryIneligibleSchemaTableClaus = `
SELECT table_schema, table_name
FROM information_schema.tables
WHERE (table_schema, table_name) NOT IN
      (select distinct table_schema, table_name
       from (SELECT table_schema, table_name
             FROM information_schema.columns
             WHERE COLUMN_KEY = 'PRI'
                OR COLUMN_KEY = 'UNI'
             union all
             SELECT table_schema, table_name
             FROM information_schema.tidb_indexes
             where NON_UNIQUE = 0) d)
  AND table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT',
                           'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA',
                           'METRICS_SCHEMA', 'INSPECTION_SCHEMA', 'sys', 'mysql', 'heartbeat',
                           'dm_heartbeat', 'information_schema', 'performance_schema',
                           'metrics_schema', 'inspection_schema');
`

	// TiCDCSystemSchema is the schema only use by TiCDC.
	TiCDCSystemSchema = "tidb_cdc"
	// LightningTaskInfoSchema is the schema only generated by Lightning
	LightningTaskInfoSchema = "lightning_task_info"

	// test connection types
	connectionTypeKafka            = "kafka"
	connectionTypeMysql            = "mysql"
	connectionTypeS3               = "s3"
	connectionTypeGCS              = "gcs"
	connectionTypeAzureBlobStorage = "azblob"
	connectionTypePulsar           = "pulsar"
)

type TableSchemaInfo struct {
	TableSchema []string `json:"table_schema"`
	TableName   []string `json:"table_name"`
}

type preCheckServerImpl struct {
	proxyModel clusterproxy.Model
}

func NewPreCheckServiceServer(proxyModel clusterproxy.Model) apiv1.TiCDCPrecheckServiceServer {
	s := &preCheckServerImpl{
		proxyModel: proxyModel,
	}
	return s
}

func (p *preCheckServerImpl) TestNetworkConnection(ctx context.Context, req *apiv1.TestNetworkConnectionRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	proxy, err := p.proxyModel.GetClusterProxy(ctx, req.ClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.ClusterId, err)
	}
	connectionModel := buildTestConnectionModel(req)
	if err := proxy.ChangefeedClient().TestConnection(ctx, connectionModel); err != nil {
		logger.Error("Failed to test connection", zap.Error(err))
		// FIXME: should return a more specific error
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (p *preCheckServerImpl) ValidateChangefeedParameters(ctx context.Context, req *apiv1.ValidateChangefeedParametersRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))
	proxy, err := p.proxyModel.GetClusterProxy(ctx, req.ClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.ClusterId, err)
	}

	// if the start ts is 0, it means the changefeed is a new changefeed replicating data from the latest ts, do not need to check the safepoint
	if req.StartTso > 0 {
		logger.Info("Checking PD safepoint", zap.Uint64("startTso", req.StartTso))
		safePoint, err := proxy.PDClient().GetPDSafepoint(ctx)
		if err != nil {
			logger.Error("Failed to get PD safepoint", zap.Error(err))
			return nil, err
		}
		if safePoint.SP > req.StartTso {
			logger.Error("StartTso is less than PD safepoint",
				zap.Uint64("startTso", req.StartTso),
				zap.Uint64("safepoint", safePoint.SP))
			return nil, status.Error(codes.InvalidArgument, "StartTso is less than PD safepoint")
		}
		logger.Info("PD safepoint checked", zap.Uint64("safepoint", safePoint.SP), zap.Uint64("startTso", req.StartTso))
	}

	if req.KafkaParam == nil {
		logger.Info("Kafka parameter is nil, skipping connection test")
		return &emptypb.Empty{}, nil
	}

	logger.Info("Starting to test connection")
	connectionModel := &api.ConnectionModel{
		Type:      req.Type,
		User:      req.KafkaParam.User,
		Password:  req.KafkaParam.Password,
		Url:       req.KafkaParam.Url,
		EnableTls: req.KafkaParam.EnableTls,
		AuthType:  req.KafkaParam.AuthType,
		Topic:     req.KafkaParam.Topic,
		KafkaItem: api.KafkaItem{
			ReplicationFactor:       req.KafkaParam.ReplicationFactor,
			PartitionNum:            req.KafkaParam.PartitionNum,
			SchemaRegistryEndpoints: req.KafkaParam.SchemaRegistryEndpoints,
		},
	}
	if err := proxy.ChangefeedClient().TestConnection(ctx, connectionModel); err != nil {
		logger.Error("Failed to validate changefeed parameter", zap.Error(err))
		// FIXME: should return a more specific error
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (p *preCheckServerImpl) VerifyTable(ctx context.Context, req *apiv1.VerifyTableRequest) (*apiv1.VerifyTableResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TidbClusterId))
	filterRules := req.FilterRules
	if len(filterRules) == 0 {
		filterRules = []string{"*.*"}
	}

	logger.Info("verify table ", zap.Strings("filter", filterRules))
	tblFilter, err := tfilter.Parse(filterRules)
	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "Failed to parse filters: %s", err)
	}
	if !req.CaseSensitive {
		tblFilter = tfilter.CaseInsensitive(tblFilter)
	}

	proxy, err := p.proxyModel.GetClusterProxy(ctx, req.TidbClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.TidbClusterId, err)
	}

	eligibleTables, err := queryAndFilterTableSchema(ctx, tblFilter, proxy.SQLClient(), req.User, req.Password, QueryEligibleSchemaTableClaus)
	if err != nil {
		logger.Error("failed query eligible schema table", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to query eligible schema table: %s", err)
	}

	ineligibleTable, err := queryAndFilterTableSchema(ctx, tblFilter, proxy.SQLClient(), req.User, req.Password, QueryIneligibleSchemaTableClaus)
	if err != nil {
		logger.Error("failed query ineligible schema table", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed query ineligible schema table: %s", err)
	}
	logger.Info("verify table finished",
		zap.Int("ineligibleTable", len(ineligibleTable)),
		zap.Int("eligibleTables", len(eligibleTables)),
	)
	return &apiv1.VerifyTableResponse{
		IneligibleTables: ineligibleTable,
		EligibleTables:   eligibleTables,
	}, nil
}

func queryAndFilterTableSchema(
	ctx context.Context,
	filter tfilter.Filter,
	client api.SQLClient,
	user string,
	password string,
	sql string,
) (queryTable []string, err error) {

	var queryInfo TableSchemaInfo
	if err := client.QuerySQL(ctx, &api.ClusterOperator{
		Username:   user,
		AuthMethod: api.AuthMethodPassword,
		Credential: password,
	}, sql, &queryInfo); err != nil {
		return nil, errors.Trace(err)
	}
	return parseQueryTable(&queryInfo, filter)
}

func parseQueryTable(queryInfo *TableSchemaInfo, filter tfilter.Filter) ([]string, error) {
	var queryTable []string
	for i := 0; i < len(queryInfo.TableSchema); i++ {
		schema := queryInfo.TableSchema[i]
		table := queryInfo.TableName[i]
		schemaRet, err := base64.StdEncoding.DecodeString(schema)
		if err != nil {
			return nil, err
		}
		tableRet, err := base64.StdEncoding.DecodeString(table)
		if err != nil {
			return nil, err
		}
		if !isSysSchema(string(schemaRet)) && filter.MatchTable(string(schemaRet), string(tableRet)) {
			queryTable = append(queryTable, fmt.Sprintf("%s.%s", schemaRet, tableRet))
		}
	}
	return queryTable, nil
}

func buildTestConnectionModel(req *apiv1.TestNetworkConnectionRequest) *api.ConnectionModel {
	connectionModel := &api.ConnectionModel{Type: req.Type}
	switch req.Type {
	case connectionTypeMysql:
		connectionModel.User = req.MysqlParam.User
		connectionModel.Password = req.MysqlParam.Password
		connectionModel.Url = req.MysqlParam.Url
	case connectionTypeKafka:
		connectionModel.User = req.KafkaParam.User
		connectionModel.Password = req.KafkaParam.Password
		connectionModel.Url = req.KafkaParam.BoostrapEndpoints
		connectionModel.EnableTls = req.KafkaParam.EnableTls
		connectionModel.AuthType = req.KafkaParam.AuthType
	case connectionTypeS3:
		connectionModel.S3Param = api.S3Parameter{
			S3Uri:           req.S3Param.S3Uri,
			AccessKeyId:     req.S3Param.AccessKeyId,
			SecretAccessKey: req.S3Param.SecretAccessKey,
			RoleArn:         req.S3Param.RoleArn,
			ExternalID:      req.S3Param.ExternalId,
		}
	case connectionTypeGCS:
		connectionModel.GCSParam = api.GcsParameter{
			GcsUrl:                    req.GcsParam.GcsUrl,
			Region:                    req.GcsParam.Region,
			ImpersonateServiceAccount: req.GcsParam.ImpersonateServiceAccount,
		}
	case connectionTypeAzureBlobStorage:
		connectionModel.AzureBlobStorageParam = api.AzureBlobStorageParameter{
			BlobUri:  req.AzureBlobStorageParam.BlobUri,
			SASToken: req.AzureBlobStorageParam.SasToken,
		}
	case connectionTypePulsar:
		connectionModel.PulsarParam = api.PulsarParameter{
			ServiceURL:         req.GetPulsarParam().GetServiceUrl(),
			AuthenticationType: req.GetPulsarParam().GetAuthenticationType(),
			Username:           req.GetPulsarParam().GetUsername(),
			Password:           req.GetPulsarParam().GetPassword(),
			Token:              req.GetPulsarParam().GetToken(),
			ConnectionTimeout:  req.GetPulsarParam().GetConnectionTimeout(),
			Tenant:             req.GetPulsarParam().GetTenant(),
			Namespace:          req.GetPulsarParam().GetNamespace(),
			Topic:              req.GetPulsarParam().GetTopic(),
		}
	}
	return connectionModel
}

// isSysSchema returns true if the given schema is a system schema
func isSysSchema(db string) bool {
	switch db {
	// TiCDCSystemSchema is used by TiCDC only.
	// Tables in TiCDCSystemSchema should not be replicated by cdc.
	case TiCDCSystemSchema:
		return true
	// Tables in LightningTaskInfoSchema should not be replicated by cdc.
	case LightningTaskInfoSchema:
		return true
	default:
		return filterutil.IsSystemSchema(db)
	}
}