package impl

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/gorhill/cronexpr"
	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	backup_model "github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	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/internal/metrics"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

const (
	AUTO_BACKUP                             = "auto_backup"
	AUTO_BACKUP_SCHEDULER_CHECK_TIME_OFFSET = 5

	// reconcileStuckRunningMinAge is how old a backup must be before we hit the
	// dataplane to ask its current phase. Set to 6h because normal BR runs
	// 5-14h depending on cluster size, and below that threshold the
	// `sync_snapshot_backup_cr_status` action's own 15s polling is the
	// authoritative source — querying again from this worker would just be
	// redundant regional-API calls. We only step in once a backup looks "long"
	// by any cluster's standard.
	reconcileStuckRunningMinAge = 6 * time.Hour

	// markFailedFromDagMinAge is how old a backup must be before we look at
	// the create DAG state in fastflow. Much smaller than the dataplane
	// threshold above because this is a cheap local DB query — we want to
	// catch a stuck DAG (e.g. timed out, panicked, manually canceled) as soon
	// as practically possible after the create-flow has visibly given up.
	markFailedFromDagMinAge = 30 * time.Minute

	// reconcileStuckRunningBatchSize bounds work per tick so a flood of stuck
	// rows doesn't hammer the regional API or fastflow store.
	reconcileStuckRunningBatchSize = 200
)

type backupHealthCheckerImpl struct {
	backupRepo         backup_repo.BackupRepo
	backupExportRepo   backup_repo.BackupExportRepo
	logBackupRepo      backup_repo.LogBackupRepo
	backupScheduleRepo backup_repo.BackupScheduleRepo

	backupSettingModel backup_model.BackupSettingModel

	coreGrpcClient       grpcli.CoreGrpcClient
	mgmtCli              grpcli.MgmtSvcGrpcClient
	backupRegionalFacade infraapi.BackupFacade
	workflowModel        workflow.Workflow
}

func NewBackupHealthChecker(
	backupRepo backup_repo.BackupRepo,
	backupExportRepo backup_repo.BackupExportRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	backupScheduleRepo backup_repo.BackupScheduleRepo,
	backupSettingModel backup_model.BackupSettingModel,
	coreGrpcClient grpcli.CoreGrpcClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	backupRegionalFacade infraapi.BackupFacade,
	workflowModel workflow.Workflow,
) backup_model.BackupHealthChecker {
	return &backupHealthCheckerImpl{
		backupRepo:           backupRepo,
		backupExportRepo:     backupExportRepo,
		logBackupRepo:        logBackupRepo,
		backupScheduleRepo:   backupScheduleRepo,
		backupSettingModel:   backupSettingModel,
		coreGrpcClient:       coreGrpcClient,
		mgmtCli:              mgmtCli,
		backupRegionalFacade: backupRegionalFacade,
		workflowModel:        workflowModel,
	}
}

// will check the auto backup scheduler running status.
// there is an offset of this check to avoid unexpected issue.
// if current time is 2020.01.01 10:00:00 and timeRangeInMinute is 30 min.
// we will check the schedule from 2020.01.01 09:25:00 to 2020.01.01 09:55:00
func (w *backupHealthCheckerImpl) CheckAutoBackupSchedule(
	ctx context.Context, timeRangeInMinute int64,
) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.Any("worker", "check-auto-backups")))
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))

	enableAutoBackupClusterIDs, clusterMap, err := w.getEnableAutoBackupClusterIDsAndMap(ctx)
	if err != nil {
		return err
	}
	if len(enableAutoBackupClusterIDs) == 0 {
		return nil
	}

	endTime := time.Now().Add(-time.Duration(AUTO_BACKUP_SCHEDULER_CHECK_TIME_OFFSET) * time.Minute)
	startTime := endTime.Add(-time.Duration(timeRangeInMinute) * time.Minute)
	backupSchedules, backups, err := w.getAutoSchedulesAndBackups(ctx, enableAutoBackupClusterIDs, clusterMap,
		startTime, endTime)
	if err != nil {
		return err
	}
	if backupSchedules == nil {
		return nil
	}

	backupsMap := make(map[entity.ID]entity.Backup)
	for _, backup := range backups {
		backupsMap[backup.BackupScheduleID] = backup
	}

	for _, schedule := range backupSchedules {
		if _, ok := backupsMap[schedule.ID]; !ok {
			logger.Error("Auto backup was not successfully scheduled",
				zap.Uint64("clusterID", uint64(schedule.ClusterID)),
				zap.Uint64("backupScheduleID", uint64(schedule.ID)),
				zap.Any("backupSchedule", schedule))
			if cluster, gotCluster := clusterMap[uint64(schedule.ClusterID)]; !gotCluster {
				logger.Error("Fail to get cluster when check auto backup",
					zap.Uint64("clusterID", uint64(schedule.ClusterID)),
					zap.Uint64("backupScheduleID", uint64(schedule.ID)))
			} else {
				metrics.BackupSchedulerAbnormal.WithLabelValues(
					fmt.Sprintf("%d", cluster.TenantId),
					fmt.Sprintf("%d", cluster.ProjectId),
					fmt.Sprintf("%d", schedule.ClusterID),
					"",
					fmt.Sprintf("%d", schedule.ID),
					AUTO_BACKUP).Inc()
			}
		}
	}
	return nil
}

func (w *backupHealthCheckerImpl) getEnableAutoBackupClusterIDsAndMap(ctx context.Context,
) ([]entity.ID, map[uint64]*core.Cluster, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))

	listReq := &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			Statuses: []core.ClusterStatus{
				core.ClusterStatus_CLUSTER_STATUS_NORMAL,
			},
			IsDevTier: pointer.To(false),
		},
	}
	listClustersResp, err := w.coreGrpcClient.ListClustersWithoutIds(ctx, listReq)
	if err != nil {
		logger.Error("fail to list clusters", zap.Error(err))
		return nil, nil, err
	}
	if len(listClustersResp.Clusters) == 0 {
		logger.Warn("No alive dedicated clusters")
		return nil, nil, nil
	}
	var clusterIds []uint64
	for _, cluster := range listClustersResp.Clusters {
		clusterIds = append(clusterIds, cluster.Id)
	}

	listBackupSettingReq := &domain.ListBackupSettingReq{
		Filter: &domain.ListBackupSettingFilter{
			ClusterIDs: clusterIds,
		},
	}
	if err != nil {
		logger.Error("fail to list BackupSettings", zap.Error(err))
		return nil, nil, err
	}
	backupSettingResp, err := w.backupSettingModel.ListBackupSetting(ctx, listBackupSettingReq)
	if err != nil {
		return nil, nil, err
	}
	if len(backupSettingResp.Items) == 0 {
		logger.Warn("No BackupSettings")
		return nil, nil, nil
	}
	var enableAutoBackupClusterIDs []entity.ID
	for _, backupSetting := range backupSettingResp.Items {
		if backupSetting.BackupSetting.AutoBackup != nil &&
			!backupSetting.BackupSetting.AutoBackup.Disable {
			enableAutoBackupClusterIDs = append(enableAutoBackupClusterIDs, backupSetting.BackupSetting.ClusterID)
		}
	}

	return enableAutoBackupClusterIDs, listClustersResp.Clusters, nil
}

func (w *backupHealthCheckerImpl) getAutoSchedulesAndBackups(ctx context.Context,
	enableAutoBackupClusterIDs []entity.ID,
	clusterMap map[uint64]*core.Cluster,
	startTime time.Time,
	endTime time.Time,
) ([]entity.BackupSchedule, []entity.Backup, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))

	backupSchedules, err := w.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.ClusterTypeEq(entity.BackupClusterTypeDedicated).
				ClusterIDIn(enableAutoBackupClusterIDs...).
				DeletedAtIsNull().
				TypeEq(entity.BackupTypeAuto)
		})
	if err != nil {
		logger.Error("get auto backup schedule failed", zap.Error(err))
		return nil, nil, err
	}
	if len(backupSchedules) == 0 {
		logger.Warn("No backup schedule")
		return nil, nil, nil
	}
	checkSchedulesAndClustersCorresponding(ctx, backupSchedules, enableAutoBackupClusterIDs, clusterMap)
	schedulesInTimeRange := getSchedulesInTimeRange(backupSchedules, startTime, endTime)
	if len(schedulesInTimeRange) == 0 {
		logger.Warn("No backup schedule in time range")
		return nil, nil, nil
	}
	backups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.BackupScheduleIDIn(pie.Map(schedulesInTimeRange, func(t entity.BackupSchedule) entity.ID {
			return t.ID
		})...).
			CreatedAtGte(startTime)
	})
	if err != nil {
		logger.Error("get backups failed", zap.Error(err))
		return nil, nil, err
	}
	return schedulesInTimeRange, backups, nil
}

func getSchedulesInTimeRange(
	backupSchedules []entity.BackupSchedule,
	startTime time.Time,
	endTime time.Time,
) []entity.BackupSchedule {
	var schedulesInTimeRange []entity.BackupSchedule
	for _, backupSchedule := range backupSchedules {
		lastAutoBackupTime := getLastAutoBackupTime(backupSchedule, endTime)
		if lastAutoBackupTime.Before(startTime) ||
			lastAutoBackupTime.After(endTime) ||
			lastAutoBackupTime.Before(backupSchedule.CreatedAt) {
			continue
		}
		schedulesInTimeRange = append(schedulesInTimeRange, backupSchedule)
	}
	return schedulesInTimeRange
}

func checkSchedulesAndClustersCorresponding(
	ctx context.Context,
	backupSchedules []entity.BackupSchedule,
	enableAutoBackupClusterIDs []entity.ID,
	clusterMap map[uint64]*core.Cluster,
) {
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))

	if len(backupSchedules) != len(enableAutoBackupClusterIDs) {
		schedulesMap := make(map[entity.ID]entity.BackupSchedule)
		for _, schedule := range backupSchedules {
			schedulesMap[schedule.ClusterID] = schedule
		}
		for _, clusterId := range enableAutoBackupClusterIDs {
			if _, ok := schedulesMap[clusterId]; !ok {
				logger.Error("Auto backup was schedule is not successfully added.",
					zap.Uint64("clusterID", uint64(clusterId)))
				if cluster, gotCluster := clusterMap[uint64(clusterId)]; !gotCluster {
					logger.Error("Fail to get cluster when check auto backup schedule",
						zap.Uint64("clusterID", uint64(clusterId)))
				} else {
					metrics.BackupSchedulerAbnormal.WithLabelValues(fmt.Sprintf("%d", cluster.TenantId),
						fmt.Sprintf("%d", cluster.ProjectId), fmt.Sprintf("%d", clusterId), "", "NO_SCHEDULE",
						AUTO_BACKUP).Inc()
				}
			}
		}
	}
}

func getLastAutoBackupTime(schedule entity.BackupSchedule, curTime time.Time) time.Time {
	expr := cronexpr.MustParse(schedule.Schedule)
	if schedule.BackupCycle == entity.BackupCycleDaily {
		nextTime := expr.Next(curTime)
		cronInterval := expr.Next(nextTime).Sub(nextTime)
		return nextTime.Add(-cronInterval)
	}
	weekBeforeCurrent := curTime.Add(-7 * 24 * time.Hour)
	preTime := weekBeforeCurrent
	nextTime := preTime
	for nextTime.Before(curTime) {
		preTime = nextTime
		nextTime = expr.Next(preTime)
	}
	return preTime
}

//nolint:funlen
func (w *backupHealthCheckerImpl) CheckSnapshotBackupCleanUp(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.Any("worker", "check-snapshot-backups-clean-up")))
	logger := log.GetLogger(ctx).With(zap.Any("cluster type", entity.BackupClusterTypeDedicated))
	backupSchedules, backups, err := w.listAutoScheduleAndBackups(ctx)
	if err != nil {
		logger.Error("list backup schedules and backups error", zap.Error(err))
		return err
	}
	schedulesMap := make(map[entity.ID]entity.BackupSchedule)
	for _, schedule := range backupSchedules {
		schedulesMap[schedule.ID] = schedule
	}
	scheduleBackups := make(map[entity.ID][]entity.Backup, len(backupSchedules))
	for i := range backups {
		backupScheduleID := backups[i].BackupScheduleID
		scheduleBackups[backupScheduleID] = append(scheduleBackups[backupScheduleID], backups[i])
	}

	yestarday := time.Now().Add(-24 * time.Hour)
	for backupScheduleID, oneScheduleBackups := range scheduleBackups {
		sort.Slice(oneScheduleBackups, func(i, j int) bool {
			return oneScheduleBackups[i].StartTimeStamp < oneScheduleBackups[j].StartTimeStamp
		})
		schedule := schedulesMap[backupScheduleID]
		backScheduleType, err := queryBackScheduleType(ctx, logger, w.logBackupRepo, &schedule)
		if err != nil {
			return err
		}
		// check if there are some backups that should be deleted yesterday
		candidates, err := calDeleteAutoBackups(
			logger, oneScheduleBackups, schedule.ExpiredTSO(yestarday), backScheduleType == domain.PiTRSchedule)
		if err != nil {
			return err
		}
		if len(candidates) == 0 {
			logger.Info("no backup need to be deleted for the auto backup schedule.",
				zap.Uint64("clusterID", uint64(oneScheduleBackups[0].ClusterID)),
				zap.Uint64("scheduleID", uint64(oneScheduleBackups[0].BackupScheduleID)),
			)

			metrics.BackupCleanUpAbnormal.DeletePartialMatch(
				map[string]string{
					metrics.ClusterIDLabel:        fmt.Sprintf("%d", schedule.ClusterID),
					metrics.BackupScheduleIDLabel: fmt.Sprintf("%d", schedule.ID),
				},
			)
			continue
		}

		logger.Warn(
			"there is more than one backup need to be deleted for the auto backup schedule.",
			zap.Uint64("clusterID", uint64(oneScheduleBackups[0].ClusterID)),
			zap.Uint64("scheduleID", uint64(oneScheduleBackups[0].BackupScheduleID)),
			zap.Uint64("len", uint64(len(candidates))))
		for _, backup := range candidates {
			logger.Error("Backup was is not successfully cleaned up.",
				zap.Uint64("clusterID", uint64(backup.ClusterID)),
				zap.Uint64("scheduleID", uint64(backup.BackupScheduleID)),
				zap.Uint64("backupID", uint64(backup.ID)))
			metrics.BackupCleanUpAbnormal.WithLabelValues(
				fmt.Sprintf("%d", backup.ClusterID),
				fmt.Sprintf("%d", backup.ID),
				fmt.Sprintf("%d", backup.BackupScheduleID)).
				Set(1)
		}
	}
	return nil
}

func (w *backupHealthCheckerImpl) listAutoScheduleAndBackups(
	ctx context.Context,
) ([]entity.BackupSchedule, []entity.Backup, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backupSchedules, err := w.backupScheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.ClusterTypeEq(entity.BackupClusterTypeDedicated).
				RetentionPeriodNe(0).
				TypeEq(entity.BackupTypeAuto).
				CleanedUpEq(false)
		})
	if err != nil {
		return nil, nil, err.Chain("list backup schedules")
	}
	if len(backupSchedules) == 0 {
		return nil, nil, nil
	}
	// fixme: currently the backups are limited. If we have too many backups later, we should consider multi-relational
	// queries. (2024.02.24, 100+ schedules and thousands backups)
	backups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.BackupScheduleIDIn(pie.Map(backupSchedules, func(t entity.BackupSchedule) entity.ID {
			return t.ID
		})...).
			DeletedAtIsNull()
	})
	if err != nil {
		return nil, nil, err.Chain("list backups")
	}
	logger.Sugar().Infof("there are %d backup schedules and %d backups", len(backupSchedules), len(backups))
	return backupSchedules, backups, nil
}

func (w *backupHealthCheckerImpl) CheckBackupNodeGroupAndNodePoolCleanUp(ctx context.Context) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("CheckBackupNodeGroupAndNodePoolCleanUp is deprecated, skipping")
	return nil
}

// ReconcileStuckRunningBackups walks every backup whose mgmt status is still
// `running` (or `pending`) past `reconcileStuckRunningMinAge` and pulls the
// authoritative phase from the dataplane Backup CR. When the CR has reached a
// final phase (Failed/Complete) but mgmt missed it — typically because the
// create DAG's `sync_snapshot_backup_cr_status` action has already timed out —
// the mgmt row is updated so that downstream cleanup logic
// (`clean_failed_snapshot_backup_storage`, `DeleteExpiredAutoBackups`) can run
// and the orphan S3 data is deleted.
func (w *backupHealthCheckerImpl) ReconcileStuckRunningBackups(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.String("worker", "reconcile-stuck-running-backups")))
	logger := log.GetLogger(ctx)

	cutoff := time.Now().Add(-reconcileStuckRunningMinAge)
	stuckBackups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.
			StatusIn(entity.BackupStatusRunning, entity.BackupStatusPending).
			CreatedAtLt(cutoff).
			DeletedAtIsNull().
			OrderAscByCreatedAt().
			Limit(reconcileStuckRunningBatchSize)
	})
	if err != nil {
		logger.Error("list stuck running backups failed", zap.Error(err))
		return err
	}
	if len(stuckBackups) == 0 {
		return nil
	}
	logger.Info("found stuck running backups", zap.Int("count", len(stuckBackups)))

	for i := range stuckBackups {
		backup := &stuckBackups[i]
		w.reconcileOneStuckBackup(ctx, backup)
	}
	return nil
}

func (w *backupHealthCheckerImpl) reconcileOneStuckBackup(ctx context.Context, backup *entity.Backup) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("backup_id", backup.ID.Uint64()),
		zap.Uint64("cluster_id", backup.ClusterID.Uint64()),
		zap.String("current_status", string(backup.Status)),
		zap.Time("created_at", backup.CreatedAt),
	)

	clusterID := backup.ClusterID.Uint64()
	cluster, err := w.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		logger.Warn("get cluster error, skip this round", zap.Error(err))
		return
	}

	// Mirror sync_cr_status behavior: if the cluster has been deleted, the CR
	// can never reach a final phase — short-circuit to Terminated so the row
	// stops blocking alerts and downstream cleanup can proceed.
	dc := cluster.GetDedicatedCluster()
	if dc != nil && dc.GetState() == dedicatedv1.Cluster_DELETED {
		w.markBackupTerminated(ctx, logger, backup, "cluster has been deleted")
		return
	}

	cr, infraErr := w.backupRegionalFacade.GetBackup(
		ctx,
		cluster.GetCloudProvider(),
		cluster.GetRegion(),
		backup.TenantID.Uint64(),
		backup.ProjectID.Uint64(),
		clusterID,
		backup.CRName(),
	)
	if infraErr != nil {
		logger.Warn("get backup CR from dataplane failed, skip this round", zap.Error(infraErr))
		return
	}

	phase := cr.Status.Phase
	logger = logger.With(zap.Any("dataplane_phase", phase))

	switch phase {
	case "":
		// CR exists but hasn't reported a phase yet. backup-manager hasn't
		// finished initial Pending/Scheduled handshake. Leave it alone.
		logger.Info("dataplane phase is empty, leave running for now")
	case cloudv1.BackupPrepare, cloudv1.BackupScheduled, cloudv1.BackupRunning:
		// BR is genuinely making progress (or about to). Don't mark failed.
		logger.Info("BR still in progress on dataplane, no reconcile needed")
	case cloudv1.BackupFailed:
		w.markBackupFinalFromCR(ctx, logger, backup, cr, entity.BackupStatusFailed)
	case cloudv1.BackupComplete:
		// We promised the user we'd "mark failed if failed", but if BR has
		// actually succeeded we must also reflect that — otherwise the row
		// stays running forever, the alert keeps firing, and the next backup
		// schedule may double-schedule.
		w.markBackupFinalFromCR(ctx, logger, backup, cr, entity.BackupStatusSuccess)
	default:
		logger.Warn("unknown dataplane phase, no reconcile")
	}
}

func (w *backupHealthCheckerImpl) markBackupFinalFromCR(
	ctx context.Context, logger *zap.Logger, backup *entity.Backup,
	cr *cloudv1.Backup, finalStatus entity.BackupStatus,
) {
	now := time.Now()
	backup.Status = finalStatus
	if backup.StartTimeStamp == "" {
		backup.StartTimeStamp = cr.Status.BackupTS
	}
	if backup.FinishedAt == nil {
		backup.FinishedAt = &now
	}
	if backup.BackupSize == 0 {
		backup.BackupSize = int(cr.Status.BackupSize)
	}
	if saveErr := w.backupRepo.Save(ctx, backup); saveErr != nil {
		logger.Error("save reconciled backup failed", zap.Error(saveErr),
			zap.String("target_status", string(finalStatus)))
		return
	}
	logger.Info("reconciled backup status from dataplane CR",
		zap.String("new_status", string(finalStatus)),
		zap.Int("backup_size", backup.BackupSize))
}

func (w *backupHealthCheckerImpl) markBackupTerminated(
	ctx context.Context, logger *zap.Logger, backup *entity.Backup, reason string,
) {
	now := time.Now()
	backup.Status = entity.BackupStatusTerminated
	if backup.ErrMsg == "" {
		backup.ErrMsg = reason
	}
	if backup.FinishedAt == nil {
		backup.FinishedAt = &now
	}
	if saveErr := w.backupRepo.Save(ctx, backup); saveErr != nil {
		logger.Error("save terminated backup failed", zap.Error(saveErr))
		return
	}
	logger.Info("backup terminated by reconciler", zap.String("reason", reason))
}

// snapshotBackupCreateDagIDs are the create-DAG IDs whose Failed/Canceled
// state implies the corresponding mgmt backup row should be marked Failed.
// Kept in this package (not imported from app/workflow/hook) to avoid an
// import cycle: hook depends on the container which depends on this model.
var snapshotBackupCreateDagIDs = []string{
	workflow.DagCreateSnapshotBackup,
	workflow.DagCreateWeeklySnapshotBackup,
}

// MarkBackupsFailedFromFailedDag walks every backup whose mgmt status is
// stuck in `running`/`pending` past `reconcileStuckRunningMinAge` and looks up
// the snapshot create DAG instance by tags. If the DAG is in Failed/Canceled
// state, the mgmt row is marked Failed with the DAG.Reason recorded so the
// user understands why. This is the legacy/fallback path: it catches backups
// whose DAG failed BEFORE the BeforeFail hook (see
// app/workflow/hook/backup_dag_failure.go) was deployed, and any case the
// hook missed.
func (w *backupHealthCheckerImpl) MarkBackupsFailedFromFailedDag(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.String("worker", "mark-backups-failed-from-failed-dag")))
	logger := log.GetLogger(ctx)

	cutoff := time.Now().Add(-markFailedFromDagMinAge)
	stuckBackups, err := w.backupRepo.List(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.
			StatusIn(entity.BackupStatusRunning, entity.BackupStatusPending).
			CreatedAtLt(cutoff).
			DeletedAtIsNull().
			OrderAscByCreatedAt().
			Limit(reconcileStuckRunningBatchSize)
	})
	if err != nil {
		logger.Error("list stuck running backups failed", zap.Error(err))
		return err
	}
	if len(stuckBackups) == 0 {
		return nil
	}
	logger.Info("checking stuck backups against fastflow DAG state", zap.Int("count", len(stuckBackups)))

	for i := range stuckBackups {
		w.markOneBackupFromFailedDag(ctx, &stuckBackups[i])
	}
	return nil
}

func (w *backupHealthCheckerImpl) markOneBackupFromFailedDag(ctx context.Context, backup *entity.Backup) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("backup_id", backup.ID.Uint64()),
		zap.Uint64("cluster_id", backup.ClusterID.Uint64()),
		zap.String("current_status", string(backup.Status)),
	)

	failedDagIns := w.findFailedCreateDag(ctx, logger, backup.ID.Uint64())
	if failedDagIns == nil {
		// Either no DAG exists (very unlikely for a backup row) or it is still
		// running — either way, this worker has nothing to do. The reconciler
		// or sync_cr_status path will handle the live cases.
		return
	}
	logger = logger.With(
		zap.String("dag_id", failedDagIns.DagID),
		zap.String("dag_ins_id", failedDagIns.ID),
		zap.String("dag_status", string(failedDagIns.Status)),
		zap.String("dag_reason", failedDagIns.Reason),
	)

	now := time.Now()
	backup.Status = entity.BackupStatusFailed
	if backup.ErrMsg == "" {
		if failedDagIns.Reason != "" {
			backup.ErrMsg = "backup DAG " + string(failedDagIns.Status) + ": " + failedDagIns.Reason
		} else {
			backup.ErrMsg = "backup DAG " + string(failedDagIns.Status)
		}
	}
	if backup.FinishedAt == nil {
		backup.FinishedAt = &now
	}
	if saveErr := w.backupRepo.Save(ctx, backup); saveErr != nil {
		logger.Error("save backup failed status", zap.Error(saveErr))
		return
	}
	logger.Info("marked backup as failed because its create DAG is in final failure state",
		zap.String("err_msg", backup.ErrMsg))
}

// findFailedCreateDag returns the most relevant Failed/Canceled snapshot
// create DAG instance for the given backup_id, or nil if none qualify.
//
// Implementation note: we deliberately bypass fastflow's ListDagIns + Tags
// path here. That path expands to
//
//	WHERE (`key`,`value`) IN (... 3 pairs ...)
//	GROUP BY dag_ins_id HAVING total = 3
//
// which fans out across every `business_type=snapshot_backup` /
// `business_action=snapshot_backup_create` tag row ever written before the
// HAVING filter prunes it. With a large `dag_instance_tags` table that
// query has been observed to time out at ~5s and never finish.
//
// FindDagInstancesByBusinessID instead does an index lookup on the
// high-selectivity `business_id` tag first, then filters the (tiny)
// candidate set by dag_id + status — sub-millisecond regardless of table
// size.
func (w *backupHealthCheckerImpl) findFailedCreateDag(
	ctx context.Context, logger *zap.Logger, backupID uint64,
) *fastflow_entity.DagInstance {
	dagIns, err := w.workflowModel.FindDagInstancesByBusinessID(
		ctx,
		strconv.FormatUint(backupID, 10),
		snapshotBackupCreateDagIDs,
		[]fastflow_entity.DagInstanceStatus{
			fastflow_entity.DagInstanceStatusFailed,
			fastflow_entity.DagInstanceStatusCanceled,
		},
		1,
	)
	if err != nil {
		logger.Warn("find failed/canceled create dag instance by business_id failed",
			zap.Error(err))
		return nil
	}
	if len(dagIns) == 0 {
		return nil
	}
	return dagIns[0]
}

// backupExportCreateDagIDs are the create-DAG IDs whose Failed/Canceled state
// implies the corresponding backup_export row should be marked Failed.
var backupExportCreateDagIDs = []string{
	workflow.DagCreateAWSBackupExport,
	workflow.DagCreateGCPBackupExport,
}

// MarkBackupExportsFailedFromFailedDag is the backup_export analog of
// MarkBackupsFailedFromFailedDag. It walks every export still stuck in
// `running`/`pending` past markFailedFromDagMinAge and looks up its create DAG
// instance by business_id. If the DAG is in Failed/Canceled state, the export
// row is marked Failed. This catches exports orphaned when the create DAG was
// watchdog-killed at the create_*_batch_ops_job step before its own error
// handler ran (empty job_id, status stuck at running) — the failure mode that
// keeps the BackupExportDurationTooLong alert firing — as well as any export
// the BeforeFail hook missed (e.g. process restart, or rows predating the hook).
func (w *backupHealthCheckerImpl) MarkBackupExportsFailedFromFailedDag(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.String("worker", "mark-backup-exports-failed-from-failed-dag")))
	logger := log.GetLogger(ctx)

	cutoff := time.Now().Add(-markFailedFromDagMinAge)
	stuckExports, err := w.backupExportRepo.ListByFilter(ctx,
		&backup_repo.BackupExportFilter{
			Statuses: []entity.BackupExportStatus{
				entity.BackupExportStatusRunning,
				entity.BackupExportStatusPending,
			},
		},
		func(db *gorm.DB) *gorm.DB {
			return db.Where("created_at < ?", cutoff).
				Order("created_at asc").
				Limit(reconcileStuckRunningBatchSize)
		},
	)
	if err != nil {
		logger.Error("list stuck running backup exports failed", zap.Error(err))
		return err
	}
	if len(stuckExports) == 0 {
		return nil
	}
	logger.Info("checking stuck backup exports against fastflow DAG state", zap.Int("count", len(stuckExports)))

	for i := range stuckExports {
		w.markOneExportFromFailedDag(ctx, &stuckExports[i])
	}
	return nil
}

func (w *backupHealthCheckerImpl) markOneExportFromFailedDag(ctx context.Context, export *entity.BackupExport) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("backup_export_id", export.ID.Uint64()),
		zap.Uint64("cluster_id", export.ClusterID.Uint64()),
		zap.String("current_status", string(export.Status)),
	)

	failedDagIns := w.findFailedExportCreateDag(ctx, logger, export.ID.Uint64())
	if failedDagIns == nil {
		// No DAG, or it is still running — nothing to do this round. The
		// BeforeFail hook handles the live failure case.
		return
	}
	logger = logger.With(
		zap.String("dag_id", failedDagIns.DagID),
		zap.String("dag_ins_id", failedDagIns.ID),
		zap.String("dag_status", string(failedDagIns.Status)),
		zap.String("dag_reason", failedDagIns.Reason),
	)

	now := time.Now()
	export.Status = entity.BackupExportStatusFailed
	if failedDagIns.Reason != "" {
		export.StatusMsg = "backup export DAG " + string(failedDagIns.Status) + ": " + failedDagIns.Reason
	} else {
		export.StatusMsg = "backup export DAG " + string(failedDagIns.Status)
	}
	if export.FinishedAt == nil {
		export.FinishedAt = &now
	}
	if saveErr := w.backupExportRepo.Save(ctx, export); saveErr != nil {
		logger.Error("save backup export failed status", zap.Error(saveErr))
		return
	}
	logger.Info("marked backup export as failed because its create DAG is in final failure state",
		zap.String("status_msg", export.StatusMsg))
}

// findFailedExportCreateDag returns the most relevant Failed/Canceled backup
// export create DAG instance for the given backup_export_id, or nil if none
// qualify. The export create DAG is tagged with business_id == backup_export_id
// (see backup_export.go RunDagWithTags), so we reuse the selective-first
// FindDagInstancesByBusinessID lookup to avoid fastflow's tag fan-out.
func (w *backupHealthCheckerImpl) findFailedExportCreateDag(
	ctx context.Context, logger *zap.Logger, backupExportID uint64,
) *fastflow_entity.DagInstance {
	dagIns, err := w.workflowModel.FindDagInstancesByBusinessID(
		ctx,
		strconv.FormatUint(backupExportID, 10),
		backupExportCreateDagIDs,
		[]fastflow_entity.DagInstanceStatus{
			fastflow_entity.DagInstanceStatusFailed,
			fastflow_entity.DagInstanceStatusCanceled,
		},
		1,
	)
	if err != nil {
		logger.Warn("find failed/canceled export create dag instance by business_id failed",
			zap.Error(err))
		return nil
	}
	if len(dagIns) == 0 {
		return nil
	}
	return dagIns[0]
}