package service

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/grpc/grpcli"
	"github.com/tidbcloud/serverless-service/internal/utils"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	"go.uber.org/zap"
)

const (
	changefeedRetentionDays = 30
	cdcClusterRetentionDays = 30
)

type CDCWorker interface {
	CleanChangefeeds(ctx context.Context) (err error)
	SyncCreatingChangefeedsAndCDCCluster(ctx context.Context) (err error)
	SyncChangefeeds(ctx context.Context) (err error)
}

type cdcWorker struct {
	clusterProxyProviderFacade  facade.ClusterProxyProviderFacade
	clusterRepo                 repo.ClusterRepo
	changefeedRepo              repo.ChangefeedRepo
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo
	tiCDCClusterRepo            repo.TiCDCClusterRepo
	cdcClusterRepo              repo.CDCClusterRepo
	transProvider               db.TransProvider
	mgmtSvc                     facade.MgmtSvcFacade
	encryptor                   facade.EncryptorFacade
	configCenter                grpcli.ConfigCenterClient
}

func NewCDCWorker(
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	clusterRepo repo.ClusterRepo,
	changefeedRepo repo.ChangefeedRepo,
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo,
	tiCDCClusterRepo repo.TiCDCClusterRepo,
	cdcClusterRepo repo.CDCClusterRepo,
	transProvider db.TransProvider,
	mgmtSvc facade.MgmtSvcFacade,
	encryptor facade.EncryptorFacade,
	configCenter grpcli.ConfigCenterClient,
	_ appworkflow.Workflow,
) CDCWorker {
	return &cdcWorker{
		clusterProxyProviderFacade:  clusterProxyProviderFacade,
		clusterRepo:                 clusterRepo,
		changefeedRepo:              changefeedRepo,
		tiCDCChangefeedResourceRepo: tiCDCChangefeedResourceRepo,
		tiCDCClusterRepo:            tiCDCClusterRepo,
		cdcClusterRepo:              cdcClusterRepo,
		transProvider:               transProvider,
		mgmtSvc:                     mgmtSvc,
		encryptor:                   encryptor,
		configCenter:                configCenter,
	}
}

// CleanChangefeeds cleans changefeeds and cdc clusters
func (w *cdcWorker) CleanChangefeeds(ctx context.Context) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("cleaning cdc changefeeds")

	count, err := w.changefeedRepo.CleanByRetention(ctx, changefeedRetentionDays)
	if err != nil {
		logger.Error("failed to clean cdc changefeeds", zap.Error(err))
		return err
	}
	logger.Info("cleaned cdc changefeeds", zap.Int64("count", count))

	count, err = w.cdcClusterRepo.CleanByRetention(ctx, cdcClusterRetentionDays)
	if err != nil {
		logger.Error("failed to clean cdc changefeeds", zap.Error(err))
		return err
	}
	logger.Info("cleaned cdc clusters", zap.Int64("count", count))
	return nil
}

// SyncCreatingChangefeedsAndCDCCluster syncs changefeeds that are in the creating state. it also create or delete cdc cluster when needed.
// we async the creation of changefeed because we need to set up the cdc cluster when the first changefeed is created.
// the cdc cluster deletion is also handled in this function to avoid race between diffierent instances and goroutines.
func (w *cdcWorker) SyncCreatingChangefeedsAndCDCCluster(ctx context.Context) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("syncing creating cdc changefeeds")

	if err := w.syncNewArchChangefeedResources(ctx); err != nil {
		logger.Error("failed to sync newarch changefeed resources", zap.Error(err))
		return err
	}
	// Opportunistic GC: prune resource rows that have been released
	// (expected and actual specs are both empty) for at least 7 days.
	// Best-effort — a transient DB error here must not block the rest
	// of the sync cycle. Gated on cdc.newArch.enabled because a pod
	// without the feature shouldn't be touching newarch tables.
	if config.Global().CDCConfig.NewArch.Enabled {
		if n, err := w.tiCDCChangefeedResourceRepo.DeleteStaleReleasedRows(ctx, 7*24*time.Hour); err != nil {
			logger.Warn("failed to gc stale newarch changefeed resources", zap.Error(err))
		} else if n > 0 {
			logger.Info("gc'd stale newarch changefeed resources", zap.Int64("rows", n))
		}
	}

	changefeeds, err := w.changefeedRepo.ListByFilter(ctx, entity.ListChangefeedsFilter{
		State: constant.ChangefeedStateCreating,
	})
	if err != nil {
		logger.Error("failed to list creating cdc changefeeds", zap.Error(err))
		return err
	}
	var syncFailClusters []string
	for _, changefeed := range changefeeds {
		if err := w.syncOneCreatingChangefeed(ctx, changefeed); err != nil {
			logger.Error("failed to sync creating changefeed",
				zap.String("changefeedId", changefeed.ID),
				zap.String("clusterId", changefeed.ClusterID),
				zap.Error(err))
			syncFailClusters = append(syncFailClusters, changefeed.ClusterID)
		}
	}

	// delete cdc clusters without changefeed
	cdcClusters, err := w.cdcClusterRepo.ListCDCWithoutChangefeed(ctx)
	if err != nil {
		logger.Error("failed to list cdc clusters without changefeed", zap.Error(err))
		return err
	}
	var deleteCDCFailClusters []string
	for _, cdc := range cdcClusters {
		if err := w.deleteCDCCluster(ctx, cdc); err != nil {
			logger.Error("failed to delete cdc cluster",
				zap.String("clusterId", cdc.ClusterID),
				zap.Error(err))
			deleteCDCFailClusters = append(deleteCDCFailClusters, cdc.ClusterID)
		}
	}

	logger.Info("syncing creating cdc changefeeds done",
		zap.Int("sync creating failed", len(syncFailClusters)),
		zap.Int("delete cdc cluster failed", len(deleteCDCFailClusters)))

	if len(syncFailClusters) > 0 || len(deleteCDCFailClusters) > 0 {
		return fmt.Errorf("sync creating changefeed failed clusters: %v, delete cdc failed clusters: %v",
			syncFailClusters,
			deleteCDCFailClusters)
	}
	return nil
}

func (w *cdcWorker) deleteCDCCluster(ctx context.Context, cdc *entity.CDCCluster) (err error) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", cdc.ClusterID))
	logger.Info("try to delete cdc cluster", zap.String("state", string(cdc.State)))

	uint64ClusterId, err := strconv.ParseUint(cdc.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return err
	}

	proxy, err := w.getServerlessClusterProxy(ctx, cdc.CloudProvider, cdc.Region, cdc.Kubernetes)
	if err != nil {
		logger.Error("failed to get cluster porxy", zap.Error(err))
	}

	// 1. delete cdc cluster
	// 2. delete cluster custom status and update state to deleted
	if cdc.State == constant.CDCClusterStateRunning {
		if err := proxy.DeleteServerlessCDC(ctx, cdc.KeySpaceID); err != nil {
			logger.Error("failed to delete cdc cluster", zap.Error(err))
			return err
		}
	}

	return w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := w.cdcClusterRepo.UpdateState(ctx, cdc.ClusterID, constant.CDCClusterStateDeleted); err != nil {
			logger.Error("failed to update cdc cluster state to running", zap.Error(err))
			return err
		}
		if err := w.cdcClusterRepo.SetClusterStatusId(ctx, cdc.ClusterID, 0); err != nil {
			logger.Error("failed to set cluster status id", zap.Error(err))
			return err
		}
		if err = w.mgmtSvc.DeleteClusterCustomStatus(ctx, uint64ClusterId, cdc.ClusterStatusId); err != nil {
			logger.Error("failed to delete cdc cluster custom status",
				zap.Uint64("cluster_status_id", cdc.ClusterStatusId),
				zap.Error(err))
			return err
		}
		return nil
	})
}

func (w *cdcWorker) createCDCCluster(
	ctx context.Context,
	cdc *entity.CDCCluster,
	proxy facade.ServerlessClusterProxyFacade,
) (err error) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", cdc.ClusterID))
	logger.Info("try to create cdc cluster", zap.String("state", string(cdc.State)))

	uint64ClusterId, err := strconv.ParseUint(cdc.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return err
	}

	// 1. set cluster custom status, skip if already set
	// 2. create cdc cluster and update state to running
	if cdc.ClusterStatusId == 0 {
		statusId, err := w.mgmtSvc.SetClusterCDCStatus(ctx, uint64ClusterId)
		if err != nil {
			logger.Error("failed to set cdc cluster custom status", zap.Error(err))
			return err
		}
		logger.Info("set cdc cluster custom status", zap.Uint64("statusId", statusId))

		if err := w.cdcClusterRepo.SetClusterStatusId(ctx, cdc.ClusterID, statusId); err != nil {
			logger.Error("failed to set cluster status id, start to rollback custom status", zap.Error(err))
			if delErr := w.mgmtSvc.DeleteClusterCustomStatus(ctx, uint64ClusterId, statusId); delErr != nil {
				logger.Error("failed to delete cdc cluster custom status, please handle it manually", zap.Error(delErr))
			}
			return err
		}
	}

	return w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := w.cdcClusterRepo.UpdateState(ctx, cdc.ClusterID, constant.CDCClusterStateRunning); err != nil {
			logger.Error("failed to set cluster cdc to running", zap.Error(err))
			return err
		}
		if err := proxy.CreateServerlessCDC(ctx, cdc.KeySpaceID); err != nil {
			logger.Error("failed to create cdc cluster", zap.Error(err))
			return err
		}
		logger.Info("create cdc cluster success", zap.Error(err))
		return nil
	})
}

func (w *cdcWorker) syncOneCreatingChangefeed(ctx context.Context, changefeed *entity.CDCChangefeed) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("changefeedId", changefeed.ID), zap.String("clusterId", changefeed.ClusterID))

	logger.Info("syncing one creating cdc changefeed")

	defer func() {
		// handle failed case
		var e bizerr.BizError
		if err != nil && errors.As(err, &e) {
			if e.ErrCode() == errorcode.CreateChangefeedUnretryableErr {
				changefeed.State = constant.ChangefeedStateCreateFailed
				changefeed.Error = fmt.Sprintf("failed to create changefeed, please create a new one: %s", e.ErrMsg())
				if e := w.changefeedRepo.ChangeStateFromCreating(ctx, changefeed); e != nil {
					logger.Error("failed to sync state to create_failed", zap.Error(e))
				}
			}
		}
	}()

	if isNewArchChangefeed(changefeed) {
		logger.Info("skip creating newarch changefeed in legacy polling worker")
		return nil
	}

	proxy, err := w.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if err != nil {
		logger.Error("failed to get cluster porxy", zap.Error(err))
		return err
	}

	// check cdc cluster
	cdcCluster, bizErr := w.cdcClusterRepo.GetById(ctx, changefeed.ClusterID)
	if bizErr != nil {
		logger.Error("failed to get cdc cluster", zap.Error(bizErr))
		return bizErr
	}

	// create cdc cluster
	if cdcCluster.State == constant.CDCClusterStateCreating || cdcCluster.State == constant.CDCClusterStateDeleted {
		if err := w.createCDCCluster(ctx, cdcCluster, proxy); err != nil {
			logger.Error("failed to create cdc cluster", zap.Error(err))
			return err
		}
	}

	// create changefeed
	changefeedConfig, err := changefeed.ToChangefeed(ctx, w.encryptor)
	if err != nil {
		logger.Error("failed to convert changefeed to changefeed config", zap.Error(err))
		return err
	}
	changefeedConfig, err = assembleLegacyChangefeedConfig(
		ctx, w.configCenter, changefeed, changefeedConfig, nil,
	)
	if err != nil {
		logger.Error("failed to assemble effective changefeed config", zap.Error(err))
		return err
	}
	err = w.transProvider.Transaction(ctx, func(ctx context.Context) error {
		changefeed.ChangefeedID = *changefeedConfig.ID
		changefeed.State = constant.ChangfeedStateRunning
		if err = w.changefeedRepo.ChangeStateFromCreating(ctx, changefeed); err != nil {
			logger.Error("sync state to running failed", zap.Error(err))
			return err
		}
		cf, err := proxy.CreateServerlessChangefeed(ctx, changefeed.KeySpaceID, changefeedConfig)
		if err != nil {
			logger.Error("failed to create changefeed", zap.Error(err))
			return err
		}
		logger.Info("create changefeed success", zap.Any("changefeed", cf))
		return nil
	})

	return err
}

func (w *cdcWorker) getServerlessClusterProxy(
	ctx context.Context,
	provider, region, k8s string,
) (facade.ServerlessClusterProxyFacade, bizerr.BizError) {
	clusterProxyUrl := config.GetClusterProxyUrl(provider, region, k8s)
	if clusterProxyUrl == "" {
		regionInfo := fmt.Sprintf("%s.%s.%s", provider, region, k8s)
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster proxy url not found in config for: %s", regionInfo)
	}
	return w.clusterProxyProviderFacade.GetServerlessHTTPProxy(ctx, clusterProxyUrl, nil, nil)
}

func (w *cdcWorker) SyncChangefeeds(ctx context.Context) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("syncing cdc changefeeds")

	// Skip states that don't expect a kernel-side row to exist:
	//   - creating / create_failed: kernel hasn't been called yet
	//   - deleted: tombstone
	//   - deleting: the delete DAG owns the deleting -> deleted
	//     transition. If we don't exclude it here, this sync sees no
	//     kernel row (either the delete DAG already removed it or it
	//     was never created in the first place) and races by writing
	//     state=deleted via SyncFromChangefeedCommon(nil), which then
	//     makes the delete DAG's GetById (where state != 'deleted')
	//     fail and the DAG retries forever.
	changefeeds, err := w.changefeedRepo.ListByFilter(ctx, entity.ListChangefeedsFilter{
		StatesNotIn: []constant.ChangfeedState{
			constant.ChangefeedStateCreating,
			constant.ChangefeedStateCreateFailed,
			constant.ChangefeedStateDeleting,
			constant.ChangefeedStateDeleted,
		},
	})
	if err != nil {
		logger.Error("failed to list running cdc changefeeds", zap.Error(err))
		return err
	}
	clusterToChangefeeds := w.groupChangefeedsByClusterArchAndShard(ctx, changefeeds)

	// concurrently sync changefeeds in the future if needed
	syncFailedClusters := make([]string, 0)
	for cluster, changefeeds := range clusterToChangefeeds {
		if err := w.SyncChangefeedsForOneCluster(ctx, changefeeds); err != nil {
			logger.Error("failed to sync changefeeds for one cluster", zap.Error(err))
			syncFailedClusters = append(syncFailedClusters, fmt.Sprintf("%s/%s", cluster.clusterID, cluster.arch))
			continue
		}
	}
	if len(syncFailedClusters) > 0 {
		return fmt.Errorf("failed to sync changefeeds for clusters: %v", syncFailedClusters)
	}
	return nil
}

type changefeedSyncGroupKey struct {
	clusterID string
	arch      constant.ChangefeedArch
	shard     string
}

func (w *cdcWorker) groupChangefeedsByClusterArchAndShard(
	ctx context.Context,
	changefeeds []*entity.CDCChangefeed,
) map[changefeedSyncGroupKey][]*entity.CDCChangefeed {
	groups := make(map[changefeedSyncGroupKey][]*entity.CDCChangefeed)
	svc := &cdcService{
		tiCDCChangefeedResourceRepo: w.tiCDCChangefeedResourceRepo,
		tiCDCChangefeedRepo:         nil,
	}
	for _, changefeed := range changefeeds {
		key := changefeedSyncGroupKey{
			clusterID: changefeed.ClusterID,
			arch:      changefeed.Arch,
		}
		if isNewArchChangefeed(changefeed) {
			key.shard = svc.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
		}
		groups[key] = append(groups[key], changefeed)
	}
	return groups
}

// synchronizes the state of CDC changefeeds with the actual state in the TiCDC clusters.
func (w *cdcWorker) SyncChangefeedsForOneCluster(
	ctx context.Context,
	changefeeds []*entity.CDCChangefeed,
) bizerr.BizError {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("sync changefeeds for one cluster", zap.String("clusterID", changefeeds[0].ClusterID))

	changefeedMap := make(map[string]*serverless.ChangefeedCommonInfo)
	if isNewArchChangefeed(changefeeds[0]) {
		svc := &cdcService{
			clusterProxyProviderFacade:  w.clusterProxyProviderFacade,
			mgmtSvc:                     w.mgmtSvc,
			tiCDCChangefeedResourceRepo: w.tiCDCChangefeedResourceRepo,
		}
		client, err := svc.getNewArchTiCDCClient(
			ctx,
			changefeeds[0].CloudProvider,
			changefeeds[0].Region,
			changefeeds[0].Kubernetes,
		)
		if err != nil {
			logger.Error("get newarch ticdc client failed", zap.Error(err))
			return err
		}
		clusterName := svc.loadNewArchShardForChangefeed(ctx, changefeeds[0]).ClusterName
		keyspaceName, keyspaceErr := svc.getNewArchKeyspaceName(ctx, changefeeds[0].ClusterID)
		if keyspaceErr != nil {
			logger.Error("get newarch keyspace name failed", zap.Error(keyspaceErr))
			return keyspaceErr
		}
		resp, reqErr := client.ListChangefeeds(ctx, clusterName, keyspaceName)
		if reqErr != nil {
			logger.Error("list newarch changefeeds failed", zap.Error(reqErr))
			return bizerr.WrapError(reqErr, "list newarch changefeeds failed")
		}
		logger.Info("list changefeeds from newarch TiCDC", zap.Any("cf", resp.Items))
		for i := range resp.Items {
			item, convErr := convertV2CommonInfoToServerless(&resp.Items[i])
			if convErr != nil {
				return bizerr.WrapError(convErr, "convert newarch changefeed state failed")
			}
			if item != nil && item.ID != nil {
				changefeedMap[*item.ID] = item
			}
		}
	} else {
		proxy, err := w.getServerlessClusterProxy(
			ctx,
			changefeeds[0].CloudProvider,
			changefeeds[0].Region,
			changefeeds[0].Kubernetes,
		)
		if err != nil {
			logger.Error("get serverless cluster proxy failed", zap.Error(err))
			return err
		}

		resp, err := proxy.ListServerlessChangefeeds(ctx, changefeeds[0].KeySpaceID, nil)
		if err != nil {
			logger.Error("list changefeeds failed", zap.Error(err))
			return err
		}

		logger.Info("list changefeeds from TiCDC", zap.Any("cf", resp.Items))
		for _, item := range resp.Items {
			changefeedMap[*item.ID] = item
		}
	}

	var hasErr bool
	for _, target := range changefeeds {
		SyncFromChangefeedCommon(ctx, target, changefeedMap[changefeedSyncKey(target)])
		if syncErr := w.changefeedRepo.SyncState(ctx, target); syncErr != nil {
			logger.Error(
				"sync changefeed state failed",
				zap.String("clusterID", target.ClusterID),
				zap.String("changefeedID", target.ID),
				zap.Error(syncErr),
			)
			hasErr = true
		}
	}
	if hasErr {
		return bizerr.New(bizerr.ErrCodeUnknown,
			"sync %s changefeed state failed, see the logs for details",
			changefeeds[0].ClusterID)
	}
	return nil
}

func changefeedSyncKey(changefeed *entity.CDCChangefeed) string {
	if changefeed == nil || changefeed.ChangefeedID == "" {
		if changefeed == nil {
			return ""
		}
		return changefeed.ID
	}
	return changefeed.ChangefeedID
}