package impl

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pingcap/errors"
	"github.com/samber/lo"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	internalconfig "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/azblob"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/dataflow-service/pkg/s3"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/logid"
	"go.uber.org/zap"
)

var _ backup.CompactBackupModel = &CompactBackupModel{}

type CompactBackupModel struct {
	// db
	dbProvider        db.TransProvider
	compactBackupRepo backup_repo.CompactBackupRepo
	logBackupRepo     backup_repo.LogBackupRepo

	// cluster job for locking
	job cluster_job.Model

	// workflow
	workflowCli workflow.Workflow

	// grpc
	coreGrpcClient grpcli.CoreGrpcClient
	mgmtSvcCli     grpcli.MgmtSvcGrpcClient

	// storage (used for sync deletion helpers)
	bucketRepo repo.BucketRepo
	encryptor  encryptor_wrapper.Encryptor
}

func NewCompactBackupModel(
	dbProvider db.TransProvider,
	compactBackupRepo backup_repo.CompactBackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	job cluster_job.Model,
	workflowCli workflow.Workflow,
	coreGrpcClient grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	bucketRepo repo.BucketRepo,
	encryptor encryptor_wrapper.Encryptor,
) *CompactBackupModel {
	return &CompactBackupModel{
		dbProvider:        dbProvider,
		compactBackupRepo: compactBackupRepo,
		logBackupRepo:     logBackupRepo,
		job:               job,
		workflowCli:       workflowCli,
		coreGrpcClient:    coreGrpcClient,
		mgmtSvcCli:        mgmtSvcCli,
		bucketRepo:        bucketRepo,
		encryptor:         encryptor,
	}
}

func (c *CompactBackupModel) AsyncCreate(ctx context.Context,
	req *domain.AsyncCreateCompactBackupReq,
) (*domain.AsyncCreateCompactBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.AsyncCreate").With(zap.Uint64("cluster_id", req.ClusterID))

	if err := req.ValidateReq(logger); err != nil {
		return nil, err
	}

	resp := &domain.AsyncCreateCompactBackupResp{}

	// pre-check: validate log backup exists and is valid
	if err := c.preCheck(ctx, req); err != nil {
		return nil, err
	}

	if req.DryRun {
		logger.Info("dry run async create compact backup success")
		return nil, nil
	}

	txF := func(ctx context.Context) error {
		// 1: create compact backup entity
		compactBackupEntity, err := c.createDBEntity(ctx, logger, req)
		if err != nil {
			return err
		}
		resp.CompactBackupID = compactBackupEntity.ID.Uint64()

		// create db record success, let's leave compact backup id in logger ctx for tracing
		logger = logger.With(zap.Uint64("compact_backup_id", compactBackupEntity.ID.Uint64()))

		// 3: create compact backup dag
		if dagID, err := c.createCompactBackupDag(ctx, logger, req, compactBackupEntity); err != nil {
			return err
		} else {
			resp.WorkflowID = dagID
		}
		return nil
	}

	if err := c.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		logger.Error("create compact backup failed", zap.Error(err))
		// when meet db or other internal error, we should return an internal to user rather than raw error
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}

	return resp, nil
}

func (c *CompactBackupModel) AsyncDelete(ctx context.Context,
	req *domain.AsyncDeleteCompactBackupReq,
) (*domain.AsyncDeleteCompactBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.AsyncDelete").
		With(zap.Uint64("compact_backup_id", req.CompactBackupID))

	// Load compact backup to validate it exists
	compactBackup, err := c.compactBackupRepo.Load(ctx, req.CompactBackupID)
	if err != nil {
		logger.Error("failed to load compact backup", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, 0)
	}

	if !compactBackup.MarkDeleting() {
		logger.Info("compact backup already in delete state, skipping")
		return &domain.AsyncDeleteCompactBackupResp{}, nil
	}
	if saveErr := c.compactBackupRepo.Save(ctx, compactBackup); saveErr != nil {
		logger.Error("failed to set compact backup status to deleting", zap.Error(saveErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	// Create DAG variables
	vars := map[string]string{
		workflow.KeyCompactBackupID: strconv.FormatUint(req.CompactBackupID, 10),
	}

	// Create DAG tags for tracking
	tags := map[string]string{
		workflow.TagClusterID:       strconv.FormatUint(compactBackup.ClusterID.Uint64(), 10),
		workflow.TagCompactBackupID: strconv.FormatUint(req.CompactBackupID, 10),
	}

	// Start delete DAG
	dagInstance, runErr := c.workflowCli.RunDagWithTags(workflow.DagDeleteCompactBackup, vars, tags)
	if runErr != nil {
		logger.Error("failed to start delete compact backup dag", zap.Error(runErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	logger.Info("compact backup delete dag started", zap.String("workflow_id", dagInstance.ID))

	return &domain.AsyncDeleteCompactBackupResp{
		WorkflowID: dagInstance.ID,
	}, nil
}

func (c *CompactBackupModel) preCheck(ctx context.Context, req *domain.AsyncCreateCompactBackupReq) bizerr.BizError {
	logger := log.GetLogger(ctx).Named("preCheck")

	// Check if log backup exists and is valid
	logBackup, err := c.logBackupRepo.Load(ctx, req.LogBackupID)
	if err != nil {
		logger.Error("failed to load log backup", zap.Uint64("log_backup_id", req.LogBackupID), zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	// Refuse if log backup is not running at all
	if lo.Contains([]entity.LogBackupStatus{entity.LogBackupStatusScheduled, entity.LogBackupStatusPending}, logBackup.Status) {
		logger.Error("log backup is not in a valid state for compacting",
			zap.String("log_backup_status", string(logBackup.Status)))
		return bizerr_wrapper.NewInvalidInputError(
			"log backup is not in a valid state for compacting, current status: %s", logBackup.Status)
	}

	// Validate log backup belongs to the same tenant as the requested cluster.
	// Use mgmt service (CLUSTER_VIEW_BASIC) so this check remains available even when the
	// regional control-plane is unavailable (e.g. CRRF / disaster-recovery scenarios).
	cluster, clusterErr := c.mgmtSvcCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if clusterErr != nil {
		logger.Error("failed to get cluster info for tenant validation",
			zap.Uint64("cluster_id", req.ClusterID), zap.Error(clusterErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	// For CRPiTR (CRRF) scenario, the compaction is triggered by the restored cluster while
	// the compaction target bucket belongs to the source cluster, so we should validate tenant based on the log backup instead of the cluster.
	if logBackup.TenantID.Uint64() != cluster.GetOrganizationId() {
		logger.Error("log backup tenant mismatch",
			zap.Uint64("log_backup_tenant_id", logBackup.TenantID.Uint64()),
			zap.Uint64("cluster_organization_id", cluster.GetOrganizationId()))
		return bizerr_wrapper.NewInvalidInputError(
			"log backup does not belong to the same tenant as the specified cluster, cluster_id: %d", req.ClusterID)
	}

	if req.BucketID != 0 {
		if logBackup.BucketSettingSnapshot == nil {
			logger.Error("log backup has no bucket setting snapshot",
				zap.Uint64("log_backup_id", req.LogBackupID))
			return bizerr_wrapper.NewInvalidInputError(
				"bucket %d is not associated with log backup %d", req.BucketID, req.LogBackupID)
		}
		bucketFound := false
		for _, b := range logBackup.BucketSettingSnapshot.Buckets {
			if b.BucketId == req.BucketID {
				bucketFound = true
				break
			}
		}
		if !bucketFound {
			logger.Error("bucket is not associated with the log backup",
				zap.Uint64("bucket_id", req.BucketID),
				zap.Uint64("log_backup_id", req.LogBackupID))
			return bizerr_wrapper.NewInvalidInputError(
				"bucket %d is not associated with log backup %d", req.BucketID, req.LogBackupID)
		}
	}

	return nil
}

func (c *CompactBackupModel) createDBEntity(ctx context.Context, logger *zap.Logger,
	req *domain.AsyncCreateCompactBackupReq,
) (*entity.CompactBackup, error) {
	compactBackup := &entity.CompactBackup{
		ClusterID:   entity.ID(req.ClusterID),
		LogBackupID: entity.ID(req.LogBackupID),
		Status:      entity.CompactBackupStatusPending,
		BucketID:    entity.ID(req.BucketID),
		StartTS:     req.StartTS,
		EndTS:       req.EndTS,
	}

	if err := c.compactBackupRepo.Save(ctx, compactBackup); err != nil {
		logger.Error("failed to save compact backup entity", zap.Error(err))
		return nil, err
	}

	// Second save: persist the deterministic downstream CR name derived from the auto-increment ID.
	// This ensures DownstreamCompactID is always set before the DAG starts, eliminating nil-polling.
	downstreamID := compactBackup.GenerateDownstreamID()
	compactBackup.DownstreamCompactID = &downstreamID
	if err := c.compactBackupRepo.Save(ctx, compactBackup); err != nil {
		logger.Error("failed to persist downstream compact ID", zap.Error(err))
		return nil, err
	}

	logger.Info("compact backup entity created",
		zap.Uint64("compact_backup_id", compactBackup.ID.Uint64()),
		zap.String("downstream_compact_id", downstreamID),
		zap.String("status", string(compactBackup.Status)))

	return compactBackup, nil
}

func (c *CompactBackupModel) createCompactBackupDag(ctx context.Context, logger *zap.Logger,
	req *domain.AsyncCreateCompactBackupReq, compactBackup *entity.CompactBackup,
) (string, error) {
	// Create DAG variables
	vars := map[string]string{
		workflow.KeyCompactBackupID: strconv.FormatUint(compactBackup.ID.Uint64(), 10),
	}

	// Create DAG tags for tracking and monitoring
	tags := map[string]string{
		workflow.TagClusterID:       strconv.FormatUint(req.ClusterID, 10),
		workflow.TagCompactBackupID: strconv.FormatUint(compactBackup.ID.Uint64(), 10),
		workflow.TagLogBackupID:     strconv.FormatUint(req.LogBackupID, 10),
		workflow.TagUserID:          strconv.FormatUint(req.UserID, 10),
	}

	// Start create compact backup DAG
	res, err := c.workflowCli.RunDagWithTags(workflow.DagCreateCompactBackup, vars, tags)
	if err != nil {
		logger.Error("failed to start create compact backup dag", zap.Error(err))
		return "", err
	}

	logger.Info("compact backup dag started",
		zap.String("workflow_id", res.ID),
		zap.String("dag_name", workflow.DagCreateCompactBackup))

	return res.ID, nil
}

// resolveStorage returns the storage URI and bucket for the given compact backup.
// Migrated from PurgeCompactionStorageAction.
func (c *CompactBackupModel) resolveStorage(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
) (string, *entity.Bucket, error) {
	var storageURI string
	var bucket *entity.Bucket

	if compactBackup.BucketID == 0 {
		b, bErr := c.bucketRepo.Load(ctx, logBackup.BucketID.Uint64())
		if bErr != nil {
			return "", nil, fmt.Errorf("load bucket %d: %w", logBackup.BucketID.Uint64(), bErr)
		}
		storageURI = logBackup.Storage
		bucket = b
	} else {
		b, bErr := c.bucketRepo.Load(ctx, compactBackup.BucketID.Uint64())
		if bErr != nil {
			return "", nil, fmt.Errorf("load bucket %d: %w", compactBackup.BucketID.Uint64(), bErr)
		}
		uri, err := b.URIInBucket(logBackup.Storage)
		if err != nil {
			return "", nil, fmt.Errorf("resolve URI in bucket %d: %w", compactBackup.BucketID.Uint64(), err)
		}
		storageURI = uri
		bucket = b
	}

	if bucket.InfraProviderType == entity.InfraProviderTypeAKS ||
		bucket.InfraProviderType == entity.InfraProviderTypeAzure {
		azurl, err := azblob.ParseAzblobURL(storageURI)
		if err != nil {
			return "", nil, errors.Annotate(err, "parse azure blob url")
		}
		sasToken, err := bucket.GetSASToken(c.encryptor)
		if err != nil {
			return "", nil, errors.Annotate(err, "get sas token")
		}
		azurl.SetSASToken(sasToken)
		storageURI = azurl.ToInnerURL(true)
	}

	return storageURI, bucket, nil
}

// deleteStorageSubDir loads the compact backup and log backup, resolves storage, then
// deletes the subdirectory returned by getSubDir. Not-found errors are silently skipped.
func (c *CompactBackupModel) deleteStorageSubDir(
	ctx context.Context,
	logger *zap.Logger,
	compactBackupID uint64,
	getSubDir func(*entity.CompactBackup) string,
) bizerr.BizError {
	compactBackup, bErr := c.compactBackupRepo.Load(ctx, compactBackupID)
	if bErr != nil {
		logger.Error("load compact backup failed",
			zap.Uint64("compact_backup_id", compactBackupID), zap.Error(bErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, 0)
	}

	logBackup, bErr := c.logBackupRepo.Load(ctx, compactBackup.LogBackupID.Uint64())
	if bErr != nil {
		logger.Error("load log backup failed",
			zap.Uint64("log_backup_id", compactBackup.LogBackupID.Uint64()), zap.Error(bErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	storageURI, bucket, err := c.resolveStorage(ctx, compactBackup, logBackup)
	if err != nil {
		logger.Error("resolve storage failed", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	opt := &blob.StorageOption{
		ProjectID:     logBackup.ProjectID.Uint64(),
		ClusterRegion: bucket.Regions[0],
	}
	if bucket.CloudAccountID != nil {
		opt.AWSRoleARN = domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID)
	}

	// Use a background context to avoid workflow cancellation interrupting the delete.
	bgCtx := context.Background()
	bgCtx = context.WithValue(bgCtx, logid.CtxKey, ctx.Value(logid.CtxKey))
	deleteOpt := &blob.DeleteOption{Concurrency: internalconfig.Global().BackupConfig.CleanupDeleteConcurrency}

	subDir := getSubDir(compactBackup)
	logger.Info("begin to delete storage subdir", zap.String("subdir", subDir))
	subDirStorageURI, joinErr := utils.JoinStorageURIWithSubDir(storageURI, subDir)
	if joinErr != nil {
		logger.Error("join storage uri with subdir failed", zap.String("subdir", subDir), zap.Error(joinErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}
	s, blobErr := blob.NewStorage(bgCtx, subDirStorageURI, opt)
	if blobErr != nil {
		if blob.IsNotFoundError(blobErr) {
			logger.Info("storage subdir not found, skipping", zap.String("subdir", subDir))
			return nil
		}
		logger.Error("init storage failed", zap.String("subdir", subDir), zap.Error(blobErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}
	deleteErr := s.DeleteSubDir(bgCtx, "", deleteOpt)
	if deleteErr != nil && !s3.IsNotFoundError(deleteErr) && !blob.IsNotFoundError(deleteErr) {
		logger.Error("delete storage subdir failed", zap.String("subdir", subDir), zap.Error(deleteErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}
	logger.Info("deleted storage subdir", zap.String("subdir", subDir))
	return nil
}

func (c *CompactBackupModel) SyncDeleteMigrations(ctx context.Context, compactBackupID uint64) bizerr.BizError {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.SyncDeleteMigrations").
		With(zap.Uint64("compact_backup_id", compactBackupID))
	return c.deleteStorageSubDir(ctx, logger, compactBackupID, func(_ *entity.CompactBackup) string {
		return "v1/migrations/"
	})
}

func (c *CompactBackupModel) DeleteCompactionsSubDir(ctx context.Context, compactBackupID uint64) bizerr.BizError {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.DeleteCompactionsSubDir").
		With(zap.Uint64("compact_backup_id", compactBackupID))
	return c.deleteStorageSubDir(ctx, logger, compactBackupID, func(cb *entity.CompactBackup) string {
		return fmt.Sprintf("v1/compactions/%s/", cb.GenerateDownstreamID())
	})
}