package controlplane

import (
	"testing"
	"time"

	wq "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

func TestProjectionRangeCutoverLifecycleBoundariesTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	t.Run("terminal_history_without_handoff_is_preserved", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		projectionCutoverExec(t, db, `DELETE FROM _System_Migration`)
		e := rangeProofEvent("terminal", 3)
		insertProjectionCutoverEvent(t, db, e)
		projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET ExecutionOwner='event',Status='processed'`)
		insertProjectionCutoverWork(t, db, rangeProofWork("terminal", e, "processed"))
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
		if err := VerifyConnectorExecutionProtocols(t.Context(), db); err != nil {
			t.Fatalf("safe terminal history must not prevent handoff: %v", err)
		}
		requireProjectionHistoryUnchanged(t, db, events, works)
		if cutoff, found, err := readProjectionProtocol(t.Context(), db, e.identity.InstalledPackID); err != nil || !found || cutoff != 3 {
			t.Fatalf("terminal checkpoint=%d/%v/%v", cutoff, found, err)
		}
	})
	t.Run("deleted_route_alone_needs_no_fence", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		projectionCutoverExec(t, db, `UPDATE _System_ConnectorInboundRoute SET IsDeleted=1`)
		projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorIngressControl`)
		if err := VerifyConnectorExecutionProtocols(t.Context(), db); err != nil {
			t.Fatalf("retained route without work is not a cutover candidate: %v", err)
		}
		requireNoProjectionCutoverWrites(t, db, nil, nil)
	})
	for _, owner := range []string{"blank", "unknown", "deleted"} {
		for _, kind := range []string{"event", "workitem"} {
			t.Run("unowned_"+owner+"_"+kind, func(t *testing.T) {
				resetProjectionCutoverFixture(t, db)
				e := rangeProofEvent("orphan", 1)
				switch owner {
				case "blank":
					e.identity.InstalledPackID = ""
				case "unknown":
					e.identity.InstalledPackID = "unknown-incarnation"
				case "deleted":
					projectionCutoverExec(t, db, `UPDATE _System_InstalledPack SET IsDeleted=1`)
				}
				if kind == "event" {
					insertProjectionCutoverEvent(t, db, e)
					projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='pending',ExecutionOwner='event'`)
				} else {
					insertProjectionCutoverWork(t, db, rangeProofWork("orphan", e, "pending"))
				}
				events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
				works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
				if err := VerifyConnectorExecutionProtocols(t.Context(), db); err == nil {
					t.Fatal("gate accepted unowned executable history")
				}
				requireNoProjectionCutoverWrites(t, db, events, works)
			})
		}
	}
	t.Run("blank_source_does_not_merge_distinct_inputs", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		a, b := rangeProofEvent("unresolved-source-a", 1), rangeProofEvent("finished-source-b", 2)
		a.identity.SourceLaneKey, b.identity.SourceLaneKey = "", ""
		insertProjectionCutoverEvent(t, db, a)
		insertProjectionCutoverEvent(t, db, b)
		projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='pending',ExecutionOwner='event',Payload='{"objectKind":"record","sourceID":"a","payload":{"id":"a"}}' WHERE Id=?`, a.id)
		projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Payload='{"objectKind":"record","sourceID":"b","payload":{"id":"b"}}' WHERE Id=?`, b.id)
		insertProjectionCutoverWork(t, db, rangeProofWork("finished-source-b", b, "processed"))
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
		if _, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, a.identity.InstalledPackID); err == nil {
			t.Fatal("checkpoint falsely covered source a with source b's terminal task")
		}
		requireNoProjectionCutoverWrites(t, db, events, works)
	})
	t.Run("blank_source_with_direct_terminal_owner_needs_no_backfill", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		e := rangeProofEvent("finished", 4)
		e.identity.SourceLaneKey = ""
		insertProjectionCutoverEvent(t, db, e)
		insertProjectionCutoverWork(t, db, rangeProofWork("finished", e, "processed"))
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
		if _, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, e.identity.InstalledPackID); err != nil {
			t.Fatalf("direct terminal responsibility is unambiguous: %v", err)
		}
		requireProjectionHistoryUnchanged(t, db, events, works)
	})
	t.Run("blank_source_does_not_hide_a_live_ordinary_task", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		e := rangeProofEvent("live-ordinary", 4)
		e.identity.SourceLaneKey = ""
		insertProjectionCutoverEvent(t, db, e)
		w := rangeProofWork("live-ordinary", e, "pending")
		w.id, w.key = "cwi:live-ordinary", ""
		insertProjectionCutoverWork(t, db, w)
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
		if _, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, e.identity.InstalledPackID); err == nil {
			t.Fatal("blank source hid live work on a settled projection route")
		}
		requireNoProjectionCutoverWrites(t, db, events, works)
	})
}

func TestProjectionRangeCutoverExecutionAuthorityTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	for _, scenario := range []string{"live_event_terminal_mirror", "indirect_input_terminal_mirror", "closed_event_terminal_mirror", "attempted_terminal_work", "duplicate_live_first", "duplicate_terminal_first"} {
		t.Run(scenario, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			e := rangeProofEvent("head", 2)
			insertProjectionCutoverEvent(t, db, e)
			insertProjectionCutoverWork(t, db, rangeProofWork("head", e, "processed"))
			projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET ExecutionOwner='event',Status='pending'`)
			wantOK := false
			switch scenario {
			case "indirect_input_terminal_mirror":
				insertProjectionCutoverEvent(t, db, rangeProofEvent("indirect", 1))
			case "closed_event_terminal_mirror":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='processed'`)
				wantOK = true
			case "attempted_terminal_work":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET Attempt=1`)
				wantOK = true
			case "duplicate_live_first", "duplicate_terminal_first":
				// The schema permits both event-backed WorkTypes for one Event.
				// Retained type-split history must not depend on WorkId ordering.
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorInboundRoute SET IsDeleted=1`)
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET ExecutionOwner='work_item'`)
				projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorInboundRoute`)
				w := rangeProofWork("duplicate", e, "pending")
				w.id, w.key, w.identity.WorkType = "aaa:duplicate", "", wq.WorkTypeWorkflowActionApply
				if scenario == "duplicate_terminal_first" {
					w.id = "zzz:duplicate"
				}
				insertProjectionCutoverWork(t, db, w)
			}
			events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			_, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, e.identity.InstalledPackID)
			if wantOK {
				if err != nil {
					t.Fatalf("proven terminal authority was rejected: %v", err)
				}
				requireProjectionHistoryUnchanged(t, db, events, works)
			} else {
				if err == nil {
					t.Fatal("checkpoint accepted ambiguous or non-authoritative work")
				}
				requireNoProjectionCutoverWrites(t, db, events, works)
			}
		})
	}
}

func TestProjectionRangeCutoverRetiredHistoryTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	for _, scenario := range []string{"legacy_cutoff", "native_terminal_range", "missing_protocol", "invalid_protocol", "negative_ordinal", "beyond_range", "before_range", "wrong_flow", "wrong_key", "wrong_head", "terminal_mirror", "live_work_below_cutoff"} {
		t.Run(scenario, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			a, b := rangeProofEvent("coalesced", 1), rangeProofEvent("head", 2)
			insertProjectionCutoverEvent(t, db, a)
			insertProjectionCutoverEvent(t, db, b)
			projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='pending',ExecutionOwner='event' WHERE Id=?`, a.id)
			insertProjectionCutoverWork(t, db, rangeProofWork("head", b, "processed"))
			projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=1`)
			projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,1,0)`, a.identity.InstalledPackID)
			projectionCutoverExec(t, db, `DELETE FROM _System_InstalledPack`)
			wantOK := scenario == "legacy_cutoff" || scenario == "native_terminal_range"
			switch scenario {
			case "legacy_cutoff":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorProjectionProtocol SET LegacyMaxOrdinal=2`)
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=NULL`)
			case "missing_protocol":
				projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorProjectionProtocol`)
			case "invalid_protocol":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorProjectionProtocol SET ProtocolVersion=99`)
			case "negative_ordinal":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET EnqueueOrdinal=-1 WHERE Id=?`, a.id)
			case "beyond_range":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET EnqueueOrdinal=3 WHERE Id=?`, a.id)
			case "before_range":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=2`)
			case "wrong_flow":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET CompanionFlowVersionId='another-flow' WHERE Id=?`, b.id)
			case "wrong_key":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET ProjectionKey='not-the-canonical-key'`)
			case "wrong_head":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET LaneOrdinal=3`)
			case "terminal_mirror":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET ExecutionOwner='event' WHERE Id=?`, b.id)
			case "live_work_below_cutoff":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorProjectionProtocol SET LegacyMaxOrdinal=2`)
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET State='pending'`)
			}
			events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			err := VerifyConnectorExecutionProtocols(t.Context(), db)
			if (err == nil) != wantOK {
				t.Fatalf("retired history acceptance=%v want %v; err=%v", err == nil, wantOK, err)
			}
			requireProjectionHistoryUnchanged(t, db, events, works)
		})
	}
}

// An ordinary authoritative WorkItem closes only its own historical Event;
// it cannot certify another input on the same settled projection route.
func TestProjectionRangeCutoverOrdinaryCompletionTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	for _, scenario := range []string{"direct_terminal", "pending_ordinary", "terminal_mirror", "uncovered_sibling"} {
		t.Run(scenario, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			projectionCutoverExec(t, db, `UPDATE _System_InstalledPack SET Category='connector-pack'`)
			e := rangeProofEvent("direct-head", 2)
			insertProjectionCutoverEvent(t, db, e)
			projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='pending',ExecutionOwner='event' WHERE Id=?`, e.id)
			now := time.Now().UTC().Truncate(time.Second)
			repo := wq.NewRepository(db, nil, nil)
			work, created, err := repo.CreateWorkItem(t.Context(), wq.CreateWorkItemInput{
				WorkID: "cwi:direct-head", EventID: e.id, InstalledPackID: e.identity.InstalledPackID,
				WorkType: e.identity.WorkType, QueueClass: "sync", State: wq.WorkStatePending,
				SourceLaneKey: e.identity.SourceLaneKey, LaneOrdinal: e.ordinal, RunAfter: now, Now: now,
			})
			if err != nil || !created || work == nil {
				t.Fatalf("create ordinary work: %v/%v/%v", work, created, err)
			}
			if scenario == "direct_terminal" || scenario == "uncovered_sibling" {
				claimed, err := repo.ClaimWorkItem(t.Context(), work.WorkID, "worker", "lease", now.Add(time.Minute), now)
				if err != nil || !claimed {
					t.Fatalf("claim=%v/%v", claimed, err)
				}
				completed, err := repo.CompleteWorkItem(t.Context(), work.WorkID, "lease", wq.WorkStateProcessed, now.Add(time.Second))
				if err != nil || !completed {
					t.Fatalf("complete=%v/%v", completed, err)
				}
			} else if scenario == "terminal_mirror" {
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET State='processed' WHERE WorkId=?`, work.WorkID)
			}
			if scenario == "uncovered_sibling" {
				sibling := rangeProofEvent("uncovered-input", 1)
				insertProjectionCutoverEvent(t, db, sibling)
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Status='pending',ExecutionOwner='event' WHERE Id=?`, sibling.id)
			}
			eventsBefore := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			worksBefore := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			result, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, e.identity.InstalledPackID)
			wantOK := scenario == "direct_terminal"
			if (err == nil) != wantOK {
				t.Fatalf("cutover=%+v/%v want success=%v", result, err, wantOK)
			}
			t.Logf("scenario=%s cutoff=%d error=%v", scenario, result.LegacyMaxOrdinal, err)
			if wantOK {
				if result.LegacyMaxOrdinal != 2 {
					t.Fatalf("cutoff=%d", result.LegacyMaxOrdinal)
				}
				requireProjectionHistoryUnchanged(t, db, eventsBefore, worksBefore)
			} else {
				requireNoProjectionCutoverWrites(t, db, eventsBefore, worksBefore)
			}
		})
	}
}