// Package hook wires global fastflow lifecycle hooks for our DAGs. The hooks
// are process-wide: registering them once at startup is enough.
package hook

import (
	"context"
	"strconv"
	"time"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/dataflow-service/app/container"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	repo_entity "github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
)

// snapshotBackupCreateDagIDs are the DAG IDs whose failure should mark the
// associated mgmt backup row as Failed. Without this hook a DAG abort (e.g.
// `sync_snapshot_backup_cr_status` hitting its timeout) leaves backup.Status
// at `running` indefinitely, which keeps the duration alert firing and blocks
// `clean_failed_snapshot_backup_storage` from ever cleaning up the S3 data.
var snapshotBackupCreateDagIDs = map[string]struct{}{
	workflow.DagCreateSnapshotBackup:       {},
	workflow.DagCreateWeeklySnapshotBackup: {},
}

// BackupDagFailureHook owns the BeforeFail handler for snapshot backup create
// DAGs. Holding the BackupRepo as a field (rather than reaching into
// container.Global() inline) keeps the core handler unit-testable — see
// backup_dag_failure_test.go.
type BackupDagFailureHook struct {
	backupRepo backup_repo.BackupRepo
}

// NewBackupDagFailureHook wires a hook against the given BackupRepo.
func NewBackupDagFailureHook(backupRepo backup_repo.BackupRepo) *BackupDagFailureHook {
	return &BackupDagFailureHook{backupRepo: backupRepo}
}

// Register attaches Handle to the global fastflow DagInstance lifecycle. Must
// be called once at startup before any snapshot backup DAG instance can fail.
// If called more than once, it chains with any previously registered
// BeforeFail hook and invokes the earlier hook first.
func (h *BackupDagFailureHook) Register() {
	chainBeforeFail("backup-dag-failure-hook", snapshotBackupCreateDagIDs, h.Handle)
}

// Handle marks the backup row associated with a failed create DAG as Failed
// (only if it isn't already in a final state). DAG-ID routing and nil-guarding
// are handled by chainBeforeFail, so Handle is only ever called for an owned DAG.
func (h *BackupDagFailureHook) Handle(ctx context.Context, dagIns *entity.DagInstance) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("worker", "backup-dag-failure-hook"),
		zap.String("dag_id", dagIns.DagID),
		zap.String("dag_ins_id", dagIns.ID),
		zap.String("dag_reason", dagIns.Reason),
	)

	backupIDStr := ""
	if dagIns.Vars != nil {
		if v, ok := dagIns.Vars[workflow.KeyBackupID]; ok {
			backupIDStr = v.Value
		}
	}
	if backupIDStr == "" {
		logger.Warn("dag instance has no backup_id var, skip")
		return
	}
	backupID, err := strconv.ParseUint(backupIDStr, 10, 64)
	if err != nil {
		logger.Warn("parse backup_id failed", zap.String("raw", backupIDStr), zap.Error(err))
		return
	}
	logger = logger.With(zap.Uint64("backup_id", backupID))

	backup, loadErr := h.backupRepo.Load(ctx, backupID)
	if loadErr != nil {
		logger.Error("load backup failed", zap.Error(loadErr))
		return
	}
	if backup.InFinalStatus() {
		logger.Info("backup already in final status, no action",
			zap.String("current_status", string(backup.Status)))
		return
	}

	now := time.Now()
	backup.Status = repo_entity.BackupStatusFailed
	if backup.ErrMsg == "" {
		if dagIns.Reason != "" {
			backup.ErrMsg = "backup DAG failed: " + dagIns.Reason
		} else {
			backup.ErrMsg = "backup DAG failed without reason"
		}
	}
	if backup.FinishedAt == nil {
		backup.FinishedAt = &now
	}
	if saveErr := h.backupRepo.Save(ctx, backup); saveErr != nil {
		logger.Error("save backup status failed", zap.Error(saveErr))
		return
	}
	logger.Info("marked backup as failed via DAG failure hook",
		zap.String("err_msg", backup.ErrMsg))
}

// RegisterBackupDagFailureHook is the convenience entrypoint used by the
// fastflow bootstrap: pulls the BackupRepo from the global container and
// installs the hook.
func RegisterBackupDagFailureHook() {
	NewBackupDagFailureHook(container.Global().BackupRepo).Register()
}