package impl

import (
	"context"
	"fmt"
	"strconv"

	"github.com/tidbcloud/dataflow-service/app/models/crrf"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

var _ crrf.CRRFModel = &crrfModelImpl{}

type crrfModelImpl struct {
	backupSettingRepo backupRepo.BackupSettingRepo
	logBackupRepo     backupRepo.LogBackupRepo
	crrfSettingRepo   crrfRepo.CRRFSettingRepo
	workflowClient    workflow.Workflow
}

func NewCRRFModel(
	backupSettingRepo backupRepo.BackupSettingRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
	workflowClient workflow.Workflow,
) crrf.CRRFModel {
	return &crrfModelImpl{
		backupSettingRepo: backupSettingRepo,
		logBackupRepo:     logBackupRepo,
		crrfSettingRepo:   crrfSettingRepo,
		workflowClient:    workflowClient,
	}
}

func (m *crrfModelImpl) EnableCRRF(ctx context.Context, req *domain.EnableCRRFReq) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("enabling CRRF",
		zap.Uint64("clusterID", req.ClusterID),
		zap.String("secondaryRegion", req.SecondaryRegion))

	if req.SecondaryRegion == "" {
		return bizerr_wrapper.NewInvalidInputError("secondary_region is required")
	}

	// 1. Load backup setting and validate preconditions.
	backupSetting, err := m.backupSettingRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{
		ClusterID: &req.ClusterID,
	})
	if err != nil {
		logger.Error("load backup setting failed", zap.Error(err))
		return err
	}
	if backupSetting == nil {
		return bizerr_wrapper.NewInvalidInputError("backup setting not found for cluster %d", req.ClusterID)
	}
	if backupSetting.Status != entity.BackupSettingStatusNormal &&
		backupSetting.Status != entity.BackupSettingStatusEditingPitr &&
		backupSetting.Status != entity.BackupSettingStatusEditingRegion {
		return bizerr_wrapper.NewInvalidInputError(
			"backup setting is in %s state, must be normal, editing_pitr, or editing_region to enable CRRF",
			backupSetting.Status)
	}

	// Only AWS is supported for CRRF.
	if backupSetting.InfraProviderType != string(domain.InfraProviderTypeAWS) &&
		backupSetting.InfraProviderType != string(domain.InfraProviderTypeEKS) {
		return bizerr_wrapper.NewInvalidInputError("CRRF is only supported on AWS clusters")
	}

	// 2. Check for existing CRRFSetting.
	existing, err := m.crrfSettingRepo.GetByClusterID(ctx, req.ClusterID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return err
	}
	if existing != nil && existing.IsActive() {
		return bizerr_wrapper.NewInvalidInputError(
			"CRRF is already %s for cluster %d", existing.Status, req.ClusterID)
	}

	// 3. Validate dual-region compatibility.
	if backupSetting.Bucket != nil && backupSetting.Bucket.DualRegionEnable {
		secondaryRegion := m.extractSecondaryRegion(backupSetting)
		if secondaryRegion != "" && secondaryRegion != req.SecondaryRegion {
			return bizerr_wrapper.NewInvalidInputError(
				"dual-region is enabled with region %s; disable dual-region first or use the same region",
				secondaryRegion)
		}
	}
	masterRegion, ok := backupSetting.MasterRegion()
	if !ok {
		return bizerr_wrapper.NewInvalidInputError("can't find master region in backup setting for cluster %d", req.ClusterID)
	}

	// 4. Create or update CRRFSetting.
	crrfSetting := existing
	if crrfSetting == nil {
		crrfSetting = &entity.CRRFSetting{
			ClusterID: entity.ID(req.ClusterID),
			ProjectID: entity.ID(req.ProjectID),
			TenantID:  entity.ID(req.TenantID),
			Status:    entity.CRRFSettingStatusDisabled,
		}
	}
	if crrfSetting.Status != entity.CRRFSettingStatusDisabled {
		return bizerr_wrapper.NewInvalidInputError(
			"CRRF must be in %q state to enable, current state is %q for cluster %d",
			entity.CRRFSettingStatusDisabled, crrfSetting.Status, req.ClusterID)
	}
	crrfSetting.Status = entity.CRRFSettingStatusEnabling
	crrfSetting.MasterRegion = masterRegion
	crrfSetting.SecondaryRegion = req.SecondaryRegion
	crrfSetting.ErrMsg = nil
	// Clear stale bucket IDs from any previous (possibly failed) enable attempt.
	// They will be populated below or in the enable DAG.
	crrfSetting.MasterBucketID = 0
	crrfSetting.SecondaryBucketID = 0

	// Populate bucket IDs if dual-region is already enabled.
	if backupSetting.Bucket != nil && backupSetting.Bucket.DualRegionEnable {
		m.populateBucketIDs(crrfSetting, backupSetting)
	}
	// Populate LogBackupID if PiTR is already enabled.
	if backupSetting.PiTR != nil && backupSetting.PiTR.Enable {
		// PiTR is already enabled, populate LogBackupID from the most recent running log backup.
		logBackups, bizErr := m.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
			return qs.ClusterIDEq(entity.ID(req.ClusterID)).
				StatusNotIn(entity.LogBackupStatusFailed, entity.LogBackupStatusStopping, entity.LogBackupStatusStopped).
				OrderDescByID()
		})
		if bizErr != nil {
			return bizErr
		}
		if len(logBackups) != 1 {
			return bizerr_wrapper.NewInternalErr("expected exactly 1 running log backup for cluster %d, found %d", req.ClusterID, len(logBackups))
		}
		crrfSetting.LogBackupID = logBackups[0].ID
	}

	if err := m.crrfSettingRepo.Save(ctx, crrfSetting); err != nil {
		return err.Chain("save crrf setting to enabling")
	}

	// 5. Start enable_crrf workflow DAG (which handles dual-region, PiTR, and periodic backup).
	dagInsVars := map[string]string{
		workflow.KeyCRRFSettingID:            strconv.FormatUint(uint64(crrfSetting.ID), 10),
		workflow.KeyBackupSettingId:          strconv.FormatUint(uint64(backupSetting.ID), 10),
		workflow.ClusterID:                   strconv.FormatUint(req.ClusterID, 10),
		workflow.KeySecondaryRegion:          req.SecondaryRegion,
		workflow.KeySkipEnablePiTRDualRegion: strconv.FormatBool(req.SkipEnablePiTRAndDualRegion),
	}
	ins, dagErr := m.workflowClient.RunDagWithTags(
		workflow.DagEnableCRRF, dagInsVars,
		map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeCRRF),
			workflow.BusinessActionKey: string(workflow.BusinessActionEnableCRRF),
			workflow.BusinessID:        strconv.FormatUint(uint64(crrfSetting.ID), 10),
			workflow.ClusterID:         strconv.FormatUint(req.ClusterID, 10),
		})
	if dagErr != nil {
		logger.Error("failed to start enable_crrf DAG", zap.Error(dagErr))
		m.markCRRFDisabledWithError(ctx, crrfSetting, fmt.Sprintf("failed to start workflow: %v", dagErr))
		return bizerr_wrapper.NewInternalErr("failed to start CRRF enable workflow")
	}

	logger.Info("CRRF enable workflow started",
		zap.Uint64("clusterID", req.ClusterID),
		zap.Uint64("crrfSettingID", uint64(crrfSetting.ID)),
		zap.String("dagInsID", ins.ID),
	)
	return nil
}

func (m *crrfModelImpl) QueryCRRF(ctx context.Context, clusterID uint64) (*domain.CRRFInfo, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Info("querying CRRF", zap.Uint64("clusterID", clusterID))

	setting, err := m.crrfSettingRepo.GetByClusterID(ctx, clusterID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err
	}

	info := &domain.CRRFInfo{
		ClusterID: clusterID,
		Status:    entity.CRRFSettingStatusDisabled,
	}
	if setting == nil {
		return info, nil
	}

	info.Status = setting.Status
	info.MasterRegion = setting.MasterRegion
	info.SecondaryRegion = setting.SecondaryRegion
	info.MasterBucketID = uint64(setting.MasterBucketID)
	info.SecondaryBucketID = uint64(setting.SecondaryBucketID)
	if setting.ErrMsg != nil {
		info.ErrorMessage = *setting.ErrMsg
	}

	return info, nil
}

func (m *crrfModelImpl) DisableCRRF(ctx context.Context, req *domain.DisableCRRFReq) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("disabling CRRF", zap.Uint64("clusterID", req.ClusterID))

	setting, err := m.crrfSettingRepo.GetByClusterID(ctx, req.ClusterID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return err
	}
	if setting == nil {
		return bizerr_wrapper.NewInvalidInputError("CRRF is not configured for cluster %d", req.ClusterID)
	}
	if setting.Status != entity.CRRFSettingStatusEnabled {
		return bizerr_wrapper.NewInvalidInputError(
			"CRRF must be in %q state to disable, current state is %q for cluster %d",
			entity.CRRFSettingStatusEnabled, setting.Status, req.ClusterID)
	}

	// Transition to disabling state and start the disable DAG.
	setting.Status = entity.CRRFSettingStatusDisabling
	if err := m.crrfSettingRepo.Save(ctx, setting); err != nil {
		return err.Chain("save crrf setting to disabling")
	}

	ins, dagErr := m.workflowClient.RunDagWithTags(
		workflow.DagDisableCRRF,
		map[string]string{
			workflow.KeyCRRFSettingID: strconv.FormatUint(uint64(setting.ID), 10),
			workflow.ClusterID:        strconv.FormatUint(req.ClusterID, 10),
		},
		map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeCRRF),
			workflow.BusinessActionKey: string(workflow.BusinessActionDisableCRRF),
			workflow.BusinessID:        strconv.FormatUint(uint64(setting.ID), 10),
			workflow.ClusterID:         strconv.FormatUint(req.ClusterID, 10),
		})
	if dagErr != nil {
		logger.Error("failed to start disable_crrf DAG", zap.Error(dagErr))
		m.markCRRFFailedWithError(ctx, setting, fmt.Sprintf("failed to start disable workflow: %v", dagErr))
		return bizerr_wrapper.NewInternalErr("failed to start CRRF disable workflow")
	}

	logger.Info("CRRF disable workflow started",
		zap.Uint64("clusterID", req.ClusterID),
		zap.Uint64("crrfSettingID", uint64(setting.ID)),
		zap.String("dagInsID", ins.ID),
	)
	return nil
}

// markCRRFDisabledWithError reverts CRRFSetting to disabled and records the error.
func (m *crrfModelImpl) markCRRFDisabledWithError(ctx context.Context, setting *entity.CRRFSetting, errMsg string) {
	setting.Status = entity.CRRFSettingStatusDisabled
	setting.ErrMsg = &errMsg
	if err := m.crrfSettingRepo.Save(ctx, setting); err != nil {
		log.GetLogger(ctx).Error("failed to revert CRRF setting to disabled on error",
			zap.Error(err), zap.Uint64("crrfSettingID", uint64(setting.ID)))
	}
}

// markCRRFFailedWithError sets CRRFSetting to failed and records the error.
func (m *crrfModelImpl) markCRRFFailedWithError(ctx context.Context, setting *entity.CRRFSetting, errMsg string) {
	setting.Status = entity.CRRFSettingStatusFailed
	setting.ErrMsg = &errMsg
	if err := m.crrfSettingRepo.Save(ctx, setting); err != nil {
		log.GetLogger(ctx).Error("failed to mark CRRF setting as failed",
			zap.Error(err), zap.Uint64("crrfSettingID", uint64(setting.ID)))
	}
}

// extractSecondaryRegion returns the secondary region from a dual-region bucket setting.
func (m *crrfModelImpl) extractSecondaryRegion(bs *entity.BackupSetting) string {
	if bs.Bucket == nil {
		return ""
	}
	for _, b := range bs.Bucket.Buckets {
		if !b.Master && len(b.Regions) > 0 {
			return b.Regions[0]
		}
	}
	return ""
}

// populateBucketIDs fills the master and secondary bucket IDs from the backup setting.
func (m *crrfModelImpl) populateBucketIDs(crrf *entity.CRRFSetting, bs *entity.BackupSetting) {
	if bs.Bucket == nil {
		return
	}
	for _, b := range bs.Bucket.Buckets {
		if b.Master {
			crrf.MasterBucketID = entity.ID(b.BucketId)
		} else {
			crrf.SecondaryBucketID = entity.ID(b.BucketId)
		}
	}
}