package run_test

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	workflowrun "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
)

func TestWorkflowArtifactTurnCommitReadAuthorizationAndRollback(t *testing.T) {
	ctx := context.Background()
	db := startRollbackDB(t, ctx)
	fixture := seedRunFixture(t, ctx, db, "artifact")
	store := workflowrun.NewWithOptions(db, workflowrun.StoreOptions{WorkspaceGroupID: fixture.GroupID})
	claimed, err := store.ClaimRunnableRuns(ctx, workflowrun.ClaimRunnableParams{WorkerID: "artifact-worker", Now: fixture.Now, LeaseTTL: timeMinute, Limit: 1})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	turn := claimed[0]
	content := []byte(`{"aggregate":{"bytes":"lossless","items":[1,2,3]}}`)
	draft, err := runartifact.NewDraftForKind(runartifact.KindJoinAggregate, fixture.WorkspaceID, fixture.RunID, "join", "act_join", content)
	require.NoError(t, err)
	state := json.RawMessage(`{"status":"waiting","input":{},"variables":{},"output":{"$workflowArtifact":` + mustJSON(t, draft.Ref) + `},"activations":[],"scopes":[],"nextActivation":1,"nextScope":0}`)
	saved, err := store.SaveTurn(ctx, workflowrun.SaveTurnParams{
		Fence: turn.Fence, ExpectedRevision: turn.Run.Revision, Status: workflowrun.RunStatusWaiting,
		StateJSON: state, Artifacts: []runartifact.Draft{draft}, RunnableAt: fixture.Now, ReleaseLease: true, Now: fixture.Now,
	})
	require.NoError(t, err)
	require.JSONEq(t, string(state), string(saved.Run.StateJSON))

	page, err := store.ReadArtifactPage(ctx, draft.Ref, "root", []string{"aggregate"}, "", 16)
	require.NoError(t, err)
	require.NotEmpty(t, page.NextCursor)
	all := append([]byte(nil), page.Bytes...)
	cursor := page.NextCursor
	for cursor != "" {
		next, nextErr := store.ReadArtifactPage(ctx, draft.Ref, "root", []string{"aggregate"}, cursor, 16)
		require.NoError(t, nextErr)
		all = append(all, next.Bytes...)
		cursor = next.NextCursor
	}
	require.Equal(t, []byte(`{"bytes":"lossless","items":[1,2,3]}`), all)
	require.Equal(t, page.TotalBytes, len(all))

	wrongRun := draft.Ref
	wrongRun.RunID = "wri_other"
	_, err = store.ReadArtifactPage(ctx, wrongRun, "root", nil, "", 16)
	require.Error(t, err)
	wrongDigest := draft.Ref
	wrongDigest.Digest = "sha256:" + strings.Repeat("0", 64)
	_, err = store.ReadArtifactPage(ctx, wrongDigest, "root", nil, "", 16)
	require.Error(t, err)

	// The immutable insert is replay-safe: another committed turn reuses the
	// deterministic identity/content instead of creating a duplicate blob.
	claimed, err = store.ClaimRunnableRuns(ctx, workflowrun.ClaimRunnableParams{WorkerID: "artifact-worker-2", Now: fixture.Now, LeaseTTL: timeMinute, Limit: 1})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	second := claimed[0]
	_, err = store.SaveTurn(ctx, workflowrun.SaveTurnParams{
		Fence: second.Fence, ExpectedRevision: second.Run.Revision, Status: workflowrun.RunStatusWaiting,
		StateJSON: state, Artifacts: []runartifact.Draft{draft}, RunnableAt: fixture.Now, ReleaseLease: true, Now: fixture.Now.Add(timeSecond),
	})
	require.NoError(t, err)
	var count int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_run_artifacts WHERE workspace_id=$1 AND run_id=$2`, fixture.WorkspaceID, fixture.RunID).Scan(&count))
	require.Equal(t, 1, count)

	// A child mutation failure rolls the artifact write back with the whole
	// reducer turn; no unreachable blob is observable after the failed commit.
	claimed, err = store.ClaimRunnableRuns(ctx, workflowrun.ClaimRunnableParams{WorkerID: "artifact-worker-3", Now: fixture.Now, LeaseTTL: timeMinute, Limit: 1})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	third := claimed[0]
	other, err := runartifact.NewDraftForKind(runartifact.KindJoinAggregate, fixture.WorkspaceID, fixture.RunID, "join", "act_other", []byte(`{"new":true}`))
	require.NoError(t, err)
	_, err = store.SaveTurn(ctx, workflowrun.SaveTurnParams{
		Fence: third.Fence, ExpectedRevision: third.Run.Revision, Status: workflowrun.RunStatusWaiting,
		StateJSON: state, Artifacts: []runartifact.Draft{other}, RunnableAt: fixture.Now, ReleaseLease: true, Now: fixture.Now.Add(2 * timeSecond),
		Activations: []workflowrun.ActivationMutation{{Operation: "unknown", Value: workflowrun.Activation{ID: "bad"}}},
	})
	require.Error(t, err)
	_, err = store.ReadArtifactPage(ctx, other.Ref, "root", nil, "", 16)
	require.Error(t, err)

	// Cancellation is an authorization fence for new page consumption. A
	// tampered immutable blob is detected before bytes are returned, and Run
	// deletion cascades metadata plus blob rows without leaving an orphan.
	mustExec(t, ctx, db, `UPDATE workflow_run_instances SET status='canceled' WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, fixture.RunID)
	_, err = store.ReadArtifactPage(ctx, draft.Ref, "root", nil, "", 16)
	require.Error(t, err)
	maintenance, err := store.RunArtifactMaintenance(ctx, fixture.Now, 10)
	require.NoError(t, err)
	require.EqualValues(t, 1, maintenance.AssignedExpiry, "terminal artifacts receive retention only during maintenance")
	require.Zero(t, maintenance.Deleted)
	var expiresAt time.Time
	require.NoError(t, db.QueryRow(ctx, `SELECT expires_at FROM workflow_run_artifacts WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, draft.Ref.ArtifactID).Scan(&expiresAt))
	require.WithinDuration(t, fixture.Now.Add(7*24*time.Hour), expiresAt, time.Microsecond)
	mustExec(t, ctx, db, `UPDATE workflow_run_instances SET status='waiting' WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, fixture.RunID)
	mustExec(t, ctx, db, `UPDATE workflow_run_artifacts SET expires_at=$3 WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, draft.Ref.ArtifactID, fixture.Now.Add(-time.Second))
	_, err = store.ReadArtifactPage(ctx, draft.Ref, "root", nil, "", 16)
	require.Error(t, err, "expired artifacts must fail closed before bytes are returned")
	// Tamper while the metadata row still exists; integrity checking must run
	// before cleanup, not after the alleged deletion.
	mustExec(t, ctx, db, `UPDATE workflow_run_artifact_blobs SET content='corrupt' WHERE workspace_id=$1 AND artifact_id=$2`, fixture.WorkspaceID, draft.Ref.ArtifactID)
	_, err = store.ReadArtifactPage(ctx, draft.Ref, "root", nil, "", 16)
	require.Error(t, err, "an extant tampered blob must fail closed")
	deleted, err := store.SweepExpiredArtifacts(ctx, fixture.Now, 10)
	require.NoError(t, err)
	require.Zero(t, deleted, "a retained waiting Run cannot be swept")
	mustExec(t, ctx, db, `UPDATE workflow_run_instances SET status='canceled' WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, fixture.RunID)
	deleted, err = store.SweepExpiredArtifacts(ctx, fixture.Now, 10)
	require.NoError(t, err)
	require.EqualValues(t, 1, deleted, "terminal expired artifact can be swept")
	mustExec(t, ctx, db, `DELETE FROM workflow_run_instances WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, fixture.RunID)
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_run_artifacts WHERE workspace_id=$1 AND run_id=$2`, fixture.WorkspaceID, fixture.RunID).Scan(&count))
	require.Zero(t, count)
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_run_artifact_blobs WHERE workspace_id=$1 AND artifact_id=$2`, fixture.WorkspaceID, draft.Ref.ArtifactID).Scan(&count))
	require.Zero(t, count)
}

func TestWorkflowArtifactMaintenanceIsBounded(t *testing.T) {
	ctx := context.Background()
	db := startRollbackDB(t, ctx)
	fixture := seedRunFixture(t, ctx, db, "artifact-bounded")
	store := workflowrun.NewWithOptions(db, workflowrun.StoreOptions{WorkspaceGroupID: fixture.GroupID})
	claimed, err := store.ClaimRunnableRuns(ctx, workflowrun.ClaimRunnableParams{WorkerID: "artifact-bounded-worker", Now: fixture.Now, LeaseTTL: timeMinute, Limit: 1})
	require.Len(t, claimed, 1)
	drafts := make([]runartifact.Draft, 0, 3)
	for index := 0; index < 3; index++ {
		draft, draftErr := runartifact.NewDraftForKind(runartifact.KindJoinAggregate, fixture.WorkspaceID, fixture.RunID, "join", fmt.Sprintf("act_%d", index), []byte(fmt.Sprintf(`{"index":%d}`, index)))
		require.NoError(t, draftErr)
		drafts = append(drafts, draft)
	}
	state := json.RawMessage(`{"status":"waiting","output":` + mustJSON(t, runartifact.Marker(drafts[0].Ref)) + `}`)
	_, err = store.SaveTurn(ctx, workflowrun.SaveTurnParams{
		Fence: claimed[0].Fence, ExpectedRevision: claimed[0].Run.Revision, Status: workflowrun.RunStatusWaiting,
		StateJSON: state, Artifacts: drafts, RunnableAt: fixture.Now, ReleaseLease: true, Now: fixture.Now,
	})
	require.NoError(t, err)
	mustExec(t, ctx, db, `UPDATE workflow_run_instances SET status='canceled' WHERE workspace_id=$1 AND id=$2`, fixture.WorkspaceID, fixture.RunID)

	first, err := store.RunArtifactMaintenance(ctx, fixture.Now, 2)
	require.NoError(t, err)
	require.EqualValues(t, 2, first.AssignedExpiry)
	var assigned int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_run_artifacts WHERE workspace_id=$1 AND run_id=$2 AND expires_at IS NOT NULL`, fixture.WorkspaceID, fixture.RunID).Scan(&assigned))
	require.Equal(t, 2, assigned, "one pass must assign at most its artifact budget")
	second, err := store.RunArtifactMaintenance(ctx, fixture.Now, 2)
	require.NoError(t, err)
	require.EqualValues(t, 1, second.AssignedExpiry)
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_run_artifacts WHERE workspace_id=$1 AND run_id=$2 AND expires_at IS NOT NULL`, fixture.WorkspaceID, fixture.RunID).Scan(&assigned))
	require.Equal(t, 3, assigned)
}

const (
	timeMinute = 60 * 1000000000
	timeSecond = 1000000000
)

func mustJSON(t *testing.T, value any) string {
	t.Helper()
	encoded, err := json.Marshal(value)
	require.NoError(t, err)
	return string(encoded)
}