package mysql

import (
	"context"
	"encoding/json"
	"os"
	"reflect"
	"sync"
	"testing"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/platform/events"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
)

func grantTaskRecovery(f *executionFixture) {
	f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY_APPEND(JSON_EXTRACT(permissions,'$.actions'),'$','tasks:review','$','tasks:retry','$','tasks:audit:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
}
func unknownTask(t *testing.T, f *executionFixture) (service.TaskView, worker.AttemptRef) {
	t.Helper()
	v := f.submit(1)
	ref := retryStart(t, f)
	retryFinish(t, f, ref, "UNKNOWN", false)
	v = f.get(v.ID)
	if v.State != pulse.TaskNeedsReview || !v.EffectUncertain {
		t.Fatal("fixture did not reach uncertainty")
	}
	return v, ref
}
func reviewInput(v service.TaskView, outcome pulse.TaskState) service.ReviewInput {
	return service.ReviewInput{ExpectedRevision: v.Revision, Reason: "operator verified execution and external result", EvidenceRef: "evidence:reconciliation-001", ExecutionSettled: true, Outcome: outcome, Result: json.RawMessage(`{"verified":true,"n":9007199254740993}`)}
}
func manualRetryInput(v service.TaskView) service.ManualRetryInput {
	return service.ManualRetryInput{ExpectedRevision: v.Revision, Reason: "operator confirmed safe same-operation retry", EvidenceRef: "evidence:safe-retry-001", SafeToRetry: true, TaskTimeoutMS: 600000}
}

func TestIntegrationTaskRecovery(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, stop := context.WithTimeout(context.Background(), 90*time.Second)
	defer stop()
	db, err := Open(ctx, dsn, 32, 8)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err := Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	fixture := func(t *testing.T) *executionFixture {
		f := newExecutionFixture(t, ctx, db)
		grantTaskRecovery(f)
		return f
	}
	t.Run("review_preserves_attempts_and_publishes_new_authoritative_result", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		grantTaskRecovery(f.executionFixture)
		sub := f.create(events.CreateInput{})
		v, ref := unknownTask(t, f.executionFixture)
		before, err := f.tasks.GetTaskAttempt(ctx, f.token, v.ID, ref.AttemptID)
		if err != nil {
			t.Fatal(err)
		}
		input := reviewInput(v, pulse.TaskSucceeded)
		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
		if err != nil || out.Task.State != pulse.TaskSucceeded || out.Task.EffectUncertain || out.Task.AttemptCount != v.AttemptCount || out.Task.Revision != v.Revision+1 || string(out.Task.Result) != string(input.Result) {
			t.Fatal(out, err)
		}
		after, err := f.tasks.GetTaskAttempt(ctx, f.token, v.ID, ref.AttemptID)
		if err != nil || !reflect.DeepEqual(before, after) {
			t.Fatal("review rewrote worker history", err)
		}
		assertQueueCount(t, f.executionFixture, 0)
		assertApplicationCounts(t, f.executionFixture, 0, 0, 0)
		replay, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
		if err != nil || !replay.Deduplicated || replay.AuditID != out.AuditID || replay.Task.Revision != out.Task.Revision {
			t.Fatal("review identity changed", err)
		}
		input.Outcome = pulse.TaskFailed
		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
		expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
		audit, err := f.tasks.GetTaskOperation(ctx, f.token, v.ID, out.AuditID)
		if err != nil || audit.EvidenceRef != "evidence:reconciliation-001" || audit.ActorID != string(f.token.CredentialID) || audit.Action != "task.review" {
			t.Fatal(audit, err)
		}
		// 旧UNKNOWN结果仍作为原快照交付；核查终态成为下一条独立不可变事件。
		g := f.grant(sub.ID)
		if g.Event.Type != "task.needs_review" || !g.Event.Snapshot.EffectUncertain {
			t.Fatal("old snapshot was rewritten", g.Event)
		}
		f.ack(sub.ID, g)
		g = f.grant(sub.ID)
		if g.Event.Type != "task.succeeded" || g.Event.Snapshot.EffectUncertain || g.Event.Snapshot.Revision != out.Task.Revision {
			t.Fatal("review result not journaled", g.Event)
		}
		f.ack(sub.ID, g)
		late := retryFinish(t, f.executionFixture, ref, "SUCCEEDED", false)
		if late.Receipt.Disposition != "EVIDENCE_ONLY" || f.get(v.ID).Revision != out.Task.Revision {
			t.Fatal("late worker result overwrote review")
		}
		if f.count("orbit_audit_events", "resource_id=? AND action='task.review'", v.ID) != 1 {
			t.Fatal("duplicate review audit")
		}
	})
	t.Run("review_authority_is_explicit_and_history_is_scoped", func(t *testing.T) {
		f := newExecutionFixture(t, ctx, db)
		v, _ := unknownTask(t, f)
		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		_, err = f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{})
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		grantTaskRecovery(f)
		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		if err != nil {
			t.Fatal(err)
		}
		g := fixture(t)
		_, err = g.tasks.GetTaskOperation(ctx, g.token, v.ID, out.AuditID)
		expectWorkerCode(t, err, "NOT_FOUND")
		other := f.submit(1)
		_, err = f.tasks.GetTaskOperation(ctx, f.token, other.ID, out.AuditID)
		expectWorkerCode(t, err, "NOT_FOUND")
		page, err := f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1})
		if err != nil || len(page.Items) != 1 || page.NextCursor == "" {
			t.Fatal(page, err)
		}
		next, err := f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1, Cursor: page.NextCursor})
		if err != nil || len(next.Items) != 1 || next.Items[0].ID == page.Items[0].ID {
			t.Fatal(next, err)
		}
		_, err = f.tasks.ListTaskOperations(ctx, f.token, other.ID, service.HistoryPageInput{Cursor: page.NextCursor})
		expectWorkerCode(t, err, "INVALID_ARGUMENT")
		_, err = f.tasks.ListTaskAttempts(ctx, f.token, v.ID, service.HistoryPageInput{Cursor: page.NextCursor})
		expectWorkerCode(t, err, "INVALID_ARGUMENT")
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(g.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		expectWorkerCode(t, err, "NOT_FOUND")
	})
	t.Run("review_cas_competition_and_atomic_audit_failure", func(t *testing.T) {
		f := fixture(t)
		v, _ := unknownTask(t, f)
		input := reviewInput(v, pulse.TaskFailed)
		trigger := "review_audit_" + string(executionID(t))
		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled review audit failure'")
		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
		f.exec("DROP TRIGGER " + trigger)
		if err == nil || !reflect.DeepEqual(v, f.get(v.ID)) {
			t.Fatal("failed audit did not roll back review", err)
		}
		assertQueueCount(t, f, 1)
		assertApplicationCounts(t, f, 1, 0, 0)
		if f.count("orbit_idempotency_keys", "operation='task.review'") != 0 || f.count("orbit_result_events", "1=1") != 1 {
			t.Fatal("review leaked identity or result event")
		}
		keys := []string{"review", "competing-review"}
		ch := make(chan error, 2)
		var wg sync.WaitGroup
		for _, key := range keys {
			wg.Add(1)
			go func() { defer wg.Done(); _, err := f.tasks.ReviewTask(ctx, f.token, v.ID, key, input); ch <- err }()
		}
		wg.Wait()
		close(ch)
		success, conflicts := 0, 0
		for err := range ch {
			if err == nil {
				success++
			} else if fault.Code(err) == "STATE_CONFLICT" {
				conflicts++
			} else {
				t.Fatal(err)
			}
		}
		if success != 1 || conflicts != 1 {
			t.Fatal(success, conflicts)
		}
		assertQueueCount(t, f, 0)
	})
	t.Run("retry_new_identity_preserves_source_and_respects_backpressure", func(t *testing.T) {
		f := fixture(t)
		configureResources(t, f, 10, 4, 1, false)
		v, oldRef := unknownTask(t, f)
		_, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(v))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		review, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		if err != nil {
			t.Fatal(err)
		}
		v = review.Task
		occupant := f.submit(1)
		in := manualRetryInput(v)
		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
		expectWorkerCode(t, err, "QUEUE_FULL")
		if f.count("pulse_tasks", "retry_of=?", v.ID) != 0 || f.count("orbit_idempotency_keys", "operation='task.retry'") != 0 {
			t.Fatal("quota rejection left a child")
		}
		if _, err := f.tasks.CancelTask(ctx, f.token, occupant.ID, "free", service.CancelInput{ExpectedRevision: occupant.Revision, Reason: "free retry quota"}); err != nil {
			t.Fatal(err)
		}
		out, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
		if err != nil || out.Task.ID == v.ID || out.Task.RetryOf == nil || *out.Task.RetryOf != v.ID || out.Task.AttemptCount != 0 || out.Task.Revision != 1 || out.Task.Retry != v.Retry || string(out.Task.Payload) != string(v.Payload) || !reflect.DeepEqual(v, f.get(v.ID)) {
			t.Fatal(out, err)
		}
		if !out.Task.DeadlineAt.After(out.Task.CreatedAt) || out.Task.DeadlineAt.Sub(out.Task.CreatedAt) != 10*time.Minute {
			t.Fatal("new budget incorrect")
		}
		replay, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
		if err != nil || !replay.Deduplicated || replay.Task.ID != out.Task.ID || replay.AuditID != out.AuditID {
			t.Fatal("retry replay changed identity", err)
		}
		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "another", in)
		expectWorkerCode(t, err, "STATE_CONFLICT")
		late := retryFinish(t, f, oldRef, "SUCCEEDED", false)
		if late.Receipt.Disposition != "EVIDENCE_ONLY" || !reflect.DeepEqual(v, f.get(v.ID)) {
			t.Fatal("old worker changed resolved task")
		}
		assignment, ref := f.claim()
		if assignment.Task.ID != out.Task.ID || assignment.Task.RetryOf == nil || *assignment.Task.RetryOf != v.ID {
			t.Fatal("worker lost retry lineage")
		}
		if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
			t.Fatal(err)
		}
		retryFinish(t, f, ref, "SUCCEEDED", false)
		assertQueueCount(t, f, 0)
	})
	t.Run("retry_competing_keys_and_audit_rollback", func(t *testing.T) {
		f := fixture(t)
		v := f.submit(1)
		cancelled, err := f.tasks.CancelTask(ctx, f.token, v.ID, "cancel", service.CancelInput{ExpectedRevision: v.Revision, Reason: "stop original"})
		if err != nil {
			t.Fatal(err)
		}
		v = cancelled.Task
		trigger := "retry_audit_" + string(executionID(t))
		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled retry audit failure'")
		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(v))
		f.exec("DROP TRIGGER " + trigger)
		if err == nil || f.count("pulse_tasks", "retry_of=?", v.ID) != 0 {
			t.Fatal("retry audit failure leaked child", err)
		}
		assertQueueCount(t, f, 0)
		ch := make(chan error, 8)
		var wg sync.WaitGroup
		for i := 0; i < 8; i++ {
			key := string(executionID(t))
			wg.Add(1)
			go func() {
				defer wg.Done()
				_, err := f.tasks.RetryTask(ctx, f.token, v.ID, key, manualRetryInput(v))
				ch <- err
			}()
		}
		wg.Wait()
		close(ch)
		success := 0
		for err := range ch {
			if err == nil {
				success++
			} else if fault.Code(err) != "STATE_CONFLICT" {
				t.Fatal(err)
			}
		}
		if success != 1 || f.count("pulse_tasks", "retry_of=?", v.ID) != 1 || f.count("orbit_audit_events", "action='task.retry'") != 1 {
			t.Fatal("multiple manual children", success)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
	})
	t.Run("review_unblocks_serial_schedule_without_manual_child", func(t *testing.T) {
		f := newScheduleFixture(t, newExecutionFixture(t, ctx, db))
		grantTaskRecovery(f.executionFixture)
		in := f.definition()
		in.Policy.Overlap = "SERIAL"
		plan := f.create(in)
		f.due(plan, 2*time.Second)
		f.tick()
		items := f.occurrences(plan.ID)
		if len(items) != 1 || items[0].TaskID == nil {
			t.Fatal("missing scheduled task")
		}
		ref := retryStart(t, f.executionFixture)
		retryFinish(t, f.executionFixture, ref, "UNKNOWN", false)
		v := f.get(*items[0].TaskID)
		f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, plan.ID)
		f.tick()
		if len(f.occurrences(plan.ID)) != 1 {
			t.Fatal("unknown execution did not block serial schedule")
		}
		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		if err != nil {
			t.Fatal(err)
		}
		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(out.Task))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, plan.ID)
		f.tick()
		if len(f.occurrences(plan.ID)) != 2 {
			t.Fatal("review did not release serial schedule")
		}
	})
	t.Run("recovery_rejects_gates_bad_owner_active_and_success", func(t *testing.T) {
		f := fixture(t)
		waiting := f.submit(1)
		_, err := f.tasks.ReviewTask(ctx, f.token, waiting.ID, "waiting", reviewInput(waiting, pulse.TaskFailed))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		ref := retryStart(t, f)
		active := f.get(waiting.ID)
		_, err = f.tasks.ReviewTask(ctx, f.token, active.ID, "active", reviewInput(active, pulse.TaskFailed))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		retryFinish(t, f, ref, "SUCCEEDED", false)
		success := f.get(waiting.ID)
		_, err = f.tasks.RetryTask(ctx, f.token, success.ID, "retry", manualRetryInput(success))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		v, _ := unknownTask(t, f)
		f.exec(`UPDATE pulse_tasks SET gate_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		f.exec(`UPDATE pulse_tasks SET gate_id=NULL,owner_kind='schedule',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		if f.get(v.ID).Revision != v.Revision {
			t.Fatal("rejected recovery changed task")
		}
	})
	t.Run("managed_tasks_cannot_bypass_recovery_owner", func(t *testing.T) {
		f := fixture(t)
		v, _ := unknownTask(t, f)
		f.exec(`UPDATE pulse_tasks SET owner_kind='workflow',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		expectWorkerCode(t, err, "STATE_CONFLICT")
		f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
		if err != nil {
			t.Fatal(err)
		}
		f.exec(`UPDATE pulse_tasks SET owner_kind='schedule',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(out.Task))
		expectWorkerCode(t, err, "STATE_CONFLICT")
	})
}