package service

import (
	"context"
	"fmt"

	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
)

func (w *cdcWorker) syncNewArchChangefeedResources(ctx context.Context) error {
	// Gate on cdc.newArch.enabled. A pod that doesn't have the
	// newarch config block loaded (e.g. a control plane that hasn't
	// opted in yet, or — common during rollout — the canary running
	// alongside a non-newarch one) MUST NOT run the sync: without
	// the configured rcu ladder it falls back to
	// DefaultNewArchSpecifications, which has different StorageGi
	// values than whatever the http pod used at CR-create time, so
	// isTiCDCCRSpecMatched perpetually returns false and the worker
	// races the legitimate sync writers by re-asserting state=creating
	// + actual_specification=''.
	if !config.Global().CDCConfig.NewArch.Enabled {
		return nil
	}

	resources, err := w.tiCDCChangefeedResourceRepo.ListNeedingSync(ctx)
	if err != nil {
		return err
	}
	if len(resources) == 0 {
		return nil
	}

	svc := &cdcService{
		transProvider:               w.transProvider,
		changefeedRepo:              w.changefeedRepo,
		tiCDCChangefeedResourceRepo: w.tiCDCChangefeedResourceRepo,
		tiCDCClusterRepo:            w.tiCDCClusterRepo,
		clusterProxyProviderFacade:  w.clusterProxyProviderFacade,
	}
	shards := make(map[string][]*entity.TiCDCChangefeedResource)
	for _, resource := range resources {
		if resource == nil {
			continue
		}
		key := fmt.Sprintf("%s/%s/%s/%s/%s",
			resource.CloudProvider,
			resource.Region,
			resource.Kubernetes,
			resource.ShareScope,
			resource.ShareKey,
		)
		shards[key] = append(shards[key], resource)
	}

	var syncErr error
	for _, shardResources := range shards {
		if len(shardResources) == 0 {
			continue
		}
		if err := svc.syncNewArchChangefeedResourcesForShard(ctx, shardResources); err != nil {
			syncErr = err
		}
	}
	return syncErr
}