package enable

import (
	"context"
	"fmt"
	"net/http"

	repo "github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/crrf/common"
	"github.com/tidbcloud/dataflow-service/internal/config"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	"go.uber.org/zap"
)

type enableCRRFDaemonParams struct {
	ClusterID uint64 `json:"cluster_id"`
}

type enableCRRFDaemonAction struct {
	backupSettingRepo backupRepo.BackupSettingRepo
	logBackupRepo     backupRepo.LogBackupRepo
	crrfSettingRepo   crrfRepo.CRRFSettingRepo
	bucketRepo        repo.BucketRepo
	crrfFacade        regionalapi.CRRFFacade
}

// NewEnableCRRFDaemonAction returns a Wrapper that calls EnableCRRF on the cluster-server
// in Run and polls GetCRRF until the daemon reaches RUNNING in RunAfter.
func NewEnableCRRFDaemonAction(
	backupSettingRepo backupRepo.BackupSettingRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
	bucketRepo repo.BucketRepo,
	crrfFacade regionalapi.CRRFFacade,
) *action.Wrapper {
	return action.MustNewActionWrapper(
		workflow.ActionCRRFEnableDaemon,
		&enableCRRFDaemonAction{
			backupSettingRepo: backupSettingRepo,
			logBackupRepo:     logBackupRepo,
			crrfSettingRepo:   crrfSettingRepo,
			bucketRepo:        bucketRepo,
			crrfFacade:        crrfFacade,
		},
	)
}

func (a *enableCRRFDaemonAction) Params() interface{} {
	return &enableCRRFDaemonParams{}
}

// Run calls EnableCRRF on cluster-server to deploy the CRRF daemon.
// It is idempotent: if GetCRRF returns RUNNING or PENDING the call is skipped.
func (a *enableCRRFDaemonAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enableCRRFDaemonParams)

	actCtx, err := common.LoadCRRFActionContext(
		ctx, p.ClusterID,
		a.backupSettingRepo, a.crrfSettingRepo, a.logBackupRepo,
	)
	if err != nil {
		return err
	}

	// Check whether the daemon is already in a terminal or active state.
	existing, bizErr := a.crrfFacade.GetCRRF(ctx, actCtx.MasterRegion, &regionalapi.GetCRRFRequest{
		ClusterId:     fmt.Sprint(p.ClusterID),
		LogBackupName: actCtx.LogBackup.CRName(),
	})
	if bizErr != nil && bizErr.HttpStatus() != http.StatusNotFound {
		return bizErr
	}
	if existing != nil {
		switch existing.Phase {
		case regionalapi.CRRFPhaseRunning, regionalapi.CRRFPhasePending:
			logger.Info("CRRF daemon already active, skipping enable call",
				zap.Uint64("clusterID", p.ClusterID),
				zap.String("phase", string(existing.Phase)))
			return nil
		}
	}

	// Load secondary bucket storage URI for downstream_storage.
	if actCtx.CRRFSetting.SecondaryBucketID == 0 {
		logger.Error("secondary bucket ID not found in CRRF setting, downstream storage will be empty",
			zap.Uint64("clusterID", p.ClusterID),
			zap.Uint64("crrfSettingID", actCtx.CRRFSetting.ID.Uint64()))
		return action.StopRetry
	}
	secondaryBucket, bizErr := a.bucketRepo.Load(ctx, uint64(actCtx.CRRFSetting.SecondaryBucketID))
	if bizErr != nil {
		return bizErr
	}
	downstreamStorage, err := secondaryBucket.URIInBucket(actCtx.LogBackup.Storage)
	if err != nil {
		logger.Warn("failed to build downstream storage URI, falling back to raw log backup storage",
			zap.Uint64("clusterID", p.ClusterID),
			zap.Uint64("secondaryBucketID", secondaryBucket.ID.Uint64()),
			zap.Error(err))
		return err
	}

	req := &regionalapi.EnableCRRFRequest{
		ClusterId:          fmt.Sprint(p.ClusterID),
		LogBackupName:      actCtx.LogBackup.CRName(),
		CrrfId:             regionalapi.CRRFDaemonID(uint64(actCtx.CRRFSetting.ID)),
		DownstreamStorage:  downstreamStorage,
		StateStorageSubDir: "", // keep default
		Image:              "",
	}
	if err = a.overrideCRRFEnableRequest(ctx, req, actCtx); err != nil {
		logger.Warn("failed to apply CRRF enable request overrides", zap.Error(err))
		return err
	}
	if _, bizErr := a.crrfFacade.EnableCRRF(ctx, actCtx.MasterRegion, req); bizErr != nil {
		logger.Warn("EnableCRRF call failed", zap.Error(bizErr))
		return bizErr
	}

	logger.Info("EnableCRRF call succeeded",
		zap.Uint64("clusterID", p.ClusterID),
		zap.String("logBackupName", actCtx.LogBackup.CRName()),
		zap.String("crrfId", req.CrrfId))
	return nil
}

// RunAfter polls GetCRRF until the daemon reaches RUNNING.
func (a *enableCRRFDaemonAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enableCRRFDaemonParams)

	actCtx, err := common.LoadCRRFActionContext(
		ctx, p.ClusterID,
		a.backupSettingRepo, a.crrfSettingRepo, a.logBackupRepo,
	)
	if err != nil {
		return err
	}

	info, bizErr := a.crrfFacade.GetCRRF(ctx, actCtx.MasterRegion, &regionalapi.GetCRRFRequest{
		ClusterId:     fmt.Sprint(p.ClusterID),
		LogBackupName: actCtx.LogBackup.CRName(),
	})
	if bizErr != nil {
		return bizErr
	}

	switch info.Phase {
	case regionalapi.CRRFPhaseRunning:
		logger.Info("CRRF daemon is RUNNING",
			zap.Uint64("clusterID", p.ClusterID),
			zap.String("crrfId", regionalapi.CRRFDaemonID(uint64(actCtx.CRRFSetting.ID))))
		return nil
	case regionalapi.CRRFPhaseFailed:
		logger.Error("CRRF daemon entered FAILED phase",
			zap.Uint64("clusterID", p.ClusterID),
			zap.String("message", info.Message))
		return action.StopRetry
	default:
		return fmt.Errorf("CRRF daemon not yet RUNNING for cluster %d (phase: %s)", p.ClusterID, info.Phase)
	}
}

func (a *enableCRRFDaemonAction) overrideCRRFEnableRequest(ctx context.Context, req *regionalapi.EnableCRRFRequest, actCtx *common.CRRFActionContext) error {
	setting := actCtx.CRRFSetting
	for _, override := range config.Global().BackupConfig.CRRFOverrides {
		// pick the first override that matches the cluster.
		if _, enabled := override.Gate.Enabled(config.FeatureGateQueryParams{
			Provider:  actCtx.MasterRegion.CloudProvider,
			Region:    actCtx.MasterRegion.Name,
			TenantID:  setting.TenantID.Uint64(),
			ProjectID: setting.ProjectID.Uint64(),
			ClusterID: setting.ClusterID.Uint64(),
		}); enabled {
			if override.Image != "" {
				req.Image = override.Image
			}

			break
		}
	}
	return nil
}