package hook

import (
	"context"
	"testing"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	repo_entity "github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow"
)

const testBackupID uint64 = 42

// dagInsWith builds a fastflow DagInstance for the create-snapshot DAG with
// the given backup_id var and reason set.
func dagInsWith(dagID, reason string, backupIDVar string) *entity.DagInstance {
	di := &entity.DagInstance{
		DagID:  dagID,
		Reason: reason,
	}
	di.ID = "test-dag-ins"
	if backupIDVar != "" {
		di.Vars = entity.DagInstanceVars{
			workflow.KeyBackupID: entity.DagInstanceVar{Value: backupIDVar},
		}
	}
	return di
}

// makeBackup builds an entity.Backup with the given status. Other fields stay
// zero so the hook handles defaulting (FinishedAt etc).
func makeBackup(status repo_entity.BackupStatus) *repo_entity.Backup {
	b := &repo_entity.Backup{Status: status}
	b.ID = repo_entity.ID(testBackupID)
	return b
}

// DAG-ID routing now lives in chainBeforeFail (see
// TestChainBeforeFail_NonOwnedDag_SkipsHandle); Handle assumes it is only
// called for an owned DAG.

func TestHandle_MissingBackupIDVar_NoOp(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	h := NewBackupDagFailureHook(repo)

	h.Handle(context.Background(), dagInsWith(workflow.DagCreateSnapshotBackup, "boom", ""))

	repo.AssertNotCalled(t, "Load", mock.Anything, mock.Anything)
}

func TestHandle_UnparseableBackupID_NoOp(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	h := NewBackupDagFailureHook(repo)

	h.Handle(context.Background(), dagInsWith(workflow.DagCreateSnapshotBackup, "boom", "not-a-number"))

	repo.AssertNotCalled(t, "Load", mock.Anything, mock.Anything)
}

func TestHandle_BackupAlreadyInFinalStatus_NoSave(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	repo.On("Load", mock.Anything, testBackupID).
		Return(makeBackup(repo_entity.BackupStatusSuccess), nil).Once()

	h := NewBackupDagFailureHook(repo)
	h.Handle(context.Background(), dagInsWith(workflow.DagCreateSnapshotBackup, "boom", "42"))

	// Backup is already success — must not be overwritten by the hook.
	repo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestHandle_RunningBackup_MarksFailedWithReason(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	loaded := makeBackup(repo_entity.BackupStatusRunning)
	repo.On("Load", mock.Anything, testBackupID).Return(loaded, nil).Once()

	var saved *repo_entity.Backup
	repo.On("Save", mock.Anything, mock.MatchedBy(func(b *repo_entity.Backup) bool {
		saved = b
		return true
	})).Return(nil).Once()

	h := NewBackupDagFailureHook(repo)
	h.Handle(context.Background(),
		dagInsWith(workflow.DagCreateSnapshotBackup, "task sync_snapshot_backup_cr_status timeout", "42"))

	require.NotNil(t, saved)
	require.Equal(t, repo_entity.BackupStatusFailed, saved.Status)
	require.Contains(t, saved.ErrMsg, "task sync_snapshot_backup_cr_status timeout")
	require.NotNil(t, saved.FinishedAt, "FinishedAt should be set when transitioning to final state")
}

func TestHandle_NoReason_StillMarksFailedWithSentinel(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	loaded := makeBackup(repo_entity.BackupStatusPending)
	repo.On("Load", mock.Anything, testBackupID).Return(loaded, nil).Once()

	var saved *repo_entity.Backup
	repo.On("Save", mock.Anything, mock.MatchedBy(func(b *repo_entity.Backup) bool {
		saved = b
		return true
	})).Return(nil).Once()

	h := NewBackupDagFailureHook(repo)
	// No reason — hook should still fall back to a sentinel ErrMsg, not leave it blank.
	h.Handle(context.Background(),
		dagInsWith(workflow.DagCreateWeeklySnapshotBackup, "", "42"))

	require.Equal(t, repo_entity.BackupStatusFailed, saved.Status)
	require.NotEmpty(t, saved.ErrMsg)
}

func TestHandle_PreservesExistingErrMsg(t *testing.T) {
	repo := repo_mock.NewBackupRepo(t)
	loaded := makeBackup(repo_entity.BackupStatusRunning)
	loaded.ErrMsg = "original cr-side error"
	repo.On("Load", mock.Anything, testBackupID).Return(loaded, nil).Once()

	var saved *repo_entity.Backup
	repo.On("Save", mock.Anything, mock.MatchedBy(func(b *repo_entity.Backup) bool {
		saved = b
		return true
	})).Return(nil).Once()

	h := NewBackupDagFailureHook(repo)
	h.Handle(context.Background(),
		dagInsWith(workflow.DagCreateSnapshotBackup, "dag aborted", "42"))

	// We should NOT clobber an existing ErrMsg that came from the dataplane —
	// it carries more diagnostic value than the generic DAG reason.
	require.Equal(t, "original cr-side error", saved.ErrMsg)
}