package controller

import (
	"context"
	"time"

	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/worker"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/infra/manager/branch"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/service/billing_svc"
	"github.com/tidbcloud/serverless-service/internal/service/cluster_svc"
	"github.com/tidbcloud/serverless-service/internal/service/dm_svc"
	"github.com/tidbcloud/serverless-service/internal/service/export_svc"
	"github.com/tidbcloud/serverless-service/internal/service/import_svc"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	StatusOK    = "ok"
	StatusError = "error"
)

type GlobalWorker struct {
	workers []*worker.PeriodWorker
}

func (w *GlobalWorker) Start() {
	for _, worker := range w.workers {
		worker.Start()
	}
}

type ServicePlanEventJob struct {
	servicePlanEventSvc *cluster_svc.ServicePlanEventService
}

func NewServicePlanEventJob(
	servicePlanEventSvc *cluster_svc.ServicePlanEventService,
) *ServicePlanEventJob {
	return &ServicePlanEventJob{
		servicePlanEventSvc: servicePlanEventSvc,
	}
}

func (w *ServicePlanEventJob) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "ServicePlanEventJob")
	return w.servicePlanEventSvc.Execute(newCtx)
}

func NewGlobalWorkers() *GlobalWorker {
	gWorker := &GlobalWorker{}

	err := DiCall(func(
		// metering for cluster in tidb-pool, with RU + Storage Size bigger then 1GiB
		clusterTidbPoolSyncer *ClusterTidbPoolSyncer,
		branchMetricCollector *BranchMetricCollector,
		branchLimiter *BranchStorageLimiter,
		branchSyncWorker *BranchSyncWorker,
		syncer *ImportSyncer,
		exportSyncer *ExportSyncer,
		exportMeteringWorker *ExportMeteringWorker,
		instanceCapacityMeterWorker *InstanceCapacityMeterWorker,
		triggerIndexAdvisorWorker *TriggerIndexAdvisorWorker,
		downtimeWorker *DowntimeWorker,
		servicePlanEventJob *ServicePlanEventJob,
		dmTaskWorker *DMTaskWorker,
		dmPrecheckWorker *DMPrecheckWorker,
		dmEndpointWorker *DMEndpointWorker,
	) {
		branchWorkers := newBranchPeriodWorkers(branchSyncWorker)
		gWorker.workers = []*worker.PeriodWorker{
			newPeriodWorkerWithFixedDelay("clusterTidbPoolSyncer", time.Second, clusterTidbPoolSyncer.Run),
			newPeriodWorkerWithFixedDelay("branchMetricCollector", 30*time.Second, branchMetricCollector.Run),
			newPeriodWorkerWithFixedDelay("branchStorageLimiter", 5*time.Minute, branchLimiter.Run),
			branchWorkers[0],
			branchWorkers[1],
			newPeriodWorkerWithFixedDelay("importSyncer", 5*time.Second, syncer.Run),
			newPeriodWorkerWithFixedDelay("exportSyncWorker", 5*time.Second, exportSyncer.Run),
			newPeriodWorkerWithFixedDelay("exportMeteringWorker", 30*time.Second, exportMeteringWorker.Run),
			newPeriodWorkerWithFixedDelay(
				"instanceCapacityMeterWorker",
				5*time.Second,
				instanceCapacityMeterWorker.Run,
			),
			newPeriodWorkerWithFixedDelay("triggerIndexAdvisor", time.Hour, triggerIndexAdvisorWorker.Run),
			newPeriodWorkerWithFixedDelay("downtimeWorker", time.Minute*2, downtimeWorker.Run),
			newPeriodWorkerWithFixedDelay("ServicePlanEventJob", time.Hour*1, servicePlanEventJob.Run),
			newPeriodWorkerWithFixedDelay("dmTaskWorker", time.Second*5, dmTaskWorker.Run),
			newPeriodWorkerWithFixedDelay("dmPrecheckWorker", time.Second*10, dmPrecheckWorker.Run),
			newPeriodWorkerWithFixedDelay("dmEndpointWorker", time.Second*5, dmEndpointWorker.Run),
		}
	})
	if err != nil {
		panic(err)
	}

	return gWorker
}

type BRWorker struct {
	GlobalWorker
}

func (b *BRWorker) Start() {
	b.GlobalWorker.Start()
}

func NewBRWorkers() *BRWorker {
	brWorker := &BRWorker{}
	// run period worker
	err := DiCall(func(
		backupCleanWorker *BackupCleanWorker,
		restoreSyncWorker *RestoreSyncWorker,
		restoreWorker *RestoreWorker,
		backupScheduleCheckWorker *BackupScheduleCheckWorker,
		virtualizedBackupSlotWorker *VirtualizedBackupSlotWorker,
		virtualizedBackupCleanupWorker *VirtualizedBackupCleanupWorker,
		cdcWorker *CDCWorker,
		plcWorker *PrivateLinkConnectionWorker,
	) {
		brWorker.workers = []*worker.PeriodWorker{
			newPeriodWorkerWithFixedDelay("backupCleanWorker", time.Hour*12, backupCleanWorker.Run),
			newPeriodWorkerWithFixedDelay("restoreSyncWorker", time.Second*10, restoreSyncWorker.Run),
			newPeriodWorkerWithFixedDelay("restoreWorker", time.Second*10, restoreWorker.Run),
			newPeriodWorkerWithFixedDelay("backupScheduleCheckWorker", time.Hour*2, backupScheduleCheckWorker.Run),
			newPeriodWorkerWithFixedDelay("virtualizedBackupSlotWorker", time.Minute, virtualizedBackupSlotWorker.Run),
			newPeriodWorkerWithFixedDelay(
				"virtualizedBackupCleanupWorker",
				time.Hour*12,
				virtualizedBackupCleanupWorker.Run,
			),
			// cdc
			newPeriodWorkerWithFixedDelay("cdcSyncCreatingWorker", time.Second*10, cdcWorker.SyncCreating),
			// not execute cleaning worker for now
			// newPeriodWorkerWithFixedDelay("cdcCleanWorker", time.Hour*12, cdcWorker.Clean),
			newPeriodWorkerWithFixedDelay("cdcSyncChangfeedStateWorker", time.Minute*1, cdcWorker.SyncChangefeedState),
			// private link connection
			newPeriodWorkerWithFixedDelay(
				"privateLinkConnectionSyncWorker",
				time.Second*30,
				plcWorker.SyncPrivateLinkConnections,
			),
			newPeriodWorkerWithFixedDelay(
				"privateLinkConnectionDeleteWorker",
				time.Second*10,
				plcWorker.DeletePrivateLinkConnections,
			),
			newPeriodWorkerWithFixedDelay(
				"plcDomainRecordSyncWorker",
				time.Second*10,
				plcWorker.SyncPLCDomainRecords,
			),
		}
	})
	if err != nil {
		panic(err)
	}

	return brWorker
}

func newPeriodWorkerWithFixedDelay(workerName string, interval time.Duration, f worker.Handler) *worker.PeriodWorker {
	logger := log.With(zap.String("worker", workerName), zap.String("interval", interval.String()))
	return worker.NewPeriodWorker(logger, interval, withMetricsFixedDelay(workerName, f))
}

func withMetricsFixedDelay(workerName string, f worker.Handler) worker.Handler {
	return func(ctx context.Context) error {
		tBegin := time.Now()
		status := StatusOK
		err := f(ctx)
		if err != nil {
			status = StatusError
		}
		duration := time.Since(tBegin).Seconds()
		metric.WorkerRunDuration.WithLabelValues(workerName, status).Observe(duration)
		return err
	}
}

type ExportMeteringWorker struct {
	meteringSvc *billing_svc.MeteringService
}

func NewExportMeteringWorker(
	meteringSvc *billing_svc.MeteringService,
) *ExportMeteringWorker {
	return &ExportMeteringWorker{
		meteringSvc: meteringSvc,
	}
}

func (w *ExportMeteringWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "exportMeteringWorker")
	return w.meteringSvc.ExecuteExport(newCtx)
}

// InstanceCapacityMeterWorker meters instance capacity every minute for contracted orgs.
type InstanceCapacityMeterWorker struct {
	svc *billing_svc.InstanceCapacityMeterService
}

func NewInstanceCapacityMeterWorker(
	svc *billing_svc.InstanceCapacityMeterService,
) *InstanceCapacityMeterWorker {
	return &InstanceCapacityMeterWorker{svc: svc}
}

func (w *InstanceCapacityMeterWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "instanceCapacityMeterWorker")
	return w.svc.Execute(newCtx)
}

type DowntimeWorker struct {
	downtimeSvc *cluster_svc.DowntimeService
}

func NewDowntimeWorker(
	downtimeSvc *cluster_svc.DowntimeService,
) *DowntimeWorker {
	return &DowntimeWorker{
		downtimeSvc: downtimeSvc,
	}
}

func (w *DowntimeWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "downtimeWorker")
	return w.downtimeSvc.Execute(newCtx)
}

type ClusterTidbPoolSyncer struct {
	syncTidbPoolSvc *cluster_svc.SyncTidbPoolService
}

func NewClusterTidbPoolSyncer(
	syncTidbPoolSvc *cluster_svc.SyncTidbPoolService,
) *ClusterTidbPoolSyncer {
	return &ClusterTidbPoolSyncer{
		syncTidbPoolSvc: syncTidbPoolSvc,
	}
}

func (w *ClusterTidbPoolSyncer) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "clusterTidbPoolSyncer")
	return w.syncTidbPoolSvc.Execute(newCtx)
}

type ClusterTidbPoolCleaner struct {
	cleanupTidbPoolSvc *cluster_svc.CleanupTidbPoolService
}

func NewClusterTidbPoolCleaner(
	cleanupTidbPoolSvc *cluster_svc.CleanupTidbPoolService,
) *ClusterTidbPoolCleaner {
	return &ClusterTidbPoolCleaner{
		cleanupTidbPoolSvc: cleanupTidbPoolSvc,
	}
}

func (w *ClusterTidbPoolCleaner) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "clusterTidbPoolCleaner")
	return w.cleanupTidbPoolSvc.Execute(newCtx)
}

type BranchMetricCollector struct {
	branchSvc service.BranchService
}

func NewBranchMetricCollector(
	branchSvc service.BranchService,
) *BranchMetricCollector {
	return &BranchMetricCollector{
		branchSvc: branchSvc,
	}
}

func (w *BranchMetricCollector) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "branchMetricCollector")
	return w.branchSvc.CollectMetricsFromDB(newCtx)
}

type BranchStorageLimiter struct {
	branchSvc *service.BranchSizeLimiterService
}

func NewBranchStorageLimiter(
	branchSvc *service.BranchSizeLimiterService,
) *BranchStorageLimiter {
	return &BranchStorageLimiter{
		branchSvc: branchSvc,
	}
}

func (w *BranchStorageLimiter) Run(ctx context.Context) (err error) {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "branchStorageLimiter")
	log.Info(ctx, "BranchStorageLimiter start")
	return w.branchSvc.Execute(newCtx)
}

type ImportSyncer struct {
	syncer *import_svc.ImportSyncerService
}

func NewImportSyncer(
	syncer *import_svc.ImportSyncerService,
) *ImportSyncer {
	return &ImportSyncer{
		syncer: syncer,
	}
}

func (s *ImportSyncer) Run(ctx context.Context) (err error) {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "importSyncer")
	log.Info(ctx, "importSyncer start")
	return s.syncer.Run(newCtx)
}

type ExportSyncer struct {
	syncer *export_svc.ExportSyncerService
}

type branchReconciler interface {
	Run(ctx context.Context) error
	RunFast(ctx context.Context) error
}

type BranchSyncWorker struct {
	reconciler branchReconciler
}

func NewBranchSyncWorker(reconciler *branch.BranchReconciler) *BranchSyncWorker {
	return &BranchSyncWorker{reconciler: reconciler}
}

type branchSyncWorkerSpec struct {
	name     string
	interval time.Duration
	handler  worker.Handler
}

func branchSyncWorkerSpecs(w *BranchSyncWorker) []branchSyncWorkerSpec {
	return []branchSyncWorkerSpec{
		{name: "branchFastSyncWorker", interval: time.Second, handler: w.RunFast},
		{name: "branchSyncWorker", interval: 5 * time.Second, handler: w.Run},
	}
}

func newBranchPeriodWorkers(w *BranchSyncWorker) []*worker.PeriodWorker {
	specs := branchSyncWorkerSpecs(w)
	workers := make([]*worker.PeriodWorker, 0, len(specs))
	for _, spec := range specs {
		workers = append(workers, newPeriodWorkerWithFixedDelay(spec.name, spec.interval, spec.handler))
	}
	return workers
}

func (w *BranchSyncWorker) RunFast(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "branchFastSyncWorker")
	log.Info(newCtx, "branchFastSyncWorker start")
	return w.reconciler.RunFast(newCtx)
}

func (w *BranchSyncWorker) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "branchSyncWorker")
	log.Info(newCtx, "branchSyncWorker start")
	return w.reconciler.Run(newCtx)
}

func NewExportSyncer(
	syncer *export_svc.ExportSyncerService,
) *ExportSyncer {
	return &ExportSyncer{
		syncer: syncer,
	}
}

func (s *ExportSyncer) Run(ctx context.Context) (err error) {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "exportSyncWorker")
	log.Info(ctx, "exportSyncWorker start")
	return s.syncer.Run(newCtx)
}

type DMPrecheckWorker struct {
	dmPrecheckWorker *dm_svc.DMPrecheckWorker
}

func NewDMPrecheckWorker(
	dmPrecheckWorker *dm_svc.DMPrecheckWorker,
) *DMPrecheckWorker {
	return &DMPrecheckWorker{
		dmPrecheckWorker: dmPrecheckWorker,
	}
}

func (w *DMPrecheckWorker) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "dmPrecheckWorker")
	return w.dmPrecheckWorker.Run(newCtx)
}

type DMEndpointWorker struct {
	dmEndpointWorker *dm_svc.DMEndpointWorker
}

func NewDMEndpointWorker(
	dmEndpointWorker *dm_svc.DMEndpointWorker,
) *DMEndpointWorker {
	return &DMEndpointWorker{
		dmEndpointWorker: dmEndpointWorker,
	}
}

func (w *DMEndpointWorker) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "dmEndpointWorker")
	return w.dmEndpointWorker.Run(newCtx)
}

type DMTaskWorker struct {
	dmTaskWorker *dm_svc.DMTaskWorker
}

func NewDMWorker(
	dmTaskWorker *dm_svc.DMTaskWorker,
) *DMTaskWorker {
	return &DMTaskWorker{
		dmTaskWorker: dmTaskWorker,
	}
}

func (w *DMTaskWorker) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "dmWorker")
	return w.dmTaskWorker.Run(newCtx)
}

type BackupCleanWorker struct {
	backupWorker service.BackupWorker
}

type VirtualizedBackupSlotWorker struct {
	worker *service.VirtualizedBackupSlotWorker
}

type VirtualizedBackupCleanupWorker struct {
	worker *service.VirtualizedBackupCleanupWorker
}

func NewVirtualizedBackupSlotWorker(worker *service.VirtualizedBackupSlotWorker) *VirtualizedBackupSlotWorker {
	return &VirtualizedBackupSlotWorker{worker: worker}
}

func (w *VirtualizedBackupSlotWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	return w.worker.Sync(context.WithValue(ctx, config.CtxWorkerNameKey, "virtualizedBackupSlotWorker"))
}

func NewVirtualizedBackupCleanupWorker(worker *service.VirtualizedBackupCleanupWorker) *VirtualizedBackupCleanupWorker {
	return &VirtualizedBackupCleanupWorker{worker: worker}
}

func (w *VirtualizedBackupCleanupWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	return w.worker.Clean(context.WithValue(ctx, config.CtxWorkerNameKey, "virtualizedBackupCleanupWorker"))
}

func NewBackupCleanWorker(backupWorker service.BackupWorker) *BackupCleanWorker {
	return &BackupCleanWorker{
		backupWorker: backupWorker,
	}
}

func (b *BackupCleanWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "backupClean")
	return b.backupWorker.CleanBackup(newCtx)
}

type RestoreSyncWorker struct {
	backupWorker service.BackupWorker
}

// NewRestoreSyncWorker sync restore progress
func NewRestoreSyncWorker(backupWorker service.BackupWorker) *RestoreSyncWorker {
	return &RestoreSyncWorker{
		backupWorker: backupWorker,
	}
}

func (b *RestoreSyncWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "restoreSync")
	return b.backupWorker.SyncRestoreProgress(newCtx)
}

type RestoreWorker struct {
	backupWorker service.BackupWorker
}

// NewRestoreWorker execute restore
func NewRestoreWorker(backupWorker service.BackupWorker) *RestoreWorker {
	return &RestoreWorker{
		backupWorker: backupWorker,
	}
}

func (b *RestoreWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "restore")
	return b.backupWorker.ExecuteRestore(newCtx)
}

type BackupScheduleCheckWorker struct {
	backupWorker service.BackupWorker
}

func NewBackupScheduleCheckWorker(backupWorker service.BackupWorker) *BackupScheduleCheckWorker {
	return &BackupScheduleCheckWorker{
		backupWorker: backupWorker,
	}
}

func (b *BackupScheduleCheckWorker) Run(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "backupScheduleCheck")
	// check yesterday's backup schedule by default
	yesterday := time.Now().Add(time.Duration(-24) * time.Hour)
	date := yesterday.UTC().Format(time.DateOnly)
	// if valid BackupScheduleCheckDate is set, use it
	if config.Global().BizConfig.BackupScheduleCheckDate != "" {
		checkDateStr := config.Global().BizConfig.BackupScheduleCheckDate
		checkDate, err := time.Parse(time.DateOnly, checkDateStr)
		if err != nil {
			log.Error(newCtx, "parse check date failed", zap.String("checkDate", checkDateStr), zap.Error(err))
		} else {
			date = checkDate.UTC().Format(time.DateOnly)
		}
	}
	return b.backupWorker.CheckBackupSchedule(newCtx, date)
}

type TriggerIndexAdvisorWorker struct {
	clusterController console.ClusterServiceServer
	featureGateFacade facade.FeatureGate
	mgmtsvcFacade     facade.MgmtSvcFacade
}

func NewTriggerIndexAdvisorWorker(
	clusterController console.ClusterServiceServer,
	featureGateFacade facade.FeatureGate,
	mgmtsvcFacade facade.MgmtSvcFacade,
) *TriggerIndexAdvisorWorker {
	return &TriggerIndexAdvisorWorker{
		clusterController: clusterController,
		featureGateFacade: featureGateFacade,
		mgmtsvcFacade:     mgmtsvcFacade,
	}
}

func (w *TriggerIndexAdvisorWorker) Run(ctx context.Context) error {
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "triggerIndexAdvisor")
	tenantIds := w.featureGateFacade.ListFeatureEnabledTenants(constant.FeatureIndexAdvisor)
	logger := log.WithContextL(newCtx).With(zap.String("worker", "triggerIndexAdvisorWorker"))
	logger.Info("trigger index advisor one round", zap.Any("tenantIds", tenantIds))
	for _, tenantId := range tenantIds {
		logger := logger.With(zap.Uint64("tenantId", tenantId))
		clusters, err := w.mgmtsvcFacade.ListAllClustersInOrg(
			newCtx,
			tenantId,
			tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
		)
		if err != nil {
			logger.Error("list clusters failed", zap.Error(err))
			continue
		}
		for _, cluster := range clusters {
			func() {
				start := time.Now()
				clusterCtx, cancel := context.WithTimeout(newCtx, time.Minute*5)
				defer cancel()

				logger := log.WithContextL(clusterCtx).
					With(zap.String("worker", "triggerIndexAdvisorWorker")).
					With(zap.Uint64("tenantId", tenantId)).
					With(zap.Uint64("clusterId", cluster.ClusterId))
				logger.Info("start to run index advisor for cluster")
				resp, err := w.clusterController.AdviseIndexes(clusterCtx, &console.ClusterAdviseIndexesReq{
					OrgId:          tenantId,
					ProjectId:      cluster.ProjectId,
					ClusterId:      cluster.ClusterId,
					WithCloudAdmin: true,
					Payload: &console.ClusterAdviseIndexesReq_Payload{
						Queries: []string{},
					},
				})
				if err != nil {
					logger.Warn("failed to advise indexes", zap.Error(err))
					return
				}
				logger.Info("advise indexes success", zap.Any("resp", resp), zap.Duration("elapsed", time.Since(start)))
			}()
		}
	}
	return nil
}

type CDCWorker struct {
	cdc service.CDCWorker
}

func NewCDCWorker(cdc service.CDCWorker) *CDCWorker {
	return &CDCWorker{
		cdc: cdc,
	}
}

func (c *CDCWorker) Clean(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "clean cdc changefeed")
	return c.cdc.CleanChangefeeds(newCtx)
}

func (c *CDCWorker) SyncCreating(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "sync creating cdc changefeed")
	return c.cdc.SyncCreatingChangefeedsAndCDCCluster(newCtx)
}

func (c *CDCWorker) SyncChangefeedState(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "sync cdc changefeed state")
	return c.cdc.SyncChangefeeds(newCtx)
}

type PrivateLinkConnectionWorker struct {
	plc service.PrivateLinkConnectionWorker
}

func NewPrivateLinkConnectionWorker(plc service.PrivateLinkConnectionWorker) *PrivateLinkConnectionWorker {
	return &PrivateLinkConnectionWorker{
		plc: plc,
	}
}

func (w *PrivateLinkConnectionWorker) SyncPrivateLinkConnections(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "sync private link connection")
	return w.plc.SyncPrivateLinkConnections(newCtx)
}

func (w *PrivateLinkConnectionWorker) DeletePrivateLinkConnections(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "delete private link connection")
	return w.plc.DeletePrivateLinkConnections(newCtx)
}

func (w *PrivateLinkConnectionWorker) SyncPLCDomainRecords(ctx context.Context) error {
	if config.IsLocalStage() {
		return nil
	}
	newCtx := context.WithValue(ctx, config.CtxWorkerNameKey, "sync plc domain records")
	return w.plc.SyncPLCDomainRecords(newCtx)
}