//go:build integration

package controlplane

import (
	"context"
	"database/sql"
	"encoding/json"
	"strings"
	"testing"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	connectorevents "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// Source resolution is the boundary. Direct reapply and offline source deltas
// share the real revision transition exercised here, including its row lock.
func TestProjectionRangeCutoverSourceRevisionActivationTiDB(t *testing.T) {
	for _, scenario := range []string{"existing_empty_checkpoint", "empty_without_checkpoint", "legacy_requires_checkpoint", "native_history_survives_second_revision", "unclaimed_projection_blocks_revision"} {
		t.Run(scenario, func(t *testing.T) {
			cfg, cleanup := newTemporaryIntegrationDatabase(t)
			t.Cleanup(cleanup)
			ctx, cancel := context.WithTimeout(t.Context(), 3*time.Minute)
			t.Cleanup(cancel)
			schemaConn := openIntegrationConnection(t, cfg)
			connectorfixture.CreateSystemTables(t, schemaConn.DB(), "_System_ConnectorEvent", "_System_ConnectorWorkItem", "_System_ConnectorProjectionProtocol")
			_ = schemaConn.Close()
			conn := initializeControlPlaneMetadata(t, ctx, cfg)
			t.Cleanup(func() { _ = conn.Close() })
			db := conn.DB()
			const name, oldID = "range-activation", "range-activation-old"
			resolve := func(revision string) *packs.ResolvedPackSource {
				source := testControlplaneOfficialRepoEmptyResolvedSourceNamed(name, revision)
				source.Manifest.Category = packs.ConnectorPack
				return source
			}
			first, err := packs.NewSourceBackedPack(resolve("revision-1"))
			if err != nil {
				t.Fatal(err)
			}
			hash, err := packs.PackPayloadHash(first)
			if err != nil {
				t.Fatal(err)
			}
			applied := appliedAppPack{pack: first, payloadHash: hash}
			seedActivatedInstalledPackCompatibility(t, ctx, db, applied, oldID)
			if err := publishFirstSourceBackedPackGeneration(ctx, db, applied); err != nil {
				t.Fatal(err)
			}
			projectionCutoverExec(t, db, `INSERT INTO _System_Migration (Id,Version,Description) VALUES (?,?,?)`, migrationIDVerifyConnectorWorkItemOnlyExecutionCutover, migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover, migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover)
			wantOldCutoff, wantOldFound := int64(9), true
			if scenario == "unclaimed_projection_blocks_revision" {
				enqueueUnclaimedProjectionAfterLegacyCheckpoint(t, ctx, db, name, oldID)
				wantOldCutoff = 0
			} else if scenario == "existing_empty_checkpoint" {
				projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,1,9)`, oldID)
			} else if scenario == "empty_without_checkpoint" {
				wantOldCutoff, wantOldFound = 0, false
			} else {
				a, b := rangeProofEvent("coalesced", 1), rangeProofEvent("head", 2)
				a.identity.InstalledPackID, b.identity.InstalledPackID = oldID, oldID
				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 Attempt=1`)
				projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorIngressControl (Id,Namespace,PackInstallId,Channel,Mode) VALUES ('fence',?,'','*','paused')`, name)
				wantOldCutoff = 2
				if scenario == "native_history_survives_second_revision" {
					wantOldCutoff = 0
					projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET InputStartOrdinal=1`)
					projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,1,0)`, oldID)
				}
			}
			target, err := packs.NewSourceBackedPack(resolve("revision-2"))
			if err != nil {
				t.Fatal(err)
			}
			events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			if scenario == "unclaimed_projection_blocks_revision" {
				err := applyAllSourceRevisionTestPlan(t, ctx, conn, target, "revision-1", "unclaimed projection must prevent revision", resolve)
				if err == nil || !strings.Contains(err.Error(), "not drained") {
					t.Fatalf("revision must reject unclaimed projection work: %v", err)
				}
				var retainedID, retainedStatus string
				if err := db.QueryRowContext(ctx, `SELECT PackInstallId,Status FROM _System_InstalledPack WHERE Name=?`, name).Scan(&retainedID, &retainedStatus); err != nil {
					t.Fatal(err)
				}
				if retainedID != oldID || retainedStatus != "active" {
					t.Fatalf("rejected revision changed incarnation: %s/%s", retainedID, retainedStatus)
				}
				if cutoff, found, err := readProjectionProtocol(ctx, db, oldID); err != nil || !found || cutoff != 0 {
					t.Fatalf("rejected revision changed checkpoint: %d/%v/%v", cutoff, found, err)
				}
				requireProjectionHistoryUnchanged(t, db, events, works)
				return
			}
			if scenario == "legacy_requires_checkpoint" {
				err := applyAllSourceRevisionTestPlan(t, ctx, conn, target, "revision-1", "missing projection checkpoint proof", resolve)
				if err == nil || !strings.Contains(err.Error(), "projection") {
					t.Fatalf("revision replaced old history without checkpoint: %v", err)
				}
				var retainedID, retainedStatus string
				if err := db.QueryRowContext(ctx, `SELECT PackInstallId,Status FROM _System_InstalledPack WHERE Name=?`, name).Scan(&retainedID, &retainedStatus); err != nil {
					t.Fatal(err)
				}
				if retainedID != oldID || retainedStatus != "active" {
					t.Fatalf("rejected revision changed incarnation: %s/%s", retainedID, retainedStatus)
				}
				requireNoProjectionCutoverWrites(t, db, events, works)
				// Only the explicit, already-required fenced cutover may write the receipt.
				if _, err := EnsureConnectorProjectionRangeProtocol(ctx, db, oldID); err != nil {
					t.Fatal(err)
				}
			}
			runApplyAllSourceRevisionTestPlan(t, ctx, conn, target, "revision-1", "projection protocol activation proof", resolve)
			var newID string
			if err := db.QueryRowContext(ctx, `SELECT PackInstallId FROM _System_InstalledPack WHERE Name=?`, name).Scan(&newID); err != nil {
				t.Fatal(err)
			}
			if newID == oldID {
				t.Fatal("revision advance did not create a new incarnation")
			}
			if cutoff, found, err := readProjectionProtocol(ctx, db, newID); err != nil || !found || cutoff != 0 {
				t.Fatalf("new active incarnation lacks protocol: %d/%v/%v", cutoff, found, err)
			}
			if cutoff, found, err := readProjectionProtocol(ctx, db, oldID); err != nil || found != wantOldFound || cutoff != wantOldCutoff {
				t.Fatalf("revision changed historical checkpoint: %d/%v/%v", cutoff, found, err)
			}
			if err := VerifyConnectorExecutionProtocols(ctx, db); err != nil {
				t.Fatalf("completed revision made retained history fail the release gate: %v", err)
			}
			requireProjectionHistoryUnchanged(t, db, events, works)
		})
	}
}

// A legacy zero-ordinal head and a later writer with an earlier clock must not
// hide newly accepted native work during the pre-first-claim window.
func enqueueUnclaimedProjectionAfterLegacyCheckpoint(t *testing.T, ctx context.Context, db *sql.DB, namespace, packID string) {
	t.Helper()
	anchor := time.Now().UTC().Truncate(time.Second)
	old := rangeProofEvent("old-zero-head", 0)
	old.identity.InstalledPackID = packID
	// The materialized source lane of the real enqueue payload below.
	old.identity.SourceLaneKey = "35537cd827e79a5332064f97605b4fa37551de23c21a16c9c3abf2437c32e381"
	insertProjectionCutoverEvent(t, db, old)
	projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET Namespace=?,CreatedDate=?,Status='skipped',ExecutionOwner='work_item' WHERE Id=?`, namespace, anchor, old.id)
	insertProjectionCutoverWork(t, db, rangeProofWork("old-zero-head", old, "processed"))
	projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET Attempt=1 WHERE EventId=?`, old.id)
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorInboundRoute (Id,Namespace,InstalledPackId,EventKind,Version,FlowDeveloperName,InvocationMode,IsCurrent) VALUES ('route',?,?,?,1,'ApplyRecord','latest_by_source_full_record_preserve_payload_settled',1)`, namespace, packID, old.identity.EventKind)
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorIngressControl (Id,Namespace,PackInstallId,Channel,Mode) VALUES ('fence',?,'','*','paused')`, namespace)
	if _, err := EnsureConnectorProjectionRangeProtocol(ctx, db, packID); err != nil {
		t.Fatal(err)
	}
	if _, err := admission.NewSQLGate(db).TransitionNamespaceFence(ctx, admission.NamespaceFenceTransition{Namespace: namespace, TargetMode: admission.ModeActive}); err != nil {
		t.Fatal(err)
	}
	svc := connectorevents.NewService(db, nil, func() time.Time { return anchor.Add(-time.Hour) })
	ids, duplicates, err := svc.EnqueueBatch(ctx, namespace, packID, "new-receipt", []connectorevents.BatchEvent{{EventKind: old.identity.EventKind, IdempotencyKey: "new-input", RouteVersion: 1, Payload: json.RawMessage(`{"objectKind":"record","sourceID":"legacy-source"}`)}})
	if err != nil || len(ids) != 1 || len(duplicates) != 1 || duplicates[0] {
		t.Fatalf("enqueue=%v/%v/%v", ids, duplicates, err)
	}
	var state, owner, sourceLane string
	var attempt, startOrdinal, ordinal int64
	if err := db.QueryRowContext(ctx, `SELECT w.State,w.Attempt,w.InputStartOrdinal,e.EnqueueOrdinal,e.ExecutionOwner,e.SourceLaneKey FROM _System_ConnectorWorkItem w JOIN _System_ConnectorEvent e ON e.Id=w.EventId WHERE e.Id=?`, ids[0]).Scan(&state, &attempt, &startOrdinal, &ordinal, &owner, &sourceLane); err != nil {
		t.Fatal(err)
	}
	if attempt != 0 || startOrdinal <= 0 || ordinal <= 0 || owner != "event" || (state != "pending" && state != "blocked") || sourceLane != old.identity.SourceLaneKey {
		t.Fatalf("expected unclaimed native successor: state=%s attempt=%d start=%d ordinal=%d owner=%s lane=%s", state, attempt, startOrdinal, ordinal, owner, sourceLane)
	}
	// Bind both timestamps through the same connection and verify their stored
	// order; mixing UTC_TIMESTAMP with driver-local times can mask this case.
	var clockReversed bool
	if err := db.QueryRowContext(ctx, `SELECT newer.CreatedDate < older.CreatedDate FROM _System_ConnectorEvent newer JOIN _System_ConnectorEvent older ON older.Id=? WHERE newer.Id=?`, old.id, ids[0]).Scan(&clockReversed); err != nil || !clockReversed {
		t.Fatalf("clock-order precondition: %v/%v", clockReversed, err)
	}
	if err := VerifyConnectorExecutionProtocols(ctx, db); err != nil {
		t.Fatalf("pre-revision protocol gate: %v", err)
	}
}