package compact

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

	"github.com/shiningrush/fastflow/pkg/entity/run"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

var errCompactBackupNotFinished error = errors.New("compact backup is not finished")

const (
	defaultSyncTimeout  = 5 * time.Hour
	statusUpdateTimeout = 30 * time.Second
)

type SyncCompactBackupStatusAction struct {
	transProvider     db.TransProvider
	compactBackupRepo backup_repo.CompactBackupRepo
	compactFacade     infraapi.CompactFacade
	mgmtGrpcClient    grpcli.MgmtSvcGrpcClient
	errorWaiter       action.Retryer
}

func NewSyncCompactBackupStatusAction(
	transProvider db.TransProvider,
	compactBackupRepo backup_repo.CompactBackupRepo,
	compactFacade infraapi.CompactFacade,
	mgmtGrpcClient grpcli.MgmtSvcGrpcClient,
) *SyncCompactBackupStatusAction {
	return &SyncCompactBackupStatusAction{
		transProvider:     transProvider,
		compactBackupRepo: compactBackupRepo,
		compactFacade:     compactFacade,
		mgmtGrpcClient:    mgmtGrpcClient,
		errorWaiter:       action.NewRetryer([]error{errCompactBackupNotFinished}),
	}
}

func (a *SyncCompactBackupStatusAction) Name() string {
	return workflow.ActionSyncCompactBackupStatus
}

// isNotFoundError checks if the error indicates that the compact backup CR was not found
func isNotFoundError(err error) bool {
	if bizErr, ok := err.(bizerr.BizError); ok {
		return bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound
	}
	return false
}

func (a *SyncCompactBackupStatusAction) Run(ctx run.ExecuteContext, params interface{}) error {
	return nil
}

// checkCompactBackupStatus checks if backup is already finished
func (a *SyncCompactBackupStatusAction) checkCompactBackupStatus(
	compactBackup *entity.CompactBackup,
) (finished bool, shouldContinue bool) {
	// If already finished, return immediately
	if compactBackup.Status.IsFinishedStatus() {
		return true, false
	}
	// Continue with sync process
	return false, true
}

// queryDownstreamStatus queries the downstream compact backup status
func (a *SyncCompactBackupStatusAction) queryDownstreamStatus(
	ctx context.Context, compactBackup *entity.CompactBackup, logger *zap.Logger,
) (*infraapi.CompactBackupResponse, error) {
	// Get cluster information for region
	cluster, err := a.mgmtGrpcClient.GetAnyDedicatedClusterBasicByID(ctx, compactBackup.ClusterID.Uint64())
	if err != nil {
		logger.Error("failed to get cluster info",
			zap.Uint64("cluster_id", compactBackup.ClusterID.Uint64()), zap.Error(err))
		return nil, err
	}
	region := cluster.GetRegion()
	cloudProvider := cluster.GetCloudProvider()

	// Query compact backup status from regional API
	resp, facadeErr := a.compactFacade.GetCompactBackup(ctx, cloudProvider, region,
		cluster.GetOrganizationId(), cluster.GetProjectId(), compactBackup.ClusterID.Uint64(),
		*compactBackup.DownstreamCompactID)

	return resp, facadeErr
}

// handleDownstreamError processes downstream query errors
func (a *SyncCompactBackupStatusAction) handleDownstreamError(
	compactBackup *entity.CompactBackup, facadeErr error, logger *zap.Logger,
) (finished bool) {
	logger.Error("failed to query compact backup status",
		zap.String("downstream_id", *compactBackup.DownstreamCompactID),
		zap.Error(facadeErr))

	// Check if it's a "not found" error - this indicates the downstream CR was deleted
	if isNotFoundError(facadeErr) {
		logger.Warn("downstream compact backup CR not found, marking as failed",
			zap.String("downstream_id", *compactBackup.DownstreamCompactID))
		compactBackup.Status = entity.CompactBackupStatusFailed
		compactBackup.ErrMsg = fmt.Sprintf("downstream compact backup not found: %v", facadeErr)
		now := time.Now()
		compactBackup.FinishedAt = &now
		return true
	}

	// For other errors (network issues, temporary failures), continue polling
	logger.Info("temporary error querying compact backup, will retry",
		zap.String("downstream_id", *compactBackup.DownstreamCompactID),
		zap.Error(facadeErr))
	return false
}

// updateCompactBackupStatus updates the compact backup status based on downstream response
func (a *SyncCompactBackupStatusAction) updateCompactBackupStatus(
	compactBackup *entity.CompactBackup, resp *infraapi.CompactBackupResponse,
	compactBackupID uint64, logger *zap.Logger,
) (finished bool) {
	switch resp.Status {
	case infraapi.CompactBackupStatusComplete:
		compactBackup.Status = entity.CompactBackupStatusSuccess
		now := time.Now()
		compactBackup.FinishedAt = &now
		finished = true
		logger.Info("compact backup completed successfully", zap.Uint64("id", compactBackupID))

	case infraapi.CompactBackupStatusFailed:
		compactBackup.Status = entity.CompactBackupStatusFailed
		compactBackup.ErrMsg = resp.ErrorMessage
		now := time.Now()
		compactBackup.FinishedAt = &now
		finished = true
		logger.Info("compact backup failed", zap.Uint64("id", compactBackupID),
			zap.String("error", resp.ErrorMessage))

	case infraapi.CompactBackupStatusRunning, infraapi.CompactBackupStatusPending:
		// Still running, keep current status
		finished = false

	case infraapi.CompactBackupStatusRetry:
		// Task is retrying, keep current status
		finished = false

	default:
		logger.Warn("unknown compact backup status",
			zap.Uint64("id", compactBackupID),
			zap.String("status", string(resp.Status)))
		finished = false
	}

	return finished
}

// processCompactBackupSync processes the compact backup sync within a transaction
func (a *SyncCompactBackupStatusAction) processCompactBackupSync(
	ctx context.Context, compactBackupID uint64, logger *zap.Logger,
) (finished bool, err error) {
	// Load current compact backup
	compactBackup, err := a.compactBackupRepo.Load(ctx, compactBackupID)
	if err != nil {
		return false, err
	}

	// If marked for deletion, don't overwrite the status — the delete DAG owns it now.
	if compactBackup.DeletingOrDeleted() {
		logger.Info("compact backup is being deleted or already deleted, skipping sync",
			zap.String("status", string(compactBackup.Status)), zap.Uint64("id", compactBackupID))
		return true, nil
	}

	// Check initial status
	finished, shouldContinue := a.checkCompactBackupStatus(compactBackup)
	if !shouldContinue {
		return finished, nil
	}

	// Query downstream status
	resp, facadeErr := a.queryDownstreamStatus(ctx, compactBackup, logger)
	if facadeErr != nil {
		finished = a.handleDownstreamError(compactBackup, facadeErr, logger)
		// Only save if it's an error that updated the status
		if finished {
			if err := a.compactBackupRepo.Save(ctx, compactBackup); err != nil {
				return false, err
			}
		}
		return finished, nil
	}

	// Update status based on response
	finished = a.updateCompactBackupStatus(compactBackup, resp, compactBackupID, logger)

	// Save the updated compact backup
	if err := a.compactBackupRepo.Save(ctx, compactBackup); err != nil {
		return false, err
	}

	return finished, nil
}

func (a *SyncCompactBackupStatusAction) syncStatusOnce(
	ctx context.Context, compactBackupID uint64, logger *zap.Logger,
) error {
	var finished bool

	err := a.transProvider.Transaction(ctx, func(ctx context.Context) error {
		var err error
		finished, err = a.processCompactBackupSync(ctx, compactBackupID, logger)
		return err
	})
	if err != nil {
		logger.Error("failed to check compact backup status", zap.Uint64("id", compactBackupID), zap.Error(err))
		return errCompactBackupNotFinished
	}

	if !finished {
		logger.Info("compact backup is still running, wait for next retry", zap.Uint64("id", compactBackupID))
		return errCompactBackupNotFinished
	}

	logger.Info("compact backup finished", zap.Uint64("id", compactBackupID))
	return nil
}

// getCompactBackupID extracts and parses the compact backup ID from share data
func (a *SyncCompactBackupStatusAction) getCompactBackupID(ctx run.ExecuteContext, logger *zap.Logger) (uint64, error) {
	compactBackupIDStr, ok := ctx.ShareData().Get(workflow.KeyCompactBackupID)
	if !ok {
		logger.Error("compact backup id not found in share data")
		return 0, fmt.Errorf("compact backup id not found in share data")
	}

	compactBackupID, err := strconv.ParseUint(compactBackupIDStr, 10, 64)
	if err != nil {
		logger.Error("invalid compact backup id", zap.String("id", compactBackupIDStr), zap.Error(err))
		return 0, fmt.Errorf("invalid compact backup id: %s", compactBackupIDStr)
	}

	return compactBackupID, nil
}

// getSyncTimeout retrieves the sync timeout configuration for the cluster
func (a *SyncCompactBackupStatusAction) getSyncTimeout(
	ctx context.Context, clusterID uint64, logger *zap.Logger,
) time.Duration {
	cluster, err := a.mgmtGrpcClient.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		logger.Error("failed to get cluster info for timeout config",
			zap.Uint64("cluster_id", clusterID), zap.Error(err))
		return defaultSyncTimeout
	}

	globalConfig := config.Global()
	syncTimeout, parseErr := globalConfig.CompactConfig.GetCompactSyncTimeoutDurationWithOverride(
		cluster.GetClusterId(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)
	if parseErr != nil {
		logger.Warn("failed to parse compact sync timeout, using default 25 minutes", zap.Error(parseErr))
		return defaultSyncTimeout
	}

	return syncTimeout
}

// handleSyncError processes the error from sync operation
func (a *SyncCompactBackupStatusAction) handleSyncError(
	ctx context.Context, retryErr error, compactBackupID uint64,
	syncTimeout time.Duration, logger *zap.Logger,
) error {
	if errors.Is(retryErr, context.DeadlineExceeded) {
		logger.Warn("sync timed out, updating to failed", zap.Uint64("id", compactBackupID))
		errMsg := fmt.Sprintf("sync status timed out after %v", syncTimeout)
		if updateErr := a.updateCompactBackupStatusToFailed(ctx, compactBackupID, errMsg); updateErr != nil {
			logger.Error("failed to update status on timeout", zap.Error(updateErr))
			return fmt.Errorf("timeout occurred but failed to update status: %w", updateErr)
		}
		return nil
	}

	if retryErr != errCompactBackupNotFinished {
		logger.Warn("sync failed with unexpected error", zap.Uint64("id", compactBackupID), zap.Error(retryErr))
		errMsg := fmt.Sprintf("sync status failed: %v", retryErr)
		if updateErr := a.updateCompactBackupStatusToFailed(ctx, compactBackupID, errMsg); updateErr != nil {
			logger.Error("failed to update status on error", zap.Error(updateErr))
		}
	}

	return retryErr
}

func (a *SyncCompactBackupStatusAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	stdCtx := log.WithLogger(ctx.Context(), logger)

	compactBackupID, err := a.getCompactBackupID(ctx, logger)
	if err != nil {
		return err
	}

	// Load compact backup to get cluster information for config
	compactBackup, err := a.compactBackupRepo.Load(stdCtx, compactBackupID)
	if err != nil {
		logger.Error("failed to load compact backup", zap.Uint64("id", compactBackupID), zap.Error(err))
		return err
	}

	// If marked for deletion, don't overwrite the status — the delete DAG owns it now.
	if compactBackup.DeletingOrDeleted() {
		logger.Info("compact backup is being deleted or already deleted, skipping sync",
			zap.String("status", string(compactBackup.Status)), zap.Uint64("id", compactBackupID))
		return nil
	}

	syncTimeout := a.getSyncTimeout(stdCtx, compactBackup.ClusterID.Uint64(), logger)

	ctxWithTimeout, cancel := context.WithTimeout(stdCtx, syncTimeout)
	defer cancel()

	retryErr := a.errorWaiter.Execute(ctxWithTimeout, logger, func() error {
		return a.syncStatusOnce(ctxWithTimeout, compactBackupID, logger)
	})

	if retryErr == nil {
		return nil
	}

	return a.handleSyncError(stdCtx, retryErr, compactBackupID, syncTimeout, logger)
}

// updateCompactBackupStatusToFailed updates the compact backup status to failed with the given error message
func (a *SyncCompactBackupStatusAction) updateCompactBackupStatusToFailed(
	ctx context.Context, compactBackupID uint64, errMsg string,
) error {
	updateCtx, updateCancel := context.WithTimeout(context.Background(), statusUpdateTimeout)
	defer updateCancel()

	return a.transProvider.Transaction(updateCtx, func(txCtx context.Context) error {
		compactBackup, err := a.compactBackupRepo.Load(txCtx, compactBackupID)
		if err != nil {
			return err
		}

		// Only update if not already in a finished status
		if !compactBackup.Status.IsFinishedStatus() {
			compactBackup.Status = entity.CompactBackupStatusFailed
			compactBackup.ErrMsg = errMsg
			now := time.Now()
			compactBackup.FinishedAt = &now

			if err := a.compactBackupRepo.Save(txCtx, compactBackup); err != nil {
				return err
			}
		}

		return nil
	})
}

func (a *SyncCompactBackupStatusAction) ParameterNew() interface{} {
	return nil
}