package controlplane

import (
	"context"
	"database/sql"
	"encoding/json"
	admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"os"
	"reflect"
	"strings"
	"testing"
	"time"

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

const rangeCutoverNamespace = "range-proof"

func projectionCutoverExec(t *testing.T, db *sql.DB, query string, args ...any) {
	t.Helper()
	if _, err := db.Exec(query, args...); err != nil {
		t.Fatalf("fixture statement failed: %v", err)
	}
}

func projectionCutoverDatabase(t *testing.T) *sql.DB {
	t.Helper()
	if os.Getenv("TIDB_HOST") == "" && os.Getenv("TIDB_PORT") == "" {
		t.Skip("set TIDB_HOST or TIDB_PORT to run the canonical TiDB cutover proof")
	}
	// A configured integration run must never report unavailable TiDB as proof.
	t.Cleanup(func() {
		if t.Skipped() {
			t.Error("configured projection cutover TiDB proof must execute, not skip")
		}
	})
	db, cleanup := newTemporaryConnectorProjectionSnapshotDatabase(t)
	t.Cleanup(cleanup)
	connectorfixture.CreateSystemTables(t, db, "_System_InstalledPack", "_System_Migration", "_System_ConnectorEvent", "_System_ConnectorWorkItem", "_System_ConnectorInboundRoute", "_System_ConnectorIngressControl", "_System_ConnectorProjectionProtocol")
	return db
}

func resetProjectionCutoverFixture(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, table := range []string{"_System_ConnectorProjectionProtocol", "_System_ConnectorWorkItem", "_System_ConnectorEvent", "_System_ConnectorInboundRoute", "_System_ConnectorIngressControl", "_System_InstalledPack", "_System_Migration"} {
		projectionCutoverExec(t, db, "DELETE FROM "+table)
	}
	pack := rangeProofIdentity().InstalledPackID
	projectionCutoverExec(t, db, `INSERT INTO _System_InstalledPack (Id,Name,PackInstallId,Category,Status) VALUES (?,?,?,'connector','active')`, "pack-row", rangeCutoverNamespace, pack)
	projectionCutoverExec(t, db, `INSERT INTO _System_Migration (Id,Version,Description) VALUES (?,?,?)`, migrationIDVerifyConnectorWorkItemOnlyExecutionCutover, migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover, migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover)
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorInboundRoute (Id,Namespace,InstalledPackId,EventKind,Version,FlowDeveloperName,InvocationMode) VALUES (?,?,?,?,1,'ApplyRecord','latest_by_source_full_record_preserve_payload_settled')`, "route", rangeCutoverNamespace, pack, rangeProofIdentity().EventKind)
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorIngressControl (Id,Namespace,PackInstallId,Channel,Mode,ExpiresAt) VALUES (?,?,'','*','paused',NULL)`, "fence", rangeCutoverNamespace)
}

func insertProjectionCutoverEvent(t *testing.T, db *sql.DB, e *projectionCutoverEvent) {
	t.Helper()
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorEvent (Id,Namespace,InstalledPackId,EventKind,IdempotencyKey,RouteVersion,SourceLaneKey,EnqueueOrdinal,ExecutionOwner,Status,ReplayOfId,Payload) VALUES (?,?,?,?,?,1,?,? ,?,'skipped',?,?)`, e.id, rangeCutoverNamespace, e.identity.InstalledPackID, e.identity.EventKind, e.id, e.identity.SourceLaneKey, e.ordinal, wq.EventExecutionOwnerWorkItem, nil, `{"fixture":"retained"}`)
}

func nullProjectionString(value string) any {
	if value == "" {
		return nil
	}
	return value
}

func insertProjectionCutoverWork(t *testing.T, db *sql.DB, w *projectionCutoverWork) {
	t.Helper()
	projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorWorkItem (WorkId,EventId,InstalledPackId,WorkType,State,SourceLaneKey,LaneOrdinal,ProjectionKey,ProjectionActiveKey) VALUES (?,?,?,?,?,?,?,?,NULL)`, w.id, w.eventID, w.identity.InstalledPackID, w.identity.WorkType, w.state, w.identity.SourceLaneKey, w.ordinal, nullProjectionString(w.key))
}

func projectionCutoverRows(t *testing.T, db *sql.DB, query string) [][]string {
	t.Helper()
	rows, err := db.Query(query)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var result [][]string
	for rows.Next() {
		values := make([]sql.RawBytes, len(columns))
		dest := make([]any, len(columns))
		for i := range values {
			dest[i] = &values[i]
		}
		if err := rows.Scan(dest...); err != nil {
			t.Fatal(err)
		}
		line := make([]string, len(columns))
		for i, v := range values {
			if v == nil {
				line[i] = "<NULL>"
			} else {
				line[i] = string(v)
			}
		}
		result = append(result, line)
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	return result
}

func requireNoProjectionCutoverWrites(t *testing.T, db *sql.DB, events, works [][]string) {
	t.Helper()
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`); !reflect.DeepEqual(got, events) {
		t.Fatalf("rejected cutover changed Event facts: before=%v after=%v", events, got)
	}
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`); !reflect.DeepEqual(got, works) {
		t.Fatalf("rejected cutover changed WorkItems: before=%v after=%v", works, got)
	}
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorProjectionProtocol`); len(got) != 0 {
		t.Fatalf("rejected cutover issued protocol: %v", got)
	}
}

func TestProjectionRangeCutoverTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	t.Run("completed_history_is_unchanged_and_checkpoint_is_immutable", 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"))
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		got, err := EnsureConnectorProjectionRangeProtocol(ctx, db, b.identity.InstalledPackID)
		if err != nil || got.AlreadyApplied || got.LegacyMaxOrdinal != 2 {
			t.Fatalf("checkpoint=%+v err=%v", got, err)
		}
		requireProjectionHistoryUnchanged(t, db, events, works)
		if rows := projectionCutoverRows(t, db, `SELECT InputStartOrdinal FROM _System_ConnectorWorkItem`); !reflect.DeepEqual(rows, [][]string{{"<NULL>"}}) {
			t.Fatalf("old range was rewritten: %v", rows)
		}
		// Native inputs after the boundary must never be reclassified as legacy.
		insertProjectionCutoverEvent(t, db, rangeProofEvent("later", 3))
		got, err = EnsureConnectorProjectionRangeProtocol(ctx, db, b.identity.InstalledPackID)
		if err != nil || !got.AlreadyApplied || got.LegacyMaxOrdinal != 2 {
			t.Fatalf("repeat advanced checkpoint: %+v/%v", got, err)
		}
	})
	for _, state := range []string{"pending", "blocked", "retry", "processing"} {
		t.Run("reject_undrained_"+state, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			e := rangeProofEvent("old-live", 1)
			insertProjectionCutoverEvent(t, db, e)
			insertProjectionCutoverWork(t, db, rangeProofWork("old-live", e, state))
			projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET Attempt=2,LeaseOwner='old-worker',LeaseToken='retained',LeaseUntil=NULL`)
			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 || !strings.Contains(err.Error(), "drain") {
				t.Fatalf("live work was accepted: %v", err)
			}
			requireNoProjectionCutoverWrites(t, db, events, works)
		})
	}
	for _, scenario := range []string{"sole_missing_task", "positive_after_last_terminal", "mixed_zero_indirect_input", "identity_mismatch", "work_without_event"} {
		t.Run(scenario, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			a, b := rangeProofEvent("unresolved", 1), rangeProofEvent("head", 2)
			if scenario == "positive_after_last_terminal" {
				a.ordinal = 3
			}
			if scenario == "mixed_zero_indirect_input" {
				a.ordinal = 0
			}
			insertProjectionCutoverEvent(t, db, a)
			if scenario != "sole_missing_task" {
				insertProjectionCutoverEvent(t, db, b)
				insertProjectionCutoverWork(t, db, rangeProofWork("head", b, "processed"))
			}
			if scenario == "identity_mismatch" {
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorWorkItem SET LaneOrdinal=99`)
			}
			if scenario == "work_without_event" {
				projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorEvent WHERE Id=?`, b.id)
			}
			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("unresolved history was accepted")
			}
			requireNoProjectionCutoverWrites(t, db, events, works)
		})
	}
	for _, scenario := range []string{"missing", "active", "expiring", "active_override"} {
		t.Run("fence_"+scenario, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			switch scenario {
			case "missing":
				projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorIngressControl`)
			case "active":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorIngressControl SET Mode='active'`)
			case "expiring":
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorIngressControl SET ExpiresAt=DATE_ADD(UTC_TIMESTAMP(),INTERVAL 1 HOUR)`)
			case "active_override":
				projectionCutoverExec(t, db, `INSERT INTO _System_ConnectorIngressControl (Id,Namespace,PackInstallId,Channel,Mode) VALUES ('override',?,?,'webhook','active')`, rangeCutoverNamespace, rangeProofIdentity().InstalledPackID)
			}
			if _, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, rangeProofIdentity().InstalledPackID); err == nil {
				t.Fatal("invalid fence was accepted")
			}
			requireNoProjectionCutoverWrites(t, db, nil, nil)
		})
	}
	t.Run("concurrent_callers_write_one_checkpoint", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		e := rangeProofEvent("head", 7)
		insertProjectionCutoverEvent(t, db, e)
		insertProjectionCutoverWork(t, db, rangeProofWork("head", e, "dead"))
		type outcome struct {
			result ProjectionRangeCutoverResult
			err    error
		}
		results := make(chan outcome, 2)
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		for i := 0; i < 2; i++ {
			go func() {
				r, err := EnsureConnectorProjectionRangeProtocol(ctx, db, e.identity.InstalledPackID)
				results <- outcome{r, err}
			}()
		}
		applied := 0
		for i := 0; i < 2; i++ {
			out := <-results
			if out.err != nil || out.result.LegacyMaxOrdinal != 7 {
				t.Fatalf("concurrent checkpoint=%+v/%v", out.result, out.err)
			}
			if !out.result.AlreadyApplied {
				applied++
			}
		}
		if applied != 1 {
			t.Fatalf("checkpoint applied %d times", applied)
		}
	})
}

func requireProjectionHistoryUnchanged(t *testing.T, db *sql.DB, events, works [][]string) {
	t.Helper()
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`); !reflect.DeepEqual(got, events) {
		t.Fatalf("cutover changed Events: before=%v after=%v", events, got)
	}
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`); !reflect.DeepEqual(got, works) {
		t.Fatalf("cutover changed WorkItems: before=%v after=%v", works, got)
	}
}

func TestProjectionRangeCutoverGreenfieldTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	t.Run("protocol_is_owned_by_activation_transaction", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if err := ensureEmptyConnectorProjectionProtocolTx(ctx, tx, rangeCutoverNamespace); err != nil {
			t.Fatal(err)
		}
		if cutoff, found, err := readProjectionProtocol(ctx, tx, rangeProofIdentity().InstalledPackID); err != nil || !found || cutoff != 0 {
			t.Fatalf("activation must issue cutoff0: cutoff=%d found=%v err=%v", cutoff, found, err)
		}
		if err := tx.Rollback(); err != nil {
			t.Fatal(err)
		}
		if rows := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorProjectionProtocol`); len(rows) != 0 {
			t.Fatalf("failed activation left protocol=%v", rows)
		}
		tx, err = db.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if err := ensureEmptyConnectorProjectionProtocolTx(ctx, tx, rangeCutoverNamespace); err != nil {
			t.Fatal(err)
		}
		if err := tx.Commit(); err != nil {
			t.Fatal(err)
		}
		insertProjectionCutoverEvent(t, db, rangeProofEvent("post-activation", 1))
		tx, err = db.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if err := ensureEmptyConnectorProjectionProtocolTx(ctx, tx, rangeCutoverNamespace); err != nil {
			t.Fatal(err)
		}
		if err := tx.Commit(); err != nil {
			t.Fatal(err)
		}
		if cutoff, found, err := readProjectionProtocol(ctx, db, rangeProofIdentity().InstalledPackID); err != nil || !found || cutoff != 0 {
			t.Fatalf("repeat activation must preserve cutoff0: cutoff=%d found=%v err=%v", cutoff, found, err)
		}
	})
	for _, kind := range []string{"event", "workitem"} {
		t.Run("occupied_"+kind+"_requires_explicit_cutover", func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			e := rangeProofEvent("occupied", 1)
			if kind == "event" {
				insertProjectionCutoverEvent(t, db, e)
			} else {
				insertProjectionCutoverWork(t, db, rangeProofWork("occupied", e, "pending"))
			}
			events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
			defer cancel()
			tx, err := db.BeginTx(ctx, nil)
			if err != nil {
				t.Fatal(err)
			}
			defer tx.Rollback()
			if err := ensureEmptyConnectorProjectionProtocolTx(ctx, tx, rangeCutoverNamespace); err == nil || !strings.Contains(err.Error(), "explicit fenced") {
				t.Fatalf("occupied activation must require cutover: %v", err)
			}
			if err := tx.Rollback(); err != nil {
				t.Fatal(err)
			}
			requireNoProjectionCutoverWrites(t, db, events, works)
		})
	}
}

func TestProjectionRangeCutoverReleaseGateTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	for _, kind := range []string{"event", "workitem"} {
		t.Run("missing_handoff_does_not_retire_"+kind, func(t *testing.T) {
			resetProjectionCutoverFixture(t, db)
			projectionCutoverExec(t, db, `DELETE FROM _System_Migration`)
			e := rangeProofEvent("legacy-backlog", 0)
			if kind == "event" {
				insertProjectionCutoverEvent(t, db, e)
				projectionCutoverExec(t, db, `UPDATE _System_ConnectorEvent SET ExecutionOwner='event',Status='pending'`)
			} else {
				insertProjectionCutoverWork(t, db, rangeProofWork("legacy-backlog", e, "pending"))
			}
			events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
			works := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
			defer cancel()
			if err := VerifyConnectorExecutionProtocols(ctx, db); err == nil {
				t.Fatalf("release gate must reject old backlog: %v", err)
			}
			requireNoProjectionCutoverWrites(t, db, events, works)
			if rows := projectionCutoverRows(t, db, `SELECT * FROM _System_Migration`); len(rows) != 0 {
				t.Fatalf("gate manufactured handoff receipt: %v", rows)
			}
		})
	}
	t.Run("existing_handoff_does_not_hide_missing_route_orphan", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		e := rangeProofEvent("missing-route-orphan", 7)
		insertProjectionCutoverEvent(t, db, e)
		projectionCutoverExec(t, db, `DELETE FROM _System_ConnectorInboundRoute`)
		events := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		if err := VerifyConnectorExecutionProtocols(ctx, db); err == nil || !strings.Contains(err.Error(), e.id) {
			t.Fatalf("release gate bypassed missing-route ownership proof: %v", err)
		}
		requireNoProjectionCutoverWrites(t, db, events, nil)
	})
	t.Run("empty_database_can_establish_both_protocols", func(t *testing.T) {
		resetProjectionCutoverFixture(t, db)
		projectionCutoverExec(t, db, `DELETE FROM _System_Migration`)
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		if err := VerifyConnectorExecutionProtocols(ctx, db); err != nil {
			t.Fatalf("greenfield release gate: %v", err)
		}
		if applied, err := workItemCutoverMigrationApplied(ctx, db); err != nil || !applied {
			t.Fatalf("greenfield handoff=%v err=%v", applied, err)
		}
		if cutoff, found, err := readProjectionProtocol(ctx, db, rangeProofIdentity().InstalledPackID); err != nil || !found || cutoff != 0 {
			t.Fatalf("greenfield protocol cutoff=%d found=%v err=%v", cutoff, found, err)
		}
	})
}

// Retained rows use the positive-ordinal, coalesced shape of Connectors #329
// (0e7d07939c98fb14b5a7bee13c08fafb88331956), before InputStartOrdinal existed.
// Cutover, subsequent producer, range reader and completion are production code.
func TestProjectionRangeCutoverThenNativeExecutionTiDB(t *testing.T) {
	db := projectionCutoverDatabase(t)
	resetProjectionCutoverFixture(t, db)
	a, b := rangeProofEvent("old-first", 1), rangeProofEvent("old-second", 2)
	a.identity.SourceLaneKey = "35537cd827e79a5332064f97605b4fa37551de23c21a16c9c3abf2437c32e381"
	b.identity.SourceLaneKey = a.identity.SourceLaneKey
	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("old-second", b, "processed"))
	oldEvents := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent ORDER BY Id`)
	oldWorks := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem ORDER BY WorkId`)
	if _, err := EnsureConnectorProjectionRangeProtocol(t.Context(), db, b.identity.InstalledPackID); err != nil {
		t.Fatal(err)
	}
	requireProjectionHistoryUnchanged(t, db, oldEvents, oldWorks)
	svc := events.NewService(db, nil, func() time.Time { return time.Now().UTC().Add(-time.Hour) })
	if err := events.RequireSourceProjectionProtocol(t.Context(), db); err != nil {
		t.Fatal(err)
	}
	if _, err := admission.NewSQLGate(db).TransitionNamespaceFence(t.Context(), admission.NamespaceFenceTransition{Namespace: rangeCutoverNamespace, TargetMode: admission.ModeActive}); err != nil {
		t.Fatal(err)
	}
	batch := []events.BatchEvent{}
	for _, key := range []string{"old-first", "new-first", "old-second", "new-second"} {
		batch = append(batch, events.BatchEvent{EventKind: b.identity.EventKind, IdempotencyKey: key, RouteVersion: 1, Payload: json.RawMessage(`{"objectKind":"record","sourceID":"legacy-source"}`)})
	}
	ids, dups, err := svc.EnqueueBatch(t.Context(), rangeCutoverNamespace, b.identity.InstalledPackID, "mixed-receipt-retry", batch)
	if err != nil || !reflect.DeepEqual(dups, []bool{true, false, true, false}) {
		t.Fatalf("mixed batch=%v/%v", dups, err)
	}
	now := time.Now().UTC()
	work, claimed, err := svc.WorkItemRepository().ClaimNextWorkItem(wq.WithDistributedClaim(t.Context()), wq.ClaimWorkItemInput{InstalledPackID: b.identity.InstalledPackID, Generation: 0, OwnerID: "checkpoint-worker", FencingToken: "checkpoint-fence", LeaseToken: "checkpoint-lease", LeaseUntil: now.Add(time.Minute), Now: now})
	if err != nil || !claimed || work == nil {
		t.Fatalf("native claim=%+v/%v/%v", work, claimed, err)
	}
	payloads, err := svc.LoadSourceProjectionPayloads(t.Context(), work.WorkID)
	if err != nil || len(payloads) != 2 || payloads[0].ID != ids[1] || payloads[1].ID != ids[3] {
		t.Fatalf("native inputs=%+v/%v want %v", payloads, err, ids)
	}
	if done, err := svc.CompleteWorkItem(t.Context(), work.WorkID, work.LeaseToken, wq.WorkStateProcessed, now); err != nil || !done {
		t.Fatalf("native completion=%v/%v", done, err)
	}
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorEvent WHERE Id IN ('old-first','old-second') ORDER BY Id`); !reflect.DeepEqual(got, oldEvents) {
		t.Fatal("native execution rewrote historical Events")
	}
	if got := projectionCutoverRows(t, db, `SELECT * FROM _System_ConnectorWorkItem WHERE InputStartOrdinal IS NULL ORDER BY WorkId`); !reflect.DeepEqual(got, oldWorks) {
		t.Fatal("native execution rewrote historical WorkItems")
	}
}