package service

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

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"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/grpc/grpcli"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	"go.uber.org/zap"
)

type cdcNewArchWorkflowHook struct {
	clusterRepo                 repo.ClusterRepo
	changefeedRepo              repo.ChangefeedRepo
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo
	tiCDCClusterRepo            repo.TiCDCClusterRepo
	tiCDCChangefeedRepo         repo.KernelTiCDCChangefeedRepo
	cdcClusterRepo              repo.CDCClusterRepo
	mgmtSvc                     facade.MgmtSvcFacade
	clusterProxyProviderFacade  facade.ClusterProxyProviderFacade
	encryptor                   facade.EncryptorFacade
	privateLinkConnectionSvc    PrivateLinkConnectionService
	configCenter                grpcli.ConfigCenterClient
}

func NewCDCNewArchWorkflowHook(
	clusterRepo repo.ClusterRepo,
	changefeedRepo repo.ChangefeedRepo,
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo,
	tiCDCClusterRepo repo.TiCDCClusterRepo,
	tiCDCChangefeedRepo repo.KernelTiCDCChangefeedRepo,
	cdcClusterRepo repo.CDCClusterRepo,
	mgmtSvc facade.MgmtSvcFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	encryptor facade.EncryptorFacade,
	privateLinkConnectionSvc PrivateLinkConnectionService,
	configCenter grpcli.ConfigCenterClient,
) appworkflow.CDCNewArchHook {
	return &cdcNewArchWorkflowHook{
		clusterRepo:                 clusterRepo,
		changefeedRepo:              changefeedRepo,
		tiCDCChangefeedResourceRepo: tiCDCChangefeedResourceRepo,
		tiCDCClusterRepo:            tiCDCClusterRepo,
		tiCDCChangefeedRepo:         tiCDCChangefeedRepo,
		cdcClusterRepo:              cdcClusterRepo,
		mgmtSvc:                     mgmtSvc,
		clusterProxyProviderFacade:  clusterProxyProviderFacade,
		encryptor:                   encryptor,
		privateLinkConnectionSvc:    privateLinkConnectionSvc,
		configCenter:                configCenter,
	}
}

func (h *cdcNewArchWorkflowHook) ScheduleChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.loadActiveNewArchChangefeed(ctx, clusterID, changefeedID)
	if bizErr != nil || changefeed == nil {
		return bizErr
	}

	svc := h.newCDCService()
	shard := resolveNewArchShard(changefeed)
	if _, _, bizErr = svc.persistNewArchTiCDCChangefeedCreateIntent(ctx, changefeed, &shard); bizErr != nil {
		if markErr := h.markCreateFailed(ctx, changefeed, bizErr); markErr != nil {
			return markErr
		}
		return bizErr
	}
	if _, bizErr = svc.ensureNewArchTiCDCCluster(ctx, changefeed); bizErr != nil {
		if markErr := h.markCreateFailed(ctx, changefeed, bizErr); markErr != nil {
			return markErr
		}
		return bizErr
	}
	return nil
}

func (h *cdcNewArchWorkflowHook) ScaleChangefeedResource(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		return bizErr
	}
	if !isNewArchChangefeed(changefeed) {
		return nil
	}
	svc := h.newCDCService()
	shard, bizErr := svc.ensureNewArchChangefeedResource(
		ctx,
		changefeed,
		formatNewArchSpecification(changefeed.RCU),
	)
	if bizErr != nil {
		return bizErr
	}
	return svc.reconcileNewArchTiCDCClusterForShard(ctx, shard)
}

func (h *cdcNewArchWorkflowHook) IsChangefeedResourceReady(
	ctx context.Context,
	clusterID, changefeedID string,
) (bool, error) {
	svc := h.newCDCService()
	return svc.isNewArchChangefeedResourceReady(ctx, clusterID, changefeedID)
}

func (h *cdcNewArchWorkflowHook) CreateChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.loadActiveNewArchChangefeed(ctx, clusterID, changefeedID)
	if bizErr != nil || changefeed == nil {
		return bizErr
	}

	svc := h.newCDCService()
	ready, bizErr := svc.isNewArchChangefeedResourceReady(ctx, changefeed.ClusterID, changefeed.ID)
	if bizErr != nil {
		return bizErr
	}
	if !ready {
		return errors.New("newarch changefeed resource is not ready yet")
	}

	client, bizErr := svc.getNewArchTiCDCClient(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if bizErr != nil {
		return bizErr
	}
	shard := resolveNewArchShard(changefeed)
	record, effectiveShard, bizErr := svc.persistNewArchTiCDCChangefeedCreateIntent(ctx, changefeed, &shard)
	if bizErr != nil {
		return bizErr
	}
	if record.KernelStatus == constant.TiCDCChangefeedKernelStatusActive {
		if changefeed.ChangefeedID == "" {
			changefeed.ChangefeedID = record.KernelChangefeedID
			_, bizErr := h.changefeedRepo.Save(ctx, changefeed)
			return bizErr
		}
		return nil
	}

	req, bizErr := changefeed.ToChangefeed(ctx, h.encryptor)
	if bizErr != nil {
		if markErr := h.markCreateFailed(ctx, changefeed, bizErr); markErr != nil {
			return markErr
		}
		return bizErr
	}
	reqV2, bizErr := assembleNewArchChangefeedConfig(ctx, h.configCenter, changefeed, req, nil)
	if bizErr != nil {
		if markErr := h.markCreateFailed(ctx, changefeed, bizErr); markErr != nil {
			return markErr
		}
		return bizErr
	}
	keyspaceName, bizErr := svc.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
	if bizErr != nil {
		return h.markCreateFailed(ctx, changefeed, bizErr)
	}
	kernelChangefeedID, reqErr := h.createOrVerifyNewArchKernelChangefeed(
		ctx,
		client,
		effectiveShard.ClusterName,
		keyspaceName,
		changefeed.ID,
		reqV2,
	)
	if reqErr != nil {
		return reqErr
	}

	changefeed.ChangefeedID = changefeed.ID
	if kernelChangefeedID != "" {
		changefeed.ChangefeedID = kernelChangefeedID
	}
	if _, bizErr := h.changefeedRepo.Save(ctx, changefeed); bizErr != nil {
		return bizErr
	}
	if bizErr := svc.ensureNewArchTiCDCChangefeedRecord(
		ctx,
		changefeed,
		effectiveShard,
		constant.TiCDCChangefeedKernelStatusActive,
	); bizErr != nil {
		return bizErr
	}
	return nil
}

func (h *cdcNewArchWorkflowHook) createOrVerifyNewArchKernelChangefeed(
	ctx context.Context,
	client *newArchTiCDCClient,
	clusterName, keyspaceName, kernelChangefeedID string,
	req *clusterproxy.ChangefeedConfig,
) (string, error) {
	resp, reqErr := client.CreateChangefeed(ctx, clusterName, keyspaceName, req)
	if reqErr == nil {
		if resp != nil && resp.ID != "" {
			return resp.ID, nil
		}
		return kernelChangefeedID, nil
	}
	if isNewArchChangefeedAlreadyExistsErr(reqErr) {
		return kernelChangefeedID, nil
	}
	exists, probeErr := h.newArchKernelChangefeedExists(ctx, client, clusterName, keyspaceName, kernelChangefeedID)
	if probeErr == nil && exists {
		return kernelChangefeedID, nil
	}
	return "", reqErr
}

func (h *cdcNewArchWorkflowHook) newArchKernelChangefeedExists(
	ctx context.Context,
	client *newArchTiCDCClient,
	clusterName, keyspaceName, kernelChangefeedID string,
) (bool, error) {
	resp, err := client.ListChangefeeds(ctx, clusterName, keyspaceName)
	if err != nil {
		return false, err
	}
	if resp == nil {
		return false, nil
	}
	for i := range resp.Items {
		if resp.Items[i].ID == kernelChangefeedID {
			return true, nil
		}
	}
	return false, nil
}

func (h *cdcNewArchWorkflowHook) DeleteChangefeedResource(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		return bizErr
	}
	if !isNewArchChangefeed(changefeed) {
		return nil
	}
	svc := h.newCDCService()
	shard, bizErr := svc.ensureNewArchChangefeedResource(ctx, changefeed, "")
	if bizErr != nil {
		return bizErr
	}
	return svc.reconcileNewArchTiCDCClusterForShard(ctx, shard)
}

func (h *cdcNewArchWorkflowHook) DeleteChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return bizErr
	}
	if !isNewArchChangefeed(changefeed) {
		return nil
	}

	svc := h.newCDCService()
	if changefeed.ChangefeedID != "" {
		client, bizErr := svc.getNewArchTiCDCClient(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
		if bizErr != nil {
			return bizErr
		}
		shard := svc.loadNewArchShardForChangefeed(ctx, changefeed)
		keyspaceName, keyspaceErr := svc.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
		if keyspaceErr != nil {
			return keyspaceErr
		}
		if reqErr := client.DeleteChangefeed(ctx, shard.ClusterName, keyspaceName, changefeed.ChangefeedID); reqErr != nil {
			// If the kernel changefeed or the whole shared TiCDC service
			// is already gone, treat delete as idempotent success and
			// mark the local kernel record deleted below.
			if !isNotFoundErr(reqErr) {
				_, getErr := client.GetChangefeed(ctx, shard.ClusterName, keyspaceName, changefeed.ChangefeedID)
				if !isNotFoundErr(getErr) {
					return reqErr
				}
			}
		}
	}

	shard := svc.loadNewArchShardForChangefeed(ctx, changefeed)
	if bizErr := svc.ensureNewArchTiCDCChangefeedRecord(
		ctx,
		changefeed,
		&shard,
		constant.TiCDCChangefeedKernelStatusDeleted,
	); bizErr != nil {
		return bizErr
	}
	return nil
}

func (h *cdcNewArchWorkflowHook) IsChangefeedResourceReleased(
	ctx context.Context,
	clusterID, changefeedID string,
) (bool, error) {
	svc := h.newCDCService()
	return svc.isNewArchChangefeedResourceReleased(ctx, clusterID, changefeedID)
}

func (h *cdcNewArchWorkflowHook) DeregisterChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return bizErr
	}
	if !isNewArchChangefeed(changefeed) {
		return nil
	}
	svc := h.newCDCService()
	return svc.tryToDeregisterPrivateLinkConnection(ctx, changefeed)
}

func (h *cdcNewArchWorkflowHook) FinalizeCreateChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.loadActiveNewArchChangefeed(ctx, clusterID, changefeedID)
	if bizErr != nil || changefeed == nil {
		return bizErr
	}
	if changefeed.ChangefeedID == "" {
		changefeed.ChangefeedID = changefeed.ID
	}
	changefeed.State = constant.ChangfeedStateRunning
	return h.changefeedRepo.ChangeStateFromCreating(ctx, changefeed)
}

func (h *cdcNewArchWorkflowHook) FinalizeScaleChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil
		}
		return bizErr
	}
	if !isNewArchChangefeed(changefeed) || changefeed.State != constant.ChangefeedStateScaling {
		return nil
	}
	changefeed.State = constant.ChangfeedStateRunning
	changefeed.Error = ""
	_, bizErr = h.changefeedRepo.Save(ctx, changefeed)
	return bizErr
}

func (h *cdcNewArchWorkflowHook) FinalizeDeleteChangefeed(ctx context.Context, clusterID, changefeedID string) error {
	_, bizErr := h.changefeedRepo.Delete(ctx, clusterID, changefeedID)
	if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return bizErr
	}
	return nil
}

func (h *cdcNewArchWorkflowHook) loadActiveNewArchChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
) (*entity.CDCChangefeed, bizerr.BizError) {
	changefeed, bizErr := h.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, nil
		}
		return nil, bizErr
	}
	if !isNewArchChangefeed(changefeed) || changefeed.State != constant.ChangefeedStateCreating {
		return nil, nil
	}
	return changefeed, nil
}

func (h *cdcNewArchWorkflowHook) markCreateFailed(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	cause error,
) error {
	if changefeed == nil {
		return cause
	}
	changefeed.State = constant.ChangefeedStateCreateFailed
	changefeed.Error = fmt.Sprintf("failed to create changefeed: %v", cause)
	if err := h.changefeedRepo.ChangeStateFromCreating(ctx, changefeed); err != nil {
		return err
	}
	svc := h.newCDCService()
	shard := svc.loadNewArchShardForChangefeed(ctx, changefeed)
	_ = svc.reconcileNewArchTiCDCClusterForShard(ctx, &shard)
	return cause
}

func (h *cdcNewArchWorkflowHook) newCDCService() *cdcService {
	return &cdcService{
		changefeedRepo:              h.changefeedRepo,
		tiCDCChangefeedResourceRepo: h.tiCDCChangefeedResourceRepo,
		tiCDCClusterRepo:            h.tiCDCClusterRepo,
		tiCDCChangefeedRepo:         h.tiCDCChangefeedRepo,
		mgmtSvc:                     h.mgmtSvc,
		clusterProxyProviderFacade:  h.clusterProxyProviderFacade,
		encryptor:                   h.encryptor,
		privateLinkConnectionSvc:    h.privateLinkConnectionSvc,
		configCenter:                h.configCenter,
	}
}

func (h *cdcNewArchWorkflowHook) RegisterCDCClusterJob(ctx context.Context, clusterID, changefeedID string) error {
	uintClusterID, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return bizerr.WrapError(err, "parse cluster id %s", clusterID)
	}
	if h.mgmtSvc == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "mgmt service is not configured")
	}

	registerErr := h.mgmtSvc.RegisterCDCClusterJob(ctx, uintClusterID, changefeedID)
	if registerErr == nil {
		return nil
	}
	// The registration is forbidden only when an existing custom status mutexes against
	// CDC. During migration this is typically the legacy cluster-level CDC lock written by
	// the old-arch SetClusterCDCStatus (its mutex still self-lists CDC). Heal it in place
	// and retry once. Any other forbidden cause (real import/restore) is left untouched by
	// ReplaceLegacyCDCClusterLock and surfaced as-is below.
	if registerErr.ErrCode() != bizerr.ErrCodePermission {
		return registerErr
	}

	replaced, oldStatusID, newStatusID, healErr := h.mgmtSvc.ReplaceLegacyCDCClusterLock(ctx, uintClusterID)
	if healErr != nil {
		return healErr
	}
	if !replaced {
		// Not blocked by the legacy CDC lock; surface the original forbidden error.
		return registerErr
	}

	log.WithContextL(ctx).Info("replaced legacy cluster-level CDC lock to unblock newarch registration",
		zap.String("cluster_id", clusterID),
		zap.String("changefeed_id", changefeedID),
		zap.Uint64("old_status_id", oldStatusID),
		zap.Uint64("new_status_id", newStatusID),
	)

	// The legacy lock id is tracked by the old-arch cdc_cluster record so its teardown can
	// delete it later. Repoint it at the replacement, otherwise the replacement would leak
	// and keep blocking cluster deletion forever.
	if h.cdcClusterRepo != nil {
		if cdcCluster, getErr := h.cdcClusterRepo.GetById(ctx, clusterID); getErr == nil &&
			cdcCluster != nil && cdcCluster.ClusterStatusId == oldStatusID {
			if setErr := h.cdcClusterRepo.SetClusterStatusId(ctx, clusterID, newStatusID); setErr != nil {
				log.WithContextL(ctx).Warn(
					"failed to repoint old-arch cdc_cluster status id after healing legacy CDC lock",
					zap.String("cluster_id", clusterID),
					zap.Uint64("new_status_id", newStatusID),
					zap.Error(setErr),
				)
			}
		}
	}

	return h.mgmtSvc.RegisterCDCClusterJob(ctx, uintClusterID, changefeedID)
}

func (h *cdcNewArchWorkflowHook) DeregisterCDCClusterJob(ctx context.Context, clusterID, changefeedID string) error {
	uintClusterID, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return bizerr.WrapError(err, "parse cluster id %s", clusterID)
	}
	if h.mgmtSvc == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "mgmt service is not configured")
	}
	return h.mgmtSvc.DeregisterCDCClusterJob(ctx, uintClusterID, changefeedID)
}