// Integration tests for the connector event queue against a live TiDB.
// Skipped automatically when TIDB_HOST / TIDB_PORT (default 127.0.0.1:4001)
// is unreachable — they will run in CI where TiDB is provisioned.
package connector_events

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	"github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

// openTiDB opens a temporary TiDB database for a single test run and returns
// a cleanup func that drops the database and closes both connections.  The
// pattern exactly mirrors automation/audit_store_integration_test.go:openTiDB.
func openTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := os.Getenv("TIDB_HOST")
	if host == "" {
		host = "127.0.0.1"
	}
	port := os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4001"
	}
	user := os.Getenv("TIDB_USER")
	if user == "" {
		user = "root"
	}
	pwd := os.Getenv("TIDB_PASSWORD")
	tlsParam := ""
	switch strings.ToLower(strings.TrimSpace(os.Getenv("TIDB_TLS"))) {
	case "1", "true", "yes", "require", "verify-full":
		tlsParam = "&tls=true"
	}

	adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=Local%s", user, pwd, host, port, tlsParam)
	adminDB, err := sql.Open("mysql", adminDSN)
	if err != nil {
		unavailableTiDB(t, "open admin TiDB: %v", err)
	}
	if err := adminDB.Ping(); err != nil {
		_ = adminDB.Close()
		unavailableTiDB(t, "TiDB not reachable at %s:%s: %v", host, port, err)
	}

	dbName := fmt.Sprintf("nexuscrm_connector_events_%d", time.Now().UnixNano())
	if _, err := adminDB.Exec("CREATE DATABASE `" + dbName + "`"); err != nil {
		_ = adminDB.Close()
		unavailableTiDB(t, "cannot create temporary database: %v", err)
	}

	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=Local%s", user, pwd, host, port, dbName, tlsParam)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
		t.Fatalf("open test TiDB connection: %v", err)
	}
	if err := db.Ping(); err != nil {
		_ = db.Close()
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
		t.Fatalf("ping test TiDB connection: %v", err)
	}

	cleanup := func() {
		_ = db.Close()
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
	}
	return db, cleanup
}

// unavailableTiDB keeps local developer runs skippable while making the named
// managed-TiDB CI lane fail hard. P0-C race tests are acceptance evidence only
// when they execute; a skip is never a pass in that lane.
func unavailableTiDB(t *testing.T, format string, args ...any) {
	t.Helper()
	if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
		t.Fatalf("required TiDB integration unavailable: "+format, args...)
	}
	t.Skipf("skipping integration test: "+format, args...)
}

// createConnectorEventTable creates _System_ConnectorEvent and the fake
// _System_InstalledPack table (required by ResolveConnectorPackByNamespace)
// in the temporary test database.  DDL matches system_tables.json schema.
func createConnectorEventTable(t *testing.T, db *sql.DB) {
	t.Helper()
	connectorfixture.CreateSystemTables(t, db, "_System_ConnectorEvent", "_System_ConnectorWorkItem", "_System_ConnectorProjectionProtocol")

	// Every materialization integration fixture uses the production admission
	// authority. The dedicated helper is idempotent so fence-specific tests can
	// still call it explicitly when documenting their setup.
	createConnectorIngressControlTable(t, db)
}

func createConnectorWorkAttemptTable(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.Exec(`CREATE TABLE _System_ConnectorWorkAttempt (
		WorkId             VARCHAR(255) NOT NULL,
		Attempt            INT NOT NULL,
		EventId            VARCHAR(255) NOT NULL,
		FlowExecutionId    VARCHAR(255) NOT NULL,
		OutcomeState       VARCHAR(32) NOT NULL,
		PresentedErrorCode VARCHAR(255) NOT NULL,
		CauseCode          VARCHAR(255) NOT NULL,
		CauseCategory      VARCHAR(64) NOT NULL,
		CauseElementType   VARCHAR(255) NOT NULL,
		CauseElementName   VARCHAR(255) NOT NULL,
		Retriable          BOOLEAN NOT NULL,
		CreatedDate        DATETIME(6) NOT NULL,
		LastModifiedDate   DATETIME(6) NOT NULL,
		UNIQUE KEY uniq_connector_work_attempt (WorkId, Attempt, FlowExecutionId)
	)`); err != nil {
		t.Fatalf("create _System_ConnectorWorkAttempt: %v", err)
	}
}

type connectorWorkAttemptQueryer interface {
	QueryRow(query string, args ...any) *sql.Row
}

func assertConnectorWorkAttempt(
	t *testing.T,
	q connectorWorkAttemptQueryer,
	flowExecutionID, workID, eventID, outcomeState, presentedErrorCode, causeCode, causeCategory, causeElementType, causeElementName string,
	attempt int,
	retriable bool,
) {
	t.Helper()
	var gotWorkID, gotEventID, gotState, gotPresentedCode, gotCauseCode, gotCauseCategory, gotElementType, gotElementName string
	var gotAttempt int
	var gotRetriable bool
	if err := q.QueryRow(`
SELECT WorkId, Attempt, EventId, OutcomeState, PresentedErrorCode, CauseCode, CauseCategory,
       CauseElementType, CauseElementName, Retriable
  FROM _System_ConnectorWorkAttempt
 WHERE FlowExecutionId = ?`, flowExecutionID).Scan(
		&gotWorkID, &gotAttempt, &gotEventID, &gotState, &gotPresentedCode, &gotCauseCode, &gotCauseCategory,
		&gotElementType, &gotElementName, &gotRetriable,
	); err != nil {
		t.Fatalf("load WorkItem attempt %s: %v", flowExecutionID, err)
	}
	if gotWorkID != workID || gotAttempt != attempt || gotEventID != eventID || gotState != outcomeState ||
		gotPresentedCode != presentedErrorCode || gotCauseCode != causeCode || gotCauseCategory != causeCategory ||
		gotElementType != causeElementType || gotElementName != causeElementName || gotRetriable != retriable {
		t.Fatalf("WorkItem attempt %s = %s/%d/%s state=%s presented=%s cause=%s/%s element=%s/%s retriable=%v",
			flowExecutionID, gotWorkID, gotAttempt, gotEventID, gotState, gotPresentedCode, gotCauseCode, gotCauseCategory,
			gotElementType, gotElementName, gotRetriable)
	}
}

func makeSettledProjectionRoute(t *testing.T, db *sql.DB, namespace, eventKind string) {
	t.Helper()
	res, err := db.Exec(`
UPDATE _System_ConnectorInboundRoute
   SET EventKind = ?, InvocationMode = ?, QueueClass = 'bulk'
 WHERE Namespace = ?`, eventKind, connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled, namespace)
	if err != nil {
		t.Fatalf("configure settled projection route: %v", err)
	}
	if rows, err := res.RowsAffected(); err != nil || rows != 1 {
		t.Fatalf("configure settled projection route rows=%d err=%v", rows, err)
	}
}

func TestService_SC400ProjectionAndInteractiveActionsMaterializeBounded(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-scale"
		packInstall = "install-token-projection-scale"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)
	if _, err := db.Exec(`INSERT INTO _System_ConnectorInboundRoute
		(Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName, InvocationMode, QueueClass, IsCurrent, IsDeleted)
		VALUES ('cir-projection-scale-action', ?, ?, 'workflow.action.execute', 1, 'TestActionHandler', 'per_event', 'interactive', 1, 0)`,
		namespace, packInstall); err != nil {
		t.Fatalf("seed interactive action route: %v", err)
	}

	events := make([]BatchEvent, 0, 4400)
	for source := 0; source < 400; source++ {
		for change := 0; change < 10; change++ {
			events = append(events, BatchEvent{
				EventKind:        eventKind,
				IdempotencyKey:   fmt.Sprintf("contact-%03d-change-%02d", source, change),
				Payload:          json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"contact-%03d","property":"p-%02d"}`, source, change)),
				RouteVersion:     1,
				DeliverySequence: len(events),
			})
		}
	}
	for action := 0; action < 400; action++ {
		events = append(events, BatchEvent{
			EventKind:        "workflow.action.execute",
			IdempotencyKey:   fmt.Sprintf("action-%03d", action),
			Payload:          json.RawMessage(fmt.Sprintf(`{"objectKind":"workflow-actions","sourceID":"contact-%03d"}`, action)),
			RouteVersion:     1,
			DeliverySequence: len(events),
		})
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	svc := NewService(db, nil, nil)
	enqueueStarted := time.Now()
	ids, dups, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-scale-delivery", events)
	enqueueElapsed := time.Since(enqueueStarted)
	if err != nil {
		t.Fatalf("enqueue projection scale batch: %v", err)
	}
	if enqueueElapsed > 3*time.Minute {
		t.Fatalf("SC400 materialization took %s, want a bounded result within 3m", enqueueElapsed)
	}
	if len(ids) != 4400 || len(dups) != 4400 {
		t.Fatalf("enqueue result ids=%d dups=%d, want 4400 each", len(ids), len(dups))
	}

	var projectionEventCount, projectionWorkItemCount, actionEventCount, actionWorkItemCount, pendingCount, skippedCount, maxLiveWorkItemsPerSource int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId = ? AND EventKind = ?`, packInstall, eventKind).Scan(&projectionEventCount); err != nil {
		t.Fatalf("count projection events: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND WorkId LIKE 'csp:%'`, packInstall).Scan(&projectionWorkItemCount); err != nil {
		t.Fatalf("count projection work items: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId = ? AND EventKind = 'workflow.action.execute'`, packInstall).Scan(&actionEventCount); err != nil {
		t.Fatalf("count action events: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND WorkType = ?`, packInstall, connector_workqueue.WorkTypeWorkflowActionApply).Scan(&actionWorkItemCount); err != nil {
		t.Fatalf("count action work items: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId = ? AND EventKind = ? AND Status = 'pending'`, packInstall, eventKind).Scan(&pendingCount); err != nil {
		t.Fatalf("count pending events: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId = ? AND EventKind = ? AND Status = 'skipped'`, packInstall, eventKind).Scan(&skippedCount); err != nil {
		t.Fatalf("count skipped events: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT COALESCE(MAX(live_count), 0)
  FROM (
        SELECT SourceLaneKey, COUNT(*) AS live_count
          FROM _System_ConnectorWorkItem
         WHERE InstalledPackId = ? AND State IN ('pending', 'retry', 'processing')
         GROUP BY SourceLaneKey
       ) live`, packInstall).Scan(&maxLiveWorkItemsPerSource); err != nil {
		t.Fatalf("count live WorkItems per source: %v", err)
	}

	if projectionEventCount != 4000 || projectionWorkItemCount != 400 || actionEventCount != 400 || actionWorkItemCount != 400 || pendingCount != 4000 || skippedCount != 0 || maxLiveWorkItemsPerSource != 1 {
		t.Fatalf("SC400 materialization: projection_events=%d projection_workitems=%d action_events=%d action_workitems=%d pending=%d skipped=%d max_live_workitems_per_source=%d; want 4000/400/400/400/4000/0/1",
			projectionEventCount, projectionWorkItemCount, actionEventCount, actionWorkItemCount, pendingCount, skippedCount, maxLiveWorkItemsPerSource)
	}
	t.Logf("evidence materialize_elapsed=%s projection_events=%d projection_workitems=%d action_events=%d action_workitems=%d pending=%d skipped=%d max_live_workitems_per_source=%d",
		enqueueElapsed, projectionEventCount, projectionWorkItemCount, actionEventCount, actionWorkItemCount, pendingCount, skippedCount, maxLiveWorkItemsPerSource)
}

func TestService_LegacyClaimDrainsMaterializedWorkItemWhenExecutionOwnerMissing(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "mixed-schema-legacy-claim"
		packInstall = "install-token-mixed-schema-legacy-claim"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	if _, err := db.Exec(`ALTER TABLE _System_ConnectorEvent DROP COLUMN ExecutionOwner`); err != nil {
		t.Fatalf("drop ExecutionOwner for mixed-schema proof: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	svc := NewService(db, deterministicID("mixed-schema-claim-token"), nil)
	eventID, duplicate, err := svc.Enqueue(
		ctx,
		namespace,
		"webhook.event",
		"mixed-schema-legacy-claim-event",
		json.RawMessage(`{"objectKind":"contact","sourceID":"mixed-schema-contact"}`),
		packInstall,
		1,
		"",
		"",
		0,
		"mixed-schema-delivery",
	)
	if err != nil || duplicate {
		t.Fatalf("enqueue mixed-schema Event = (%q, %v, %v), want fresh Event", eventID, duplicate, err)
	}
	var workItemCount int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, eventID).Scan(&workItemCount); err != nil {
		t.Fatalf("count materialized WorkItem: %v", err)
	}
	if workItemCount != 1 {
		t.Fatalf("materialized WorkItems = %d, want 1", workItemCount)
	}

	event, claimed, err := svc.ClaimNext(ctx, packInstall, "mixed-schema-legacy-worker")
	if err != nil {
		t.Fatalf("legacy ClaimNext with missing ExecutionOwner: %v", err)
	}
	if !claimed || event == nil || event.ID != eventID {
		t.Fatalf("legacy ClaimNext = (%+v, %v), want materialized Event %q", event, claimed, eventID)
	}
	var status string
	if err := db.QueryRowContext(ctx, `
SELECT Status
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID).Scan(&status); err != nil {
		t.Fatalf("load claimed Event status: %v", err)
	}
	if status != statusProcessing {
		t.Fatalf("claimed Event status = %q, want %q", status, statusProcessing)
	}
}

func TestService_LegacyClaimDrainsMaterializedWorkItemBeforeOwnershipTransfer(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "fallback-materialized-legacy-claim"
		packInstall = "install-token-fallback-materialized-legacy-claim"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	svc := NewService(db, deterministicID("fallback-materialized-claim-token"), nil)
	eventID, duplicate, err := svc.Enqueue(
		ctx,
		namespace,
		"webhook.event",
		"fallback-materialized-legacy-event",
		json.RawMessage(`{"objectKind":"contact","sourceID":"fallback-materialized-contact"}`),
		packInstall,
		1,
		"",
		"",
		0,
		"fallback-materialized-delivery",
	)
	if err != nil || duplicate {
		t.Fatalf("enqueue materialized Event = (%q, %v, %v), want fresh Event", eventID, duplicate, err)
	}
	var workItemCount int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, eventID).Scan(&workItemCount); err != nil {
		t.Fatalf("count materialized WorkItem: %v", err)
	}
	if workItemCount != 1 {
		t.Fatalf("materialized WorkItems = %d, want 1", workItemCount)
	}

	event, claimed, err := svc.ClaimNext(ctx, packInstall, "fallback-materialized-legacy-worker")
	if err != nil {
		t.Fatalf("legacy ClaimNext before WorkItem ownership transfer: %v", err)
	}
	if !claimed || event == nil || event.ID != eventID {
		t.Fatalf("legacy ClaimNext = (%+v, %v), want materialized Event %q", event, claimed, eventID)
	}
	var status, executionOwner string
	if err := db.QueryRowContext(ctx, `
SELECT Status, ExecutionOwner
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID).Scan(&status, &executionOwner); err != nil {
		t.Fatalf("load claimed Event ownership: %v", err)
	}
	if status != statusProcessing || executionOwner != connector_workqueue.EventExecutionOwnerEvent {
		t.Fatalf("claimed Event = %s/%s, want processing/event", status, executionOwner)
	}
}

func TestService_SettledProjectionLateEventCreatesHistoricalAndActiveGenerations(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-race"
		packInstall = "install-token-projection-race"
		eventKind   = "contact.propertychange"
		claimToken  = "projection-race-token"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	repo := connector_workqueue.NewRepository(db, nil, nil)
	firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-race-1", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-race-first",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-race","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue first projection event: %v", err)
	}
	firstEventID := firstIDs[0]
	var workID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, firstEventID).Scan(&workID); err != nil {
		t.Fatalf("load projection work item: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='processing', ClaimToken=? WHERE Id=?`, claimToken, firstEventID); err != nil {
		t.Fatalf("mark first event processing: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State='processing', LeaseToken=?, LeaseOwner='worker', LeaseUntil=DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE WorkId=?`, claimToken, workID); err != nil {
		t.Fatalf("mark projection work item processing: %v", err)
	}

	secondIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-race-2", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-race-second",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-race","property":"phone"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue late projection event: %v", err)
	}
	secondEventID := secondIDs[0]

	var workItemCount int
	var selectedEventID, state string
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ?`, packInstall).Scan(&workItemCount); err != nil {
		t.Fatalf("count race work items: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT EventId, State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&selectedEventID, &state); err != nil {
		t.Fatalf("load in-flight projection work item: %v", err)
	}
	if workItemCount != 2 || selectedEventID != firstEventID || state != connector_workqueue.WorkStateProcessing {
		t.Fatalf("in-flight projection changed early: count=%d event=%s state=%s", workItemCount, selectedEventID, state)
	}
	var successorMirrorCount int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE EventId = ?`, secondEventID).Scan(&successorMirrorCount); err != nil {
		t.Fatalf("count late successor mirrors: %v", err)
	}
	if successorMirrorCount != 1 {
		t.Fatalf("late source-projection successor has %d queued WorkItems, want 1", successorMirrorCount)
	}
	readinessGap, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(ctx, connector_workqueue.LegacyEventMirrorReadinessInput{InstalledPackID: packInstall})
	if err != nil {
		t.Fatalf("check source-projection successor readiness: %v", err)
	}
	if readinessGap {
		t.Fatal("source-projection successor without a direct EventId mirror must remain readiness-safe")
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorInboundRoute SET IsCurrent = 0, IsDeleted = 1 WHERE Namespace = ?`, namespace); err != nil {
		t.Fatalf("sweep settled projection route: %v", err)
	}
	duplicateIDs, duplicates, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-race-duplicate", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-race-second",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-race","property":"phone"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("duplicate late projection event after route sweep: %v", err)
	}
	if len(duplicateIDs) != 1 || duplicateIDs[0] != secondEventID || len(duplicates) != 1 || !duplicates[0] {
		t.Fatalf("swept-route duplicate = ids=%v duplicates=%v, want existing successor %s", duplicateIDs, duplicates, secondEventID)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ?`, packInstall).Scan(&workItemCount); err != nil {
		t.Fatalf("count swept-route duplicate WorkItems: %v", err)
	}
	if workItemCount != 2 {
		t.Fatalf("swept-route duplicate created %d WorkItems, want active plus queued successor", workItemCount)
	}

	now := time.Now().UTC().Truncate(time.Second)
	completed, err := svc.CompleteSourceProjectionEventForWorkItem(ctx, firstEventID, workID, claimToken, statusProcessed, "", now)
	if err != nil || !completed {
		t.Fatalf("atomically complete event and advance projection work item: completed=%v err=%v", completed, err)
	}
	var firstStatus string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, firstEventID).Scan(&firstStatus); err != nil {
		t.Fatalf("load completed projection event: %v", err)
	}
	var historicalProjectionKey string
	var historicalActiveKey sql.NullString
	var terminalAt sql.NullTime
	if err := db.QueryRowContext(ctx, `
SELECT EventId, State, COALESCE(ProjectionKey, ''), ProjectionActiveKey, TerminalAt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(
		&selectedEventID, &state, &historicalProjectionKey, &historicalActiveKey, &terminalAt,
	); err != nil {
		t.Fatalf("load historical projection generation: %v", err)
	}
	if firstStatus != statusProcessing || selectedEventID != firstEventID || state != connector_workqueue.WorkStateProcessed ||
		historicalProjectionKey != sourceProjectionKeyFromWorkID(workID) || historicalActiveKey.Valid || !terminalAt.Valid {
		t.Fatalf("historical generation immutable-event-status=%s selected=%s state=%s projection=%s active=%v terminal=%v",
			firstStatus, selectedEventID, state, historicalProjectionKey, historicalActiveKey, terminalAt)
	}
	var activeWorkID string
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, EventId, State
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey = ?`, sourceProjectionKeyFromWorkID(workID)).Scan(&activeWorkID, &selectedEventID, &state); err != nil {
		t.Fatalf("load active projection generation: %v", err)
	}
	if activeWorkID == workID || selectedEventID != secondEventID ||
		(state != connector_workqueue.WorkStatePending && state != connector_workqueue.WorkStateBlocked) {
		t.Fatalf("active generation work=%s selected=%s state=%s, want distinct work for %s pending/blocked",
			activeWorkID, selectedEventID, state, secondEventID)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE ProjectionKey = ?`, sourceProjectionKeyFromWorkID(workID)).Scan(&workItemCount); err != nil {
		t.Fatalf("count projection history: %v", err)
	}
	if workItemCount != 2 {
		t.Fatalf("projection generation count=%d, want historical + active", workItemCount)
	}
}

// Once the WorkItem claimant atomically transfers a source projection, every
// legacy Event claim surface observes the durable ExecutionOwner fence.
func TestService_WorkItemOwnedProjectionFencesEveryLegacyClaimSurface(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-legacy-mirror"
		packInstall = "install-token-projection-legacy-mirror"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "legacy-mirror-delivery", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "legacy-mirror-event",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"legacy-mirror-contact","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue source projection Event: %v", err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET RunAfter = ?
 WHERE EventId = ?`, now, ids[0]); err != nil {
		t.Fatalf("make settled source projection WorkItem due: %v", err)
	}
	work, claimed, err := svc.WorkItemRepository().ClaimNextWorkItem(
		connector_workqueue.WithDistributedClaim(ctx),
		connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: packInstall,
			Generation:      0,
			OwnerID:         "workitem-owner",
			FencingToken:    "workitem-fence",
			LeaseToken:      "workitem-lease",
			LeaseUntil:      now.Add(time.Minute),
			Now:             now,
		},
	)
	if err != nil || !claimed || work == nil || work.EventID != ids[0] {
		t.Fatalf("transfer source projection to WorkItem = (%+v, %v, %v)", work, claimed, err)
	}

	// The default legacy worker uses PeekBatch followed by ClaimByID, while
	// central schedulers use PeekClaimableBatch and ClaimBatchByID. All of them
	// must observe the same transferred owner.
	legacyPeek, err := svc.PeekBatch(ctx, packInstall, 1)
	if err != nil {
		t.Fatalf("legacy PeekBatch: %v", err)
	}
	if len(legacyPeek) != 0 {
		t.Fatalf("legacy PeekBatch returned source-projection Event %s", legacyPeek[0].ID)
	}
	claimablePeek, err := svc.PeekClaimableBatch(ctx, packInstall, 1)
	if err != nil {
		t.Fatalf("legacy PeekClaimableBatch: %v", err)
	}
	if len(claimablePeek) != 0 {
		t.Fatalf("legacy PeekClaimableBatch returned source-projection Event %s", claimablePeek[0].ID)
	}
	if _, _, claimed, err := svc.ClaimByID(ctx, ids[0], "legacy-event-worker", "", sql.NullString{}, 0); err != nil || claimed {
		t.Fatalf("legacy ClaimByID = (claimed=%v, err=%v), want false, nil", claimed, err)
	}
	batchClaimed, err := svc.ClaimBatchByID(ctx, []BatchClaimRequest{{EventID: ids[0], WorkerID: "legacy-event-worker"}})
	if err != nil {
		t.Fatalf("legacy ClaimBatchByID: %v", err)
	}
	if len(batchClaimed) != 0 {
		t.Fatalf("legacy ClaimBatchByID claimed source-projection Event: %v", batchClaimed)
	}

	legacyEvent, claimed, err := svc.ClaimNext(ctx, packInstall, "legacy-event-worker")
	if err != nil {
		t.Fatalf("legacy ClaimNext: %v", err)
	}
	if claimed || legacyEvent != nil {
		t.Fatalf("legacy ClaimNext = (%+v, %v), want transferred projection fenced", legacyEvent, claimed)
	}

	var eventStatus, workState string
	if err := db.QueryRowContext(ctx, `
SELECT e.Status, w.State
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e ON e.Id = w.EventId
 WHERE e.Id = ?`, ids[0]).Scan(&eventStatus, &workState); err != nil {
		t.Fatalf("load preserved source projection mirror: %v", err)
	}
	if eventStatus != statusSkipped || workState != connector_workqueue.WorkStateProcessing {
		t.Fatalf("transferred Event/WorkItem = %s/%s, want skipped/processing", eventStatus, workState)
	}
}

func TestService_WorkItemOwnershipDurablyFencesEveryLegacyClaimSurface(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		eventID    = "durable-legacy-fence-event"
		workID     = "cwi:durable-legacy-fence-event"
		packID     = "durable-legacy-fence-pack"
		workerID   = "legacy-worker"
		eventKind  = "contact.updated"
		sourceLane = "contact:durable-legacy-fence"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, SourceLaneKey, ExecutionOwner, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'durable-legacy-fence', ?, ?, 'durable-legacy-fence-key', 'pending',
        'default', ?, 'work_item', 1, ?, ?)`,
		eventID, packID, eventKind, sourceLane, now, now); err != nil {
		t.Fatalf("seed legacy Event candidate: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, SourceLaneKey, Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'pending', ?,
        0, ?, 0, 1, ?, ?)`,
		workID, eventID, packID, now, sourceLane, now, now); err != nil {
		t.Fatalf("seed durable WorkItem owner: %v", err)
	}

	legacy := NewService(db, nil, func() time.Time { return now })
	if events, err := legacy.PeekBatch(ctx, packID, 1); err != nil || len(events) != 0 {
		t.Fatalf("PeekBatch = (%+v, %v), want WorkItem-owned Event fenced", events, err)
	}
	if events, err := legacy.PeekClaimableBatch(ctx, packID, 1); err != nil || len(events) != 0 {
		t.Fatalf("PeekClaimableBatch = (%+v, %v), want WorkItem-owned Event fenced", events, err)
	}
	if _, _, claimed, err := legacy.ClaimByID(ctx, eventID, workerID, "", sql.NullString{}, 0); err != nil || claimed {
		t.Fatalf("ClaimByID = (claimed=%v, err=%v), want durable WorkItem fence", claimed, err)
	}
	claimedBatch, err := legacy.ClaimBatchByID(ctx, []BatchClaimRequest{{EventID: eventID, WorkerID: workerID}})
	if err != nil || len(claimedBatch) != 0 {
		t.Fatalf("ClaimBatchByID = (%v, %v), want durable WorkItem fence", claimedBatch, err)
	}
	if event, claimed, err := legacy.ClaimNext(ctx, packID, workerID); err != nil || claimed || event != nil {
		t.Fatalf("ClaimNext = (%+v, %v, %v), want durable WorkItem fence", event, claimed, err)
	}

	var eventStatus, workState string
	if err := db.QueryRowContext(ctx, `
SELECT e.Status, w.State
  FROM _System_ConnectorEvent e
  JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id
 WHERE e.Id = ?`, eventID).Scan(&eventStatus, &workState); err != nil {
		t.Fatalf("reload fenced ownership rows: %v", err)
	}
	if eventStatus != statusPending || workState != connector_workqueue.WorkStatePending {
		t.Fatalf("fenced rows changed to Event=%s WorkItem=%s, want pending/pending", eventStatus, workState)
	}
}

// A stale Event remains exclusively WorkItem-owned after the Event-to-WorkItem
// transfer. Every legacy stale-reclaim surface must leave its Event lease and
// processing WorkItem lease untouched.
func TestService_WorkItemOwnershipDurablyFencesLegacyStaleReclaimSurfaces(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	staleAt := now.Add(-(StaleProcessingSeconds + 1) * time.Second)
	leaseUntil := now.Add(time.Minute)
	const (
		eventID           = "durable-stale-reclaim-fence-event"
		controlID         = "durable-stale-reclaim-fence-control"
		workID            = "cwi:durable-stale-reclaim-fence-event"
		packID            = "durable-stale-reclaim-fence-pack"
		workerID          = "legacy-stale-worker"
		eventToken        = "legacy-stale-event-lease"
		controlToken      = "legacy-stale-control-lease"
		workToken         = "workitem-stale-lease"
		sourceLane        = "contact:durable-stale-reclaim-fence"
		controlSourceLane = "contact:durable-stale-reclaim-control"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, ClaimedBy, ClaimedAt, ClaimToken, ExecutionOwner,
        SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'durable-stale-reclaim-fence', ?, 'contact.updated',
        'durable-stale-reclaim-fence-key', 'processing',
        'default', 'legacy-stale-worker', ?, ?, 'work_item', ?, 1, ?, ?)`,
		eventID, packID, staleAt, eventToken, sourceLane, staleAt, staleAt); err != nil {
		t.Fatalf("seed stale WorkItem-owned Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, OwnershipEpoch, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey,
        Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'processing', ?,
        3, 7, 'workitem-stale-owner', ?, ?, ?, 0, 1, ?, ?)`,
		workID, eventID, packID, staleAt, workToken, leaseUntil, sourceLane, staleAt, now); err != nil {
		t.Fatalf("seed processing WorkItem owner: %v", err)
	}
	// This control is identical on the legacy claim predicates and differs only
	// in the Event ownership fact (and its isolated lane). It proves that an
	// empty result below comes from the WorkItem fence, rather than a vacuous
	// stale-reclaim query.
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, ClaimedBy, ClaimedAt, ClaimToken, ExecutionOwner,
        SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'durable-stale-reclaim-fence', ?, 'contact.updated',
        'durable-stale-reclaim-fence-control-key', 'processing',
        'default', 'legacy-stale-worker', ?, ?, 'event', ?, 2, ?, ?)`,
		controlID, packID, staleAt, controlToken, controlSourceLane, staleAt, staleAt); err != nil {
		t.Fatalf("seed stale Event-owned control: %v", err)
	}

	var rawStaleCandidates int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND Status = 'processing'
   AND LastModifiedDate < ?`, packID, now.Add(-StaleProcessingSeconds*time.Second)).Scan(&rawStaleCandidates); err != nil {
		t.Fatalf("count otherwise reclaimable stale Event: %v", err)
	}
	if rawStaleCandidates != 2 {
		t.Fatalf("otherwise reclaimable stale Events = %d, want 2", rawStaleCandidates)
	}

	legacy := NewService(db, deterministicID("durable-stale-reclaim-fence-token"), func() time.Time { return now })
	for _, tt := range []struct {
		name string
		peek func() ([]*ConnectorEvent, error)
	}{
		{name: "peek_stale", peek: func() ([]*ConnectorEvent, error) {
			return legacy.PeekStaleBatch(ctx, packID, 2)
		}},
		{name: "peek_stale_for_classes", peek: func() ([]*ConnectorEvent, error) {
			return legacy.PeekStaleBatchForClasses(ctx, packID, []string{"default"}, 2)
		}},
		{name: "peek_claimable_stale", peek: func() ([]*ConnectorEvent, error) {
			return legacy.PeekClaimableStaleBatch(ctx, packID, 2)
		}},
		{name: "peek_claimable_stale_for_classes", peek: func() ([]*ConnectorEvent, error) {
			return legacy.PeekClaimableStaleBatchForClasses(ctx, packID, []string{"default"}, 2)
		}},
	} {
		t.Run(tt.name, func(t *testing.T) {
			events, err := tt.peek()
			if err != nil || len(events) != 1 || events[0] == nil || events[0].ID != controlID {
				t.Fatalf("%s = (%+v, %v), want only Event-owned stale control %q", tt.name, events, err, controlID)
			}
		})
	}
	resetControl := func(t *testing.T) {
		t.Helper()
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = 'processing',
       ClaimedBy = 'legacy-stale-worker',
       ClaimedAt = ?,
       ClaimToken = ?,
       ExecutionOwner = 'event',
       LastModifiedDate = ?
 WHERE Id = ?`, staleAt, controlToken, staleAt, controlID); err != nil {
			t.Fatalf("reset stale Event-owned control: %v", err)
		}
	}
	for _, tt := range []struct {
		name    string
		reclaim func() (*ConnectorEvent, bool, error)
	}{
		{name: "reclaim_stale", reclaim: func() (*ConnectorEvent, bool, error) {
			return legacy.ReclaimStaleNext(ctx, packID, workerID)
		}},
		{name: "reclaim_stale_for_classes", reclaim: func() (*ConnectorEvent, bool, error) {
			return legacy.ReclaimStaleNextForClasses(ctx, packID, workerID, []string{"default"})
		}},
	} {
		t.Run(tt.name, func(t *testing.T) {
			resetControl(t)
			event, claimed, err := tt.reclaim()
			if err != nil || !claimed || event == nil || event.ID != controlID {
				t.Fatalf("%s = (%+v, %v, %v), want Event-owned stale control %q", tt.name, event, claimed, err, controlID)
			}
			event, claimed, err = tt.reclaim()
			if err != nil || claimed || event != nil {
				t.Fatalf("%s after control claim = (%+v, %v, %v), want WorkItem-owned stale Event fenced", tt.name, event, claimed, err)
			}
		})
	}
	if _, _, claimed, err := legacy.ClaimByID(ctx, eventID, workerID, "", sql.NullString{}, 0); err != nil || claimed {
		t.Fatalf("ClaimByID on WorkItem-owned stale Event = (claimed=%v, err=%v), want false, nil", claimed, err)
	}
	claimedBatch, err := legacy.ClaimBatchByID(ctx, []BatchClaimRequest{{EventID: eventID, WorkerID: workerID}})
	if err != nil || len(claimedBatch) != 0 {
		t.Fatalf("ClaimBatchByID on WorkItem-owned stale Event = (%v, %v), want no claims", claimedBatch, err)
	}

	var eventStatus, eventOwner, eventClaimedBy, eventClaimToken string
	var eventClaimedAt, eventModifiedAt time.Time
	if err := db.QueryRowContext(ctx, `
SELECT Status, ExecutionOwner, COALESCE(ClaimedBy, ''), COALESCE(ClaimToken, ''),
       ClaimedAt, LastModifiedDate
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID).Scan(
		&eventStatus, &eventOwner, &eventClaimedBy, &eventClaimToken, &eventClaimedAt, &eventModifiedAt,
	); err != nil {
		t.Fatalf("reload fenced stale Event: %v", err)
	}
	var workState, workLeaseOwner, workLeaseToken string
	var workAttempt, workEpoch int64
	var workLeaseUntil, workModifiedAt time.Time
	if err := db.QueryRowContext(ctx, `
SELECT State, Attempt, OwnershipEpoch, COALESCE(LeaseOwner, ''), COALESCE(LeaseToken, ''),
       LeaseUntil, LastModifiedDate
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(
		&workState, &workAttempt, &workEpoch, &workLeaseOwner, &workLeaseToken, &workLeaseUntil, &workModifiedAt,
	); err != nil {
		t.Fatalf("reload fenced processing WorkItem: %v", err)
	}
	if eventStatus != statusProcessing || eventOwner != connector_workqueue.EventExecutionOwnerWorkItem ||
		eventClaimedBy != "legacy-stale-worker" || eventClaimToken != eventToken ||
		!eventClaimedAt.Equal(staleAt) || !eventModifiedAt.Equal(staleAt) ||
		workState != connector_workqueue.WorkStateProcessing || workAttempt != 3 || workEpoch != 7 ||
		workLeaseOwner != "workitem-stale-owner" || workLeaseToken != workToken ||
		!workLeaseUntil.Equal(leaseUntil) || !workModifiedAt.Equal(now) {
		t.Fatalf("stale fence mutated Event/WorkItem: event=%s/%s/%s/%s/%v/%v work=%s/%d/%d/%s/%s/%v/%v",
			eventStatus, eventOwner, eventClaimedBy, eventClaimToken, eventClaimedAt, eventModifiedAt,
			workState, workAttempt, workEpoch, workLeaseOwner, workLeaseToken, workLeaseUntil, workModifiedAt)
	}
}

func TestService_PartialProjectionSchemaKeepsLegacyEventClaimLive(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		selectedID = "partial-projection-selected"
		targetID   = "partial-projection-target"
		packID     = "partial-projection-pack"
		laneKey    = "contact:partial-projection"
		token      = "partial-projection-event-token"
	)
	for i, event := range []struct {
		id, status, claimToken string
	}{
		{selectedID, statusPending, ""},
		{targetID, statusProcessing, token},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, ClaimToken, RouteVersion, SourceLaneKey, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'partial-projection', ?, 'contact.updated', ?, ?, 'default',
        NULLIF(?, ''), 1, ?, ?, ?, ?)`,
			event.id, packID, event.id, event.status, event.claimToken, laneKey, i+1,
			now.Add(time.Duration(i)*time.Microsecond), now); err != nil {
			t.Fatalf("seed Event %s: %v", event.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, SourceLaneKey, Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES ('csp:partial-projection', ?, ?, 'connector_event', 'default', 'processing', ?,
        1, ?, 0, 1, ?, ?)`,
		selectedID, packID, now, laneKey, now, now); err != nil {
		t.Fatalf("seed legacy source projection WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX uniq_connector_work_item_projection_active`); err != nil {
		t.Fatalf("drop projection active index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX idx_connector_work_item_projection_history`); err != nil {
		t.Fatalf("drop projection history index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX idx_connector_work_item_projection_order`); err != nil {
		t.Fatalf("drop projection order index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP COLUMN ProjectionActiveKey`); err != nil {
		t.Fatalf("drop ProjectionActiveKey: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP COLUMN ProjectionKey`); err != nil {
		t.Fatalf("drop ProjectionKey: %v", err)
	}

	err := NewService(db, nil, func() time.Time { return now.Add(time.Second) }).ReleaseClaim(ctx, targetID, token)
	if err != nil {
		t.Fatalf("ReleaseClaim with partial projection schema: %v", err)
	}
	var status, claimToken string
	if err := db.QueryRowContext(ctx, `
SELECT Status, COALESCE(ClaimToken, '')
  FROM _System_ConnectorEvent
 WHERE Id = ?`, targetID).Scan(&status, &claimToken); err != nil {
		t.Fatalf("reload fenced target Event: %v", err)
	}
	if status != statusPending || claimToken != "" {
		t.Fatalf("partial-schema legacy release left target at %s/%s, want pending with no claim token", status, claimToken)
	}
}

func TestService_MissingUnqueriedClaimColumnKeepsLegacyProjectionGuardTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		eventID = "partial-claim-column-event"
		workID  = "csp:partial-claim-column"
		packID  = "partial-claim-column-pack"
		laneKey = "contact:partial-claim-column"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, RouteVersion, SourceLaneKey, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'partial-claim-column', ?, 'contact.updated', ?, 'pending',
        'default', 1, ?, 1, ?, ?)`,
		eventID, packID, eventID, laneKey, now, now); err != nil {
		t.Fatalf("seed partial-schema Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, SourceLaneKey, Generation, LaneOrdinal,
        ProjectionKey, ProjectionActiveKey, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'pending', ?,
        0, ?, 0, 1, 'projection-partial', 'projection-partial', ?, ?)`,
		workID, eventID, packID, now, laneKey, now, now); err != nil {
		t.Fatalf("seed active projection WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
ALTER TABLE _System_ConnectorWorkItem DROP COLUMN FailureCount`); err != nil {
		t.Fatalf("drop required claim column: %v", err)
	}

	svc := NewService(db, deterministicID("partial-claim-column-token"), func() time.Time {
		return now.Add(time.Second)
	})
	event, claimed, err := svc.ClaimNext(ctx, packID, "legacy-partial-worker")
	if err != nil {
		t.Fatalf("ClaimNext with incomplete WorkItem schema: %v", err)
	}
	if claimed || event != nil {
		t.Fatalf("ClaimNext = (%+v, %v), want durable projection guard to fail closed", event, claimed)
	}
}

func TestService_TerminalProjectionHistoryDoesNotFenceLegacyLaneTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		selectedID = "terminal-projection-selected"
		targetID   = "terminal-projection-target"
		packID     = "terminal-projection-pack"
		laneKey    = "contact:terminal-projection"
	)
	for ordinal, item := range []struct {
		id     string
		status string
	}{
		{selectedID, statusProcessed},
		{targetID, statusPending},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, RouteVersion, SourceLaneKey, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'terminal-projection', ?, 'contact.updated', ?, ?,
        'default', 1, ?, ?, ?, ?)`,
			item.id, packID, item.id, item.status, laneKey, ordinal+1,
			now.Add(time.Duration(ordinal)*time.Microsecond), now); err != nil {
			t.Fatalf("seed Event %s: %v", item.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, SourceLaneKey, Generation, LaneOrdinal,
        ProjectionKey, ProjectionActiveKey, TerminalAt, CreatedDate, LastModifiedDate)
VALUES ('csp:terminal-projection', ?, ?, 'connector_event', 'default', 'processed', ?,
        1, ?, 0, 1, 'projection-terminal', NULL, ?, ?, ?)`,
		selectedID, packID, now, laneKey, now, now, now); err != nil {
		t.Fatalf("seed terminal projection WorkItem: %v", err)
	}

	svc := NewService(db, deterministicID("terminal-projection-token"), func() time.Time {
		return now.Add(time.Second)
	})
	event, claimed, err := svc.ClaimNext(ctx, packID, "legacy-terminal-history-worker")
	if err != nil {
		t.Fatalf("ClaimNext after terminal projection: %v", err)
	}
	if !claimed || event == nil || event.ID != targetID {
		t.Fatalf("ClaimNext = (%+v, %v), want target after terminal projection", event, claimed)
	}
}

func TestService_WorkItemRowDurablyFencesEveryLegacyFinalizer(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	type finalizerCase struct {
		name       string
		retryCount int
		finalize   func(*Service, string, string) error
	}
	cases := []finalizerCase{
		{name: "release", finalize: func(s *Service, eventID, token string) error {
			return s.ReleaseClaim(ctx, eventID, token)
		}},
		{name: "renew", finalize: func(s *Service, eventID, token string) error {
			return s.RenewLease(ctx, eventID, token)
		}},
		{name: "release_for_retry", finalize: func(s *Service, eventID, token string) error {
			return s.ReleaseClaimForRetry(ctx, eventID, token, "transient")
		}},
		{name: "processed", finalize: func(s *Service, eventID, token string) error {
			return s.MarkProcessed(ctx, eventID, token)
		}},
		{name: "skipped", finalize: func(s *Service, eventID, token string) error {
			return s.MarkSkipped(ctx, eventID, token)
		}},
		{name: "failed_retry", finalize: func(s *Service, eventID, token string) error {
			return s.MarkFailed(ctx, eventID, token, "failure")
		}},
		{name: "failed_dead", retryCount: MaxRetryAttempts - 1, finalize: func(s *Service, eventID, token string) error {
			return s.MarkFailed(ctx, eventID, token, "terminal failure")
		}},
	}

	for i, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			eventID := fmt.Sprintf("durable-finalizer-event-%d", i)
			workID := fmt.Sprintf("cwi:durable-finalizer-event-%d", i)
			claimToken := fmt.Sprintf("legacy-finalizer-token-%d", i)
			if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, RetryCount, ClaimedBy, ClaimedAt, ClaimToken, ExecutionOwner,
        EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'durable-finalizer-fence', 'durable-finalizer-pack', 'contact.updated', ?,
        'processing', 'default', ?, 'legacy-worker', ?, ?, 'work_item', ?, ?, ?)`,
				eventID, eventID, tt.retryCount, now, claimToken, i+1, now, now); err != nil {
				t.Fatalf("seed Event: %v", err)
			}
			if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, LeaseOwner, LeaseToken, LeaseUntil, Generation, LaneOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, ?, 'durable-finalizer-pack', 'connector_event', 'default', 'processing', ?,
        1, 'workitem-worker', ?, DATE_ADD(?, INTERVAL 1 MINUTE), 0, ?, ?, ?)`,
				workID, eventID, now, "workitem-"+claimToken, now, i+1, now, now); err != nil {
				t.Fatalf("seed WorkItem: %v", err)
			}

			svc := NewService(db, nil, func() time.Time { return now.Add(time.Second) })
			if err := tt.finalize(svc, eventID, claimToken); !errors.Is(err, ErrLeaseLost) {
				t.Fatalf("legacy finalizer error = %v, want ErrLeaseLost", err)
			}

			var status, token, workState, workLease string
			var retryCount int
			var lastModified time.Time
			if err := db.QueryRowContext(ctx, `
SELECT Status, RetryCount, COALESCE(ClaimToken, ''), LastModifiedDate
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID).Scan(&status, &retryCount, &token, &lastModified); err != nil {
				t.Fatalf("reload Event: %v", err)
			}
			if err := db.QueryRowContext(ctx, `
SELECT State, COALESCE(LeaseToken, '')
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(&workState, &workLease); err != nil {
				t.Fatalf("reload WorkItem: %v", err)
			}
			if status != statusProcessing || retryCount != tt.retryCount || token != claimToken ||
				!lastModified.Equal(now) || workState != connector_workqueue.WorkStateProcessing ||
				workLease != "workitem-"+claimToken {
				t.Fatalf("fenced finalizer mutated Event/WorkItem: status=%s retry=%d token=%s modified=%v work=%s/%s",
					status, retryCount, token, lastModified, workState, workLease)
			}
		})
	}
}

func TestService_WorkItemOwnershipDurablyFencesLegacyCoalescedHandoff(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		currentID = "durable-coalesce-current"
		nextID    = "durable-coalesce-next"
		workID    = "cwi:durable-coalesce-current"
		packID    = "durable-coalesce-pack"
		token     = "durable-coalesce-token"
		laneKey   = "contact:durable-coalesce"
	)
	for i, event := range []struct {
		id, status, claimToken string
	}{
		{currentID, statusProcessing, token},
		{nextID, statusPending, ""},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, Payload, ClaimToken, SourceLaneKey, ExecutionOwner, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'durable-coalesce', ?, 'contact.updated', ?, ?, 'default',
        '{"objectKind":"contact","sourceID":"durable-coalesce"}', NULLIF(?, ''), ?,
        CASE WHEN ? = 0 THEN 'work_item' ELSE 'event' END, ?, ?, ?)`,
			event.id, packID, event.id, event.status, event.claimToken, laneKey, i, i+1, now.Add(time.Duration(i)*time.Microsecond), now); err != nil {
			t.Fatalf("seed Event %s: %v", event.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey,
        Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'processing', ?,
        1, 'workitem-worker', 'workitem-token', DATE_ADD(?, INTERVAL 1 MINUTE), ?,
        0, 1, ?, ?)`,
		workID, currentID, packID, now, now, laneKey, now, now); err != nil {
		t.Fatalf("seed WorkItem: %v", err)
	}

	next, claimed, err := NewService(db, nil, func() time.Time { return now.Add(time.Second) }).
		ClaimNextCoalescedSibling(ctx, currentID, token, "legacy-worker")
	if !errors.Is(err, ErrLeaseLost) || claimed || next != nil {
		t.Fatalf("coalesced handoff = (%+v, %v, %v), want nil, false, ErrLeaseLost", next, claimed, err)
	}

	var currentStatus, currentToken, nextStatus, nextToken, workState string
	if err := db.QueryRowContext(ctx, `SELECT Status, COALESCE(ClaimToken, '') FROM _System_ConnectorEvent WHERE Id = ?`, currentID).
		Scan(&currentStatus, &currentToken); err != nil {
		t.Fatalf("reload current Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT Status, COALESCE(ClaimToken, '') FROM _System_ConnectorEvent WHERE Id = ?`, nextID).
		Scan(&nextStatus, &nextToken); err != nil {
		t.Fatalf("reload next Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&workState); err != nil {
		t.Fatalf("reload WorkItem: %v", err)
	}
	if currentStatus != statusProcessing || currentToken != token ||
		nextStatus != statusPending || nextToken != "" ||
		workState != connector_workqueue.WorkStateProcessing {
		t.Fatalf("coalesced fence mutated current=%s/%s next=%s/%s work=%s",
			currentStatus, currentToken, nextStatus, nextToken, workState)
	}
}

func TestService_WorkItemExecutorCoalescingAdvancesWorkItemsAndKeepsEventsImmutable(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createInstalledPackTable(t, db, "executor-coalesce")
	createConnectorEventTable(t, db)

	ctx, cancel := context.WithTimeout(connector_workqueue.WithDistributedClaim(context.Background()), 30*time.Second)
	defer cancel()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		currentID    = "executor-coalesce-current"
		nextID       = "executor-coalesce-next"
		currentWork  = "cwi:executor-coalesce-current"
		nextWork     = "cwi:executor-coalesce-next"
		packID       = "executor-coalesce-pack"
		laneKey      = "contact:executor-coalesce"
		currentLease = "executor-coalesce-current-lease"
		nextLease    = "executor-coalesce-next-lease"
	)
	for i, event := range []struct {
		id, status, executionOwner string
	}{
		{currentID, statusSkipped, connector_workqueue.EventExecutionOwnerWorkItem},
		{nextID, statusPending, connector_workqueue.EventExecutionOwnerEvent},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
        QueueClass, Payload, SourceLaneKey, ExecutionOwner, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'executor-coalesce', ?, 'contact.updated', ?, ?, 'default',
        '{"objectKind":"contact","sourceID":"executor-coalesce"}', ?, ?, ?, ?, ?)`,
			event.id, packID, event.id, event.status, laneKey, event.executionOwner, i+1,
			now.Add(time.Duration(i)*time.Microsecond), now); err != nil {
			t.Fatalf("seed Event %s: %v", event.id, err)
		}
	}
	for i, work := range []struct {
		id, eventID, state, leaseOwner, leaseToken string
		attempt                                    int
	}{
		{currentWork, currentID, connector_workqueue.WorkStateProcessing, "executor-worker", currentLease, 1},
		{nextWork, nextID, connector_workqueue.WorkStatePending, "", "", 0},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey,
        Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', ?, ?, ?,
        NULLIF(?, ''), NULLIF(?, ''), CASE WHEN ? = '' THEN NULL ELSE DATE_ADD(?, INTERVAL 1 MINUTE) END,
        ?, 0, ?, ?, ?)`,
			work.id, work.eventID, packID, work.state, now, work.attempt,
			work.leaseOwner, work.leaseToken, work.leaseToken, now, laneKey, i+1,
			now.Add(time.Duration(i)*time.Microsecond), now); err != nil {
			t.Fatalf("seed WorkItem %s: %v", work.id, err)
		}
	}

	svc := NewService(db, nil, func() time.Time { return now })
	claim, claimed, err := svc.ClaimNextCoalescedExecutorWorkItemSibling(
		ctx, currentID, currentWork, currentLease, "executor-worker",
		"executor-owner", "executor-fence", nextLease, 0,
		now.Add(5*time.Minute), now,
	)
	if err != nil {
		t.Fatalf("ClaimNextCoalescedExecutorWorkItemSibling: %v", err)
	}
	if !claimed || claim == nil || claim.Event == nil || claim.Event.ID != nextID ||
		claim.Work == nil || claim.Work.WorkID != nextWork || claim.Work.LeaseToken != nextLease {
		t.Fatalf("executor coalesced claim = (%+v, %v), want %s/%s", claim, claimed, nextID, nextWork)
	}

	var currentWorkState, nextWorkState, nextWorkLease string
	if err := db.QueryRowContext(ctx, `
SELECT current_work.State, next_work.State, COALESCE(next_work.LeaseToken, '')
  FROM _System_ConnectorWorkItem current_work
  JOIN _System_ConnectorWorkItem next_work ON next_work.WorkId = ?
 WHERE current_work.WorkId = ?`, nextWork, currentWork).
		Scan(&currentWorkState, &nextWorkState, &nextWorkLease); err != nil {
		t.Fatalf("load coalesced WorkItems: %v", err)
	}
	if currentWorkState != connector_workqueue.WorkStateProcessed ||
		nextWorkState != connector_workqueue.WorkStateProcessing || nextWorkLease != nextLease {
		t.Fatalf("coalesced WorkItems = current %s next %s/%s", currentWorkState, nextWorkState, nextWorkLease)
	}

	rows, err := db.QueryContext(ctx, `
SELECT Id, Status, ExecutionOwner, COALESCE(ClaimToken, '')
  FROM _System_ConnectorEvent
 WHERE Id IN (?, ?)
 ORDER BY Id`, currentID, nextID)
	if err != nil {
		t.Fatalf("load immutable Events: %v", err)
	}
	defer rows.Close()
	for rows.Next() {
		var eventID, status, owner, claimToken string
		if err := rows.Scan(&eventID, &status, &owner, &claimToken); err != nil {
			t.Fatalf("scan immutable Event: %v", err)
		}
		if status != statusSkipped || owner != connector_workqueue.EventExecutionOwnerWorkItem || claimToken != "" {
			t.Fatalf("Event %s executor fields = %s/%s/%q, want skipped/work_item/empty", eventID, status, owner, claimToken)
		}
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate immutable Events: %v", err)
	}
}

func TestService_WorkItemLifecycleLeavesEventExecutorFieldsImmutable(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorEventTable(t, db)
	createConnectorWorkAttemptTable(t, db)

	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		eventID = "immutable-event-lifecycle"
		workID  = "cwi:immutable-event-lifecycle"
		packID  = "immutable-event-pack"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, ExecutionOwner,
        QueueClass, Payload, RetryCount, LastError, LastErrorAt, NextAttemptAt,
        ClaimedBy, ClaimedAt, ClaimToken, ProcessedAt, ReplayedAt, ReplayOfId,
        RouteVersion, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'immutable-event', ?, 'contact.updated', 'immutable-event-key', 'skipped', 'work_item',
        'default', '{"id":"immutable"}', 77, 'historical error', ?, ?,
        'historical-worker', ?, 'historical-token', ?, ?, 'historical-event',
        9, 4, 12, ?, ?)`,
		eventID, packID,
		now.Add(-6*time.Minute), now.Add(-5*time.Minute), now.Add(-4*time.Minute),
		now.Add(-3*time.Minute), now.Add(-2*time.Minute), now.Add(-time.Minute), now); err != nil {
		t.Fatalf("seed immutable Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'pending', ?,
        0, 0, 12, ?, ?)`,
		workID, eventID, packID, now, now, now); err != nil {
		t.Fatalf("seed WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, LeaseOwner, LeaseToken, LeaseUntil, Generation, LaneOrdinal,
        CreatedDate, LastModifiedDate)
VALUES ('complete-attempt-work', 'complete-attempt-event', ?, 'synthetic', 'default', 'processing', ?,
        3, 'complete-worker', 'complete-lease', DATE_ADD(?, INTERVAL 1 MINUTE), 0, 13, ?, ?)`,
		packID, now, now, now, now); err != nil {
		t.Fatalf("seed completion WorkItem: %v", err)
	}

	svc := NewService(db, nil, func() time.Time { return now })
	before, err := svc.loadByID(ctx, eventID)
	if err != nil {
		t.Fatalf("load Event before WorkItem lifecycle: %v", err)
	}
	repo := connector_workqueue.NewRepository(db, nil, func() time.Time { return now })
	claimCtx := connector_workqueue.WithDistributedClaim(ctx)
	first, claimed, err := repo.ClaimNextWorkItem(claimCtx, connector_workqueue.ClaimWorkItemInput{
		InstalledPackID: packID,
		Generation:      0,
		OwnerID:         "worker-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-1",
		LeaseUntil:      now.Add(time.Minute),
		Now:             now,
	})
	if err != nil || !claimed || first == nil {
		t.Fatalf("first WorkItem claim = (%+v, %v, %v)", first, claimed, err)
	}
	if renewed, err := repo.RenewWorkItemLease(ctx, workID, "lease-1", now.Add(2*time.Minute), now.Add(time.Second)); err != nil || !renewed {
		t.Fatalf("renew WorkItem = (%v, %v)", renewed, err)
	}
	retryAt := now.Add(10 * time.Second)
	retryCtx := connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID: "immutable-lifecycle-retry",
		Cause: &contractmodels.FlowExecutionCause{
			Code:      "DB_LOCK_WAIT_TIMEOUT",
			Category:  contractmodels.FlowExecutionCauseCategoryConcurrency,
			Retriable: true,
		},
	})
	retryTx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin retry caller transaction: %v", err)
	}
	defer retryTx.Rollback()
	if retried, err := repo.WithTx(retryTx).RetryWorkItem(retryCtx, workID, "lease-1", retryAt, now.Add(2*time.Second)); err != nil || !retried {
		t.Fatalf("retry WorkItem = (%v, %v)", retried, err)
	}
	assertConnectorWorkAttempt(t, retryTx, "immutable-lifecycle-retry", workID, eventID,
		connector_workqueue.WorkStateRetry, "", "DB_LOCK_WAIT_TIMEOUT", "concurrency", "", "", 1, true)
	if err := retryTx.Commit(); err != nil {
		t.Fatalf("commit retry caller transaction: %v", err)
	}
	second, claimed, err := repo.ClaimNextWorkItem(claimCtx, connector_workqueue.ClaimWorkItemInput{
		InstalledPackID: packID,
		Generation:      0,
		OwnerID:         "worker-2",
		FencingToken:    "fence-2",
		LeaseToken:      "lease-2",
		LeaseUntil:      retryAt.Add(time.Minute),
		Now:             retryAt,
	})
	if err != nil || !claimed || second == nil {
		t.Fatalf("second WorkItem claim = (%+v, %v, %v)", second, claimed, err)
	}
	failCtx := connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID:    "immutable-lifecycle-failure",
		PresentedErrorCode: "FLOW_ERROR",
		Cause: &contractmodels.FlowExecutionCause{
			Code:     "VALIDATION_FAILED",
			Category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
	})
	result, failed, err := repo.FailWorkItem(failCtx, connector_workqueue.FailWorkItemInput{
		WorkID:         workID,
		EventID:        eventID,
		LeaseToken:     "lease-2",
		FailureClass:   connector_workqueue.FailureClassInboundExecution,
		FailureSummary: connector_workqueue.FailureSummaryInbound,
		RetryThreshold: 1,
		RetryDelays:    []time.Duration{time.Second},
		Now:            retryAt.Add(time.Second),
	})
	if err != nil || !failed || result == nil || result.State != connector_workqueue.WorkStateDead {
		t.Fatalf("fail WorkItem = (%+v, %v, %v)", result, failed, err)
	}

	after, err := svc.loadByID(ctx, eventID)
	if err != nil {
		t.Fatalf("load Event after WorkItem lifecycle: %v", err)
	}
	if !reflect.DeepEqual(before, after) {
		t.Fatalf("WorkItem lifecycle mutated Event:\nbefore=%+v\nafter=%+v", before, after)
	}
	var state string
	var failureCount int
	if err := db.QueryRowContext(ctx, `
SELECT State, FailureCount
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(&state, &failureCount); err != nil {
		t.Fatalf("reload terminal WorkItem: %v", err)
	}
	if state != connector_workqueue.WorkStateDead || failureCount != 1 {
		t.Fatalf("terminal WorkItem = %s/%d, want dead/1", state, failureCount)
	}
	assertConnectorWorkAttempt(t, db, "immutable-lifecycle-failure", workID, eventID,
		connector_workqueue.WorkStateDead, "FLOW_ERROR", "VALIDATION_FAILED", "validation", "", "", 2, false)
	completeCtx := connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID:    "immutable-lifecycle-completion",
		PresentedErrorCode: "FLOW_ERROR",
		Cause: &contractmodels.FlowExecutionCause{
			Code:        "DB_LOCK_WAIT_TIMEOUT",
			Category:    contractmodels.FlowExecutionCauseCategoryConcurrency,
			ElementType: "ActionCall",
			ElementName: "ApplyMappings",
			Retriable:   true,
		},
	})
	if completed, err := repo.CompleteWorkItem(completeCtx, "complete-attempt-work", "complete-lease", connector_workqueue.WorkStateProcessed, now.Add(3*time.Second)); err != nil || !completed {
		t.Fatalf("complete WorkItem = (%v, %v)", completed, err)
	}
	assertConnectorWorkAttempt(t, db, "immutable-lifecycle-completion", "complete-attempt-work", "complete-attempt-event",
		connector_workqueue.WorkStateProcessed, "FLOW_ERROR", "DB_LOCK_WAIT_TIMEOUT", "concurrency", "ActionCall", "ApplyMappings", 3, true)
}

func TestService_SettledProjectionHistoricalGenerationLeavesLateEventAsReadinessGap(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		packInstall       = "projection-history-claim-pack"
		eventKind         = "contact.propertychange"
		laneKey           = "contact:projection-history-claim"
		historicalEventID = "projection-history-terminal"
		pendingEventID    = "projection-history-pending"
		projectionKey     = "csp:projection-history-claim"
	)
	createConnectorEventTable(t, db)
	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	for _, event := range []struct {
		id, kind, status string
		ordinal          int64
		createdAt        time.Time
	}{
		{historicalEventID, eventKind, statusProcessed, 1, now},
		// Zero ordinal exercises the historical CreatedDate/DeliverySequence/Id
		// boundary used by rows that predate durable enqueue ordinals.
		{pendingEventID, " Contact.PropertyChange ", statusPending, 0, now.Add(time.Second)},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RouteVersion, SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'projection-history-claim', ?, ?, ?, ?, 'default', 1, ?, ?, ?, ?)`,
			event.id, packInstall, event.kind, event.id, event.status, laneKey, event.ordinal, event.createdAt, now); err != nil {
			t.Fatalf("seed %s: %v", event.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        SourceLaneKey, LaneOrdinal, ProjectionKey, ProjectionActiveKey, TerminalAt,
        CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'processed', ?, ?, 1, ?, NULL, ?, ?, ?)`,
		projectionKey, historicalEventID, packInstall, now, laneKey, projectionKey, now, now, now); err != nil {
		t.Fatalf("seed historical projection generation: %v", err)
	}

	svc := NewService(db, nil, nil)
	readinessGap, err := svc.WorkItemRepository().HasNonTerminalLegacyEventWithoutWorkItem(
		ctx,
		connector_workqueue.LegacyEventMirrorReadinessInput{InstalledPackID: packInstall},
	)
	if err != nil {
		t.Fatalf("check historical projection readiness boundary: %v", err)
	}
	if !readinessGap {
		t.Fatal("late Event beyond the terminal projection boundary must stop WorkItem-only startup")
	}
	_, _, claimed, err := svc.ClaimByID(ctx, pendingEventID, "legacy-history-worker", "", sql.NullString{}, 0)
	if err != nil || !claimed {
		t.Fatalf("legacy ClaimByID after terminal projection = (claimed=%v, err=%v), want true, nil", claimed, err)
	}
}

// The current source-projection Event can be processing while a newer Event in
// the same lane waits as its successor. The successor has no direct WorkItem
// EventId yet, so this proves the legacy guard covers the whole source lane,
// not merely exact EventId mirrors.
func TestService_SettledProjectionLegacyClaimCannotBypassUnselectedSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-legacy-successor"
		packInstall = "install-token-projection-legacy-successor"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "legacy-successor-first", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "legacy-successor-first",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"legacy-successor-contact","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue first source projection Event: %v", err)
	}
	var workID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, firstIDs[0]).Scan(&workID); err != nil {
		t.Fatalf("load first source projection WorkItem: %v", err)
	}
	const leaseToken = "legacy-successor-workitem-lease"
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'processing', ClaimToken = ? WHERE Id = ?`, leaseToken, firstIDs[0]); err != nil {
		t.Fatalf("mark first Event processing: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State = 'processing', LeaseOwner = 'workitem-worker', LeaseToken = ?, LeaseUntil = DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE WorkId = ?`, leaseToken, workID); err != nil {
		t.Fatalf("mark first WorkItem processing: %v", err)
	}
	secondIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "legacy-successor-second", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "legacy-successor-second",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"legacy-successor-contact","property":"phone"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue source-projection successor: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET NextAttemptAt = NOW() WHERE Id = ?`, secondIDs[0]); err != nil {
		t.Fatalf("make source-projection successor due: %v", err)
	}

	legacyEvent, claimed, err := svc.ClaimNext(ctx, packInstall, "legacy-event-worker")
	if err != nil {
		t.Fatalf("legacy ClaimNext while successor waits: %v", err)
	}
	if claimed {
		t.Fatalf("legacy Event claimant bypassed source-projection successor %s: %#v", secondIDs[0], legacyEvent)
	}

	completed, err := svc.CompleteSourceProjectionEventForWorkItem(ctx, firstIDs[0], workID, leaseToken, statusProcessed, "", time.Now().UTC().Truncate(time.Second))
	if err != nil || !completed {
		t.Fatalf("complete first source projection Event = (%v, %v), want true, nil", completed, err)
	}
	var selectedEventID string
	if err := db.QueryRowContext(ctx, `SELECT EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey = ?`, sourceProjectionKeyFromWorkID(workID)).Scan(&selectedEventID); err != nil {
		t.Fatalf("load selected successor: %v", err)
	}
	if selectedEventID != secondIDs[0] {
		t.Fatalf("source WorkItem selected EventId=%s, want successor=%s", selectedEventID, secondIDs[0])
	}
}

func TestService_LegacyTerminalCompletionKeepsEventAuthorityBeforeTransfer(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		firstEventID  = "legacy-source-terminal-first"
		secondEventID = "legacy-source-terminal-second"
		workID        = "csp:legacy-source-terminal"
		packInstall   = "legacy-source-terminal-pack"
		eventKind     = "contact.propertychange"
		laneKey       = "contact:legacy-source-terminal"
		claimToken    = "legacy-source-terminal-lease"
	)
	createConnectorEventTable(t, db)

	ctx := context.Background()
	now := time.Date(2026, 7, 16, 3, 30, 0, 0, time.UTC)
	for _, event := range []struct {
		id, idempotencyKey, status string
		ordinal                    int64
	}{
		{firstEventID, "legacy-source-terminal-first-key", statusProcessing, 1},
		{secondEventID, "legacy-source-terminal-second-key", statusPending, 2},
	} {
		claim := any(nil)
		if event.id == firstEventID {
			claim = claimToken
		}
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        ClaimToken, RouteVersion, SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'legacy-source-terminal', ?, ?, ?, ?, 'default', ?, 1, ?, ?, ?, ?)`,
			event.id, packInstall, eventKind, event.idempotencyKey, event.status, claim, laneKey, event.ordinal, now, now); err != nil {
			t.Fatalf("seed %s: %v", event.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        SourceLaneKey, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'blocked', ?, ?, 1, ?, ?)`,
		workID, firstEventID, packInstall, now, laneKey, now, now); err != nil {
		t.Fatalf("seed blocked source projection WorkItem: %v", err)
	}

	svc := NewService(db, nil, func() time.Time { return now })
	if err := svc.MarkProcessed(ctx, firstEventID, claimToken); err != nil {
		t.Fatalf("legacy MarkProcessed before WorkItem ownership transfer: %v", err)
	}

	var firstStatus, selectedEventID, state string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, firstEventID).Scan(&firstStatus); err != nil {
		t.Fatalf("read terminal Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT EventId, State
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(&selectedEventID, &state); err != nil {
		t.Fatalf("read fenced WorkItem: %v", err)
	}
	if firstStatus != statusProcessed || selectedEventID != firstEventID || state != connector_workqueue.WorkStateBlocked {
		t.Fatalf("legacy terminal winner = Event %s, WorkItem %s/%s; want processed and original blocked compatibility row",
			firstStatus, selectedEventID, state)
	}
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem`).Scan(&count); err != nil {
		t.Fatalf("count WorkItems: %v", err)
	}
	if count != 1 {
		t.Fatalf("legacy finalizer created %d WorkItems, want original owner only", count)
	}
}

func TestService_LegacyTerminalConcurrentSourceEnqueueKeepsPreTransferEventAuthority(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	now := time.Date(2026, 7, 16, 3, 40, 0, 0, time.UTC)
	const (
		firstEventID  = "legacy-concurrent-first"
		secondEventID = "legacy-concurrent-second"
		packInstall   = "legacy-concurrent-pack"
		eventKind     = "contact.propertychange"
		laneKey       = "contact:legacy-concurrent"
		claimToken    = "legacy-concurrent-event-lease"
	)
	workID := sourceProjectionWorkID(packInstall, eventKind, 1, "", "", connector_workqueue.WorkTypeConnectorEvent, laneKey)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        ClaimToken, RouteVersion, SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'legacy-concurrent', ?, ?, 'legacy-concurrent-first-key', 'processing', 'default',
        ?, 1, ?, 1, ?, ?)`, firstEventID, packInstall, eventKind, claimToken, laneKey, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey, LaneOrdinal, InputStartOrdinal, ProjectionKey, ProjectionActiveKey, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'processing', ?, 'lost-workitem-owner',
        'lost-workitem-lease', DATE_ADD(?, INTERVAL 1 MINUTE), ?, 1, 1, ?, ?, ?, ?)`,
		workID, firstEventID, packInstall, now, now, laneKey, workID, workID, now, now); err != nil {
		t.Fatalf("seed processing source WorkItem: %v", err)
	}

	eventLocked := make(chan struct{})
	resumeTerminalization := make(chan struct{})
	markErr := make(chan error, 1)
	svc := NewService(db, nil, func() time.Time { return now })
	svc.afterLegacyTerminalEventLock = func() {
		close(eventLocked)
		<-resumeTerminalization
	}
	go func() {
		markErr <- svc.MarkProcessed(ctx, firstEventID, claimToken)
	}()
	<-eventLocked

	enqueueTx, enqueueCleanup, err := sqltx.BeginPessimisticTx(ctx, db)
	if err != nil {
		t.Fatalf("begin Event-first enqueue transaction: %v", err)
	}
	enqueueCommitted := false
	defer func() {
		if !enqueueCommitted {
			_ = enqueueTx.Rollback()
		}
		_ = enqueueCleanup()
	}()
	if _, err := enqueueTx.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RouteVersion, SourceLaneKey, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'legacy-concurrent', ?, ?, 'legacy-concurrent-second-key', 'pending', 'default',
        1, ?, 2, ?, ?)`, secondEventID, packInstall, eventKind, laneKey, now, now); err != nil {
		t.Fatalf("insert concurrent successor Event: %v", err)
	}
	created, selected, err := connector_workqueue.NewRepository(db, nil, nil).WithTx(enqueueTx).AdmitSourceProjectionWorkItem(ctx, connector_workqueue.SourceProjectionWorkItemInput{InputStartOrdinal: 2, WorkItem: connector_workqueue.CreateWorkItemInput{
		WorkID:          workID,
		EventID:         secondEventID,
		InstalledPackID: packInstall,
		WorkType:        connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:      connector_workqueue.DefaultQueueClass,
		State:           connector_workqueue.WorkStatePending,
		RunAfter:        now,
		SourceLaneKey:   laneKey,
		LaneOrdinal:     2,
		Now:             now,
	}}, true)
	if err != nil || !created || !selected {
		t.Fatalf("enqueue current processing projection = created=%t selected=%t err=%v, want true/true/nil queued successor", created, selected, err)
	}
	if err := enqueueTx.Commit(); err != nil {
		t.Fatalf("commit Event-first enqueue: %v", err)
	}
	enqueueCommitted = true

	close(resumeTerminalization)
	if err := <-markErr; err != nil {
		t.Fatalf("legacy terminal completion before ownership transfer: %v", err)
	}
	var firstStatus, selectedEventID, state string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, firstEventID).Scan(&firstStatus); err != nil {
		t.Fatalf("read terminal Event: %v", err)
	}
	var activeWorkID string
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, EventId, State
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey = ?`, sourceProjectionKeyFromWorkID(workID)).Scan(&activeWorkID, &selectedEventID, &state); err != nil {
		t.Fatalf("read active source WorkItem: %v", err)
	}
	if firstStatus != statusProcessed || activeWorkID != workID || selectedEventID != firstEventID ||
		state != connector_workqueue.WorkStateProcessing {
		t.Fatalf("concurrent compatibility state = Event %s WorkItem %s %s/%s; want Event winner with original processing WorkItem",
			firstStatus, activeWorkID, selectedEventID, state)
	}
	var queuedWorkID, queuedState string
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, State
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?
   AND WorkId <> ?`, secondEventID, workID).Scan(&queuedWorkID, &queuedState); err != nil {
		t.Fatalf("read queued successor WorkItem: %v", err)
	}
	if queuedWorkID == "" || (queuedState != connector_workqueue.WorkStateBlocked && queuedState != connector_workqueue.WorkStateBlockedRetry) {
		t.Fatalf("queued successor = %s/%s, want distinct blocked generation", queuedWorkID, queuedState)
	}
}

func TestService_LegacyOwnerTerminalizesAndConvergesPerEventWorkItem(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx := context.Background()
	now := time.Date(2026, 7, 16, 3, 45, 0, 0, time.UTC)
	const (
		eventID    = "legacy-dead-per-event"
		workID     = "cwi:legacy-dead-per-event"
		claimToken = "legacy-dead-per-event-lease"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RetryCount, ClaimToken, CreatedDate, LastModifiedDate)
VALUES (?, 'legacy-dead', 'legacy-dead-pack', 'object.changed', 'legacy-dead-key', 'processing',
        'default', ?, ?, ?, ?)`,
		eventID, MaxRetryAttempts-1, claimToken, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, CreatedDate, LastModifiedDate)
VALUES (?, ?, 'legacy-dead-pack', 'connector_event', 'default', 'pending', ?, ?, ?)`,
		workID, eventID, now, now, now); err != nil {
		t.Fatalf("seed pending per-event WorkItem: %v", err)
	}

	svc := NewService(db, nil, func() time.Time { return now })
	if err := svc.MarkFailed(ctx, eventID, claimToken, "terminal failure"); err != nil {
		t.Fatalf("legacy owner MarkFailed: %v", err)
	}
	work, claimed, err := connector_workqueue.NewRepository(db, nil, nil).ClaimNextWorkItem(
		connector_workqueue.WithDistributedClaim(ctx),
		connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: "legacy-dead-pack",
			Generation:      0,
			OwnerID:         "new-binary",
			FencingToken:    "new-binary-fence",
			LeaseToken:      "new-binary-lease",
			LeaseUntil:      now.Add(time.Minute),
			Now:             now.Add(time.Second),
		},
	)
	if err != nil || claimed || work != nil {
		t.Fatalf("terminal convergence claim = (%+v, %v, %v), want no business execution", work, claimed, err)
	}

	var eventStatus, executionOwner, workState string
	var retryCount int
	if err := db.QueryRowContext(ctx, `SELECT Status, RetryCount, ExecutionOwner FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&eventStatus, &retryCount, &executionOwner); err != nil {
		t.Fatalf("read terminal Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&workState); err != nil {
		t.Fatalf("read terminal WorkItem: %v", err)
	}
	if eventStatus != statusDead || retryCount != MaxRetryAttempts ||
		executionOwner != connector_workqueue.EventExecutionOwnerEvent ||
		workState != connector_workqueue.WorkStateDead {
		t.Fatalf("legacy terminal convergence = Event %s/%s retry=%d WorkItem=%s; want dead/event/%d/dead",
			eventStatus, executionOwner, retryCount, workState, MaxRetryAttempts)
	}
}

func TestRepository_WorkItemFailureDeadPreservesTerminalProjectionGenerationAndAttempt(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	createConnectorWorkAttemptTable(t, db)
	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	const (
		eventID    = "workitem-failure-dead-event"
		workID     = "csp:workitem-failure-dead"
		claimToken = "workitem-failure-dead-lease"
		laneKey    = "contact:workitem-failure-dead"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RetryCount, ClaimToken, RouteVersion, SourceLaneKey, EnqueueOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, 'workitem-failure', 'workitem-failure-pack', 'contact.propertychange',
        'workitem-failure-dead-key', 'processing', 'default', ?, ?, 1, ?, 1, ?, ?)`,
		eventID, MaxRetryAttempts-1, claimToken, laneKey, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, FailureCount, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey,
        Generation, LaneOrdinal, ProjectionKey, ProjectionActiveKey,
        CreatedDate, LastModifiedDate)
VALUES (?, ?, 'workitem-failure-pack', 'connector_event', 'default', 'processing', ?,
        3, ?, 'worker-1', ?, DATE_ADD(?, INTERVAL 1 MINUTE), ?,
        7, 1, ?, ?, ?, ?)`,
		workID, eventID, now, MaxRetryAttempts-1, claimToken, now, laneKey,
		workID, workID, now, now); err != nil {
		t.Fatalf("seed processing WorkItem: %v", err)
	}

	failCtx := connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID:    "projection-terminal-failure",
		PresentedErrorCode: "FLOW_ERROR",
		Cause: &contractmodels.FlowExecutionCause{
			Code:        "VALIDATION_FAILED",
			Category:    contractmodels.FlowExecutionCauseCategoryValidation,
			ElementType: "ActionCall",
			ElementName: "ApplyMappings",
		},
	})
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin caller transaction: %v", err)
	}
	defer tx.Rollback()
	result, failed, err := connector_workqueue.NewRepository(db, nil, func() time.Time { return now }).WithTx(tx).
		FailWorkItemDeferredLaneReconcile(failCtx, connector_workqueue.FailWorkItemInput{
			WorkID:         workID,
			EventID:        eventID,
			LeaseToken:     claimToken,
			FailureClass:   workItemFailureClassInboundExecution,
			FailureSummary: workItemFailureSummaryInbound,
			RetryThreshold: MaxRetryAttempts,
			RetryDelays:    workItemFailureRetryDelays(MaxRetryAttempts),
			Now:            now,
		})
	if err != nil {
		t.Fatalf("FailWorkItemDeferredLaneReconcile: %v", err)
	}
	if !failed || result == nil || result.State != connector_workqueue.WorkStateDead || result.FailureCount != MaxRetryAttempts {
		t.Fatalf("FailEventForWorkItem = (%+v, %v), want dead count %d", result, failed, MaxRetryAttempts)
	}
	// The caller-owned transaction must already contain the attempt before its
	// compatibility-mirror work can commit the complete handoff.
	assertConnectorWorkAttempt(t, tx, "projection-terminal-failure", workID, eventID,
		connector_workqueue.WorkStateDead, "FLOW_ERROR", "VALIDATION_FAILED", "validation", "ActionCall", "ApplyMappings", 3, false)
	if err := tx.Commit(); err != nil {
		t.Fatalf("commit caller transaction: %v", err)
	}

	var eventStatus, workState, failureClass, failureSummary string
	var eventRetryCount, workFailureCount int
	var projectionActiveKey sql.NullString
	var terminalAt sql.NullTime
	if err := db.QueryRowContext(ctx, `
SELECT Status, RetryCount
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID).Scan(&eventStatus, &eventRetryCount); err != nil {
		t.Fatalf("read failed Event mirror: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT State, FailureCount, FailureClass, FailureSummary, ProjectionActiveKey, TerminalAt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, workID).Scan(
		&workState, &workFailureCount, &failureClass, &failureSummary, &projectionActiveKey, &terminalAt,
	); err != nil {
		t.Fatalf("read terminal WorkItem: %v", err)
	}
	if eventStatus != statusProcessing || eventRetryCount != MaxRetryAttempts-1 ||
		workState != connector_workqueue.WorkStateDead || workFailureCount != MaxRetryAttempts ||
		failureClass != workItemFailureClassInboundExecution || failureSummary != workItemFailureSummaryInbound ||
		projectionActiveKey.Valid || !terminalAt.Valid {
		t.Fatalf("terminal failure state = immutable Event %s/%d WorkItem %s/%d class=%q summary=%q active=%v terminal=%v",
			eventStatus, eventRetryCount, workState, workFailureCount, failureClass, failureSummary, projectionActiveKey, terminalAt)
	}
}

func TestService_PerEventTerminalizationFallsBackBeforeTerminalAtSchema(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	createConnectorWorkAttemptTable(t, db)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX idx_connector_work_item_projection_history`); err != nil {
		t.Fatalf("drop TerminalAt index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX idx_connector_work_item_dlq`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP COLUMN TerminalAt`); err != nil {
		t.Fatalf("drop TerminalAt column: %v", err)
	}

	now := time.Now().UTC().Truncate(time.Second)
	const (
		eventID    = "per-event-old-terminal-schema-event"
		workID     = "per-event-old-terminal-schema-work"
		claimToken = "per-event-old-terminal-schema-lease"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RetryCount, ClaimToken, RouteVersion, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'old-terminal-schema', 'old-terminal-schema-pack', 'contact.propertychange',
        'old-terminal-schema-key', 'processing', 'default', 0, ?, 1, 1, ?, ?)`,
		eventID, claimToken, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, LeaseOwner, LeaseToken, LeaseUntil, Generation, LaneOrdinal,
        CreatedDate, LastModifiedDate)
VALUES (?, ?, 'old-terminal-schema-pack', 'connector_event', 'default', 'processing', ?,
        2, 'worker-1', ?, DATE_ADD(?, INTERVAL 1 MINUTE), 1, 1, ?, ?)`,
		workID, eventID, now, claimToken, now, now, now); err != nil {
		t.Fatalf("seed processing WorkItem: %v", err)
	}

	completeCtx := connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID: "old-schema-completion",
	})
	completed, err := NewService(db, nil, func() time.Time { return now }).CompleteEventForWorkItem(
		completeCtx, eventID, workID, claimToken, statusProcessed, "", now,
	)
	if err != nil || !completed {
		t.Fatalf("CompleteEventForWorkItem = (%v, %v), want true with old schema", completed, err)
	}

	var eventStatus, workState string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&eventStatus); err != nil {
		t.Fatalf("read terminal Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&workState); err != nil {
		t.Fatalf("read terminal WorkItem: %v", err)
	}
	if eventStatus != statusProcessing || workState != connector_workqueue.WorkStateProcessed {
		t.Fatalf("old-schema terminal state = immutable Event %s WorkItem %s, want processing/processed", eventStatus, workState)
	}
	assertConnectorWorkAttempt(t, db, "old-schema-completion", workID, eventID,
		connector_workqueue.WorkStateProcessed, "", "", "", "", "", 2, false)
}

func TestService_ListDeadWorkItemsUsesWorkItemAuthorityAndLegacyRows(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Second)
	type eventSeed struct {
		id        string
		eventKind string
		status    string
		retry     int
		createdAt time.Time
	}
	for _, seed := range []eventSeed{
		{id: "dlq-legacy-event", eventKind: "contact.updated", status: statusDead, retry: 5, createdAt: now.Add(-3 * time.Minute)},
		{id: "dlq-legacy-mirror-event", eventKind: "contact.updated", status: statusDead, retry: 4, createdAt: now.Add(-2 * time.Minute)},
		{id: "dlq-active-event", eventKind: "contact.updated", status: statusProcessing, retry: 99, createdAt: now.Add(-time.Minute)},
		{id: "dlq-workflow-event", eventKind: "workflow.action.execute", status: statusProcessing, retry: 99, createdAt: now.Add(-30 * time.Second)},
		{id: "dlq-conflict-event", eventKind: "contact.updated", status: statusDead, retry: 5, createdAt: now},
	} {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RetryCount, LastErrorAt, RouteVersion, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, 'dlq-owner', 'dlq-owner-pack', ?, CONCAT('key-', ?),
        ?, 'default', ?, ?, 1, 1, ?, ?)`,
			seed.id, seed.eventKind, seed.id, seed.status, seed.retry, seed.createdAt, seed.createdAt, seed.createdAt); err != nil {
			t.Fatalf("seed Event %s: %v", seed.id, err)
		}
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, FailureCount, FailureClass, FailureSummary, LastFailureAt, TerminalAt,
        Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES ('dlq-legacy-mirror-work', 'dlq-legacy-mirror-event', 'dlq-owner-pack', 'connector_event',
        'default', 'dead', ?, 0, 99, 'should_not_win', 'compatibility mirror',
        ?, ?, 1, 0, ?, ?),
       ('dlq-active-work', 'dlq-active-event', 'dlq-owner-pack', 'connector_event',
        'default', 'dead', ?, 1, 3, 'inbound_execution_failed', 'safe WorkItem summary',
        ?, ?, 1, 1, ?, ?),
       ('dlq-workflow-work', 'dlq-workflow-event', 'dlq-owner-pack', 'workflow_action_apply',
        'default', 'dead', ?, 1, 2, 'workflow_action_failed', 'safe workflow summary',
        ?, ?, 1, 2, ?, ?),
       ('dlq-conflict-work', 'dlq-conflict-event', 'dlq-owner-pack', 'connector_event',
        'default', 'retry', ?, 1, 2, 'inbound_execution_failed', 'retrying WorkItem',
        ?, NULL, 1, 3, ?, ?)`,
		now, now.Add(-2*time.Minute), now.Add(-2*time.Minute), now.Add(-2*time.Minute), now.Add(-2*time.Minute),
		now, now.Add(-time.Minute), now.Add(-time.Minute), now.Add(-time.Minute), now.Add(-time.Minute),
		now, now.Add(-30*time.Second), now.Add(-30*time.Second), now.Add(-30*time.Second), now.Add(-30*time.Second),
		now, now, now, now); err != nil {
		t.Fatalf("seed WorkItems: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Payload = '{"access_token":"secret"}',
       LastError = 'raw provider body',
       ClaimedBy = 'sensitive-worker',
       ClaimToken = 'sensitive-lease',
       CompanionPackInstallId = 'sensitive-companion',
       CompanionFlowVersionId = 'sensitive-flow',
       SourceLaneKey = 'contact:sensitive'
 WHERE Id IN ('dlq-legacy-mirror-event', 'dlq-active-event')`); err != nil {
		t.Fatalf("seed secret-bearing Event detail: %v", err)
	}

	svc := NewService(db, nil, func() time.Time { return now })
	items, err := svc.ListDeadWorkItems(
		ctx,
		allowPerm(string(constants.PermViewConnectorEvents)),
		"dlq-owner",
		Page{Limit: 50},
	)
	if err != nil {
		t.Fatalf("ListDeadWorkItems: %v", err)
	}
	if len(items) != 4 {
		t.Fatalf("ListDeadWorkItems len = %d, want two claimed WorkItems + two legacy Event rows", len(items))
	}
	if !items[0].LegacyEvent || items[0].EventID != "dlq-legacy-event" || items[0].FailureCount != 5 {
		t.Fatalf("legacy DLQ row = %+v", items[0])
	}
	if !items[1].LegacyEvent || items[1].WorkID != "" ||
		items[1].EventID != "dlq-legacy-mirror-event" || items[1].FailureCount != 4 ||
		items[1].FailureClass != "" || items[1].FailureSummary != "" {
		t.Fatalf("zero-attempt compatibility mirror overrode legacy Event: %+v", items[1])
	}
	active := items[2]
	if active.LegacyEvent || active.WorkID != "dlq-active-work" || active.EventID != "dlq-active-event" ||
		active.State != connector_workqueue.WorkStateDead || active.FailureCount != 3 ||
		active.FailureClass != "" || active.FailureSummary != "" || active.TerminalAt == nil {
		t.Fatalf("active WorkItem DLQ row = %+v", active)
	}
	workflow := items[3]
	if workflow.LegacyEvent || workflow.WorkID != "dlq-workflow-work" ||
		workflow.EventKind != "workflow.action.execute" || workflow.FailureCount != 2 ||
		workflow.FailureClass != "" || workflow.FailureSummary != "" {
		t.Fatalf("workflow-action WorkItem DLQ row = %+v", workflow)
	}
	legacyDetail, err := svc.GetDeadWorkItem(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dlq-legacy-mirror-event",
	)
	if err != nil {
		t.Fatalf("GetDeadWorkItem legacy mirror: %v", err)
	}
	if !legacyDetail.LegacyEvent || legacyDetail.WorkID != "" || legacyDetail.FailureCount != 4 ||
		legacyDetail.FailureClass != "" || legacyDetail.FailureSummary != "" {
		t.Fatalf("zero-attempt legacy detail = %+v", legacyDetail)
	}
	workflowDetail, err := svc.GetDeadWorkItem(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dlq-workflow-event",
	)
	if err != nil {
		t.Fatalf("GetDeadWorkItem workflow action: %v", err)
	}
	if workflowDetail.LegacyEvent || workflowDetail.WorkID != "dlq-workflow-work" ||
		workflowDetail.FailureClass != "" || workflowDetail.FailureSummary != "" {
		t.Fatalf("workflow-action WorkItem detail = %+v", workflowDetail)
	}
	for _, eventID := range []string{"dlq-legacy-mirror-event", "dlq-active-event"} {
		detail, err := svc.GetDeadEvent(
			ctx,
			allowPerm(string(constants.PermReplayConnectorEvents)),
			eventID,
		)
		if err != nil {
			t.Fatalf("GetDeadEvent %s: %v", eventID, err)
		}
		if len(detail.Payload) != 0 || detail.LastError != "" || detail.ClaimedBy != "" ||
			detail.ClaimToken != "" || detail.CompanionPackInstallID != "" ||
			detail.CompanionFlowVersionID != "" || detail.SourceLaneKey != "" {
			t.Fatalf("GetDeadEvent %s leaked Event detail: %+v", eventID, detail)
		}
	}
	for _, item := range items {
		if item.EventID == "dlq-conflict-event" {
			t.Fatalf("Event dead mirror overrode retrying WorkItem: %+v", item)
		}
	}
}

func TestService_ListDeadWorkItemsFallsBackBeforeFailureSchema(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP COLUMN FailureCount`); err != nil {
		t.Fatalf("drop FailureCount column: %v", err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        RetryCount, LastErrorAt, RouteVersion, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES ('dlq-old-schema-event', 'dlq-old-schema', 'dlq-old-schema-pack', 'contact.updated',
        'dlq-old-schema-key', 'dead', 'default', 5, ?, 1, 1, ?, ?)`,
		now, now, now); err != nil {
		t.Fatalf("seed legacy Event: %v", err)
	}

	items, err := NewService(db, nil, func() time.Time { return now }).ListDeadWorkItems(
		ctx,
		allowPerm(string(constants.PermViewConnectorEvents)),
		"dlq-old-schema",
		Page{Limit: 50},
	)
	if err != nil {
		t.Fatalf("ListDeadWorkItems: %v", err)
	}
	if len(items) != 1 || !items[0].LegacyEvent || items[0].EventID != "dlq-old-schema-event" {
		t.Fatalf("old-schema DLQ fallback = %+v", items)
	}
}

func TestService_LegacyTerminalLeaseLossLeavesMirrorUntouched(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx := context.Background()
	now := time.Date(2026, 7, 16, 3, 50, 0, 0, time.UTC)
	const (
		eventID = "legacy-terminal-lease-lost"
		workID  = "cwi:legacy-terminal-lease-lost"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass,
        ClaimToken, CreatedDate, LastModifiedDate)
VALUES (?, 'legacy-lease-lost', 'legacy-lease-lost-pack', 'object.changed', 'legacy-lease-lost-key',
        'processing', 'default', 'current-lease', ?, ?)`, eventID, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, CreatedDate, LastModifiedDate)
VALUES (?, ?, 'legacy-lease-lost-pack', 'connector_event', 'default', 'blocked', ?, ?, ?)`,
		workID, eventID, now, now, now); err != nil {
		t.Fatalf("seed blocked WorkItem: %v", err)
	}

	err := NewService(db, nil, func() time.Time { return now }).MarkProcessed(ctx, eventID, "stale-lease")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("legacy MarkProcessed stale lease = %v, want ErrLeaseLost", err)
	}
	var eventStatus, workState string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&eventStatus); err != nil {
		t.Fatalf("read Event after lease loss: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&workState); err != nil {
		t.Fatalf("read WorkItem after lease loss: %v", err)
	}
	if eventStatus != statusProcessing || workState != connector_workqueue.WorkStateBlocked {
		t.Fatalf("lease-lost terminalization mutated Event/WorkItem to %s/%s, want processing/blocked", eventStatus, workState)
	}
}

func TestService_SettledProjectionRetryKeepsImmutableEventWithSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-retry-atomic"
		packInstall = "install-token-projection-retry-atomic"
		eventKind   = "contact.propertychange"
		claimToken  = "projection-retry-token"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	repo := connector_workqueue.NewRepository(db, nil, nil)
	firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-retry-first", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-retry-first",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-retry","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue first projection event: %v", err)
	}
	firstEventID := firstIDs[0]
	var workID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, firstEventID).Scan(&workID); err != nil {
		t.Fatalf("load projection WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='processing', ClaimToken=? WHERE Id=?`, claimToken, firstEventID); err != nil {
		t.Fatalf("mark first Event processing: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State='processing', LeaseToken=?, LeaseOwner='worker', LeaseUntil=DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE WorkId=?`, claimToken, workID); err != nil {
		t.Fatalf("mark projection WorkItem processing: %v", err)
	}
	eventBefore, err := svc.loadByID(ctx, firstEventID)
	if err != nil {
		t.Fatalf("load Event before WorkItem retry: %v", err)
	}

	now := time.Now().UTC().Truncate(time.Second)
	runAfter := now.Add(30 * time.Second)
	retried, err := svc.RetrySourceProjectionEventForWorkItem(ctx, firstEventID, workID, claimToken, "retry evidence", runAfter, now, 1)
	if err != nil || !retried {
		t.Fatalf("retry projection WorkItem: retried=%v err=%v", retried, err)
	}
	var retryWorkDue time.Time
	if err := db.QueryRowContext(ctx, `SELECT RunAfter FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&retryWorkDue); err != nil {
		t.Fatalf("load projection WorkItem backoff before successor: %v", err)
	}
	secondIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-retry-second", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-retry-second",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-retry","property":"phone"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue successor after projection retry: %v", err)
	}
	secondEventID := secondIDs[0]

	eventAfter, err := svc.loadByID(ctx, firstEventID)
	if err != nil {
		t.Fatalf("reload Event after WorkItem retry: %v", err)
	}
	if !reflect.DeepEqual(eventBefore, eventAfter) {
		t.Fatalf("source projection retry mutated Event:\nbefore=%+v\nafter=%+v", eventBefore, eventAfter)
	}
	var selectedEventID, workState, leaseToken string
	var workRunAfter time.Time
	if err := db.QueryRowContext(ctx, `SELECT EventId, State, RunAfter, COALESCE(LeaseToken, '') FROM _System_ConnectorWorkItem WHERE WorkId = ?`, workID).Scan(&selectedEventID, &workState, &workRunAfter, &leaseToken); err != nil {
		t.Fatalf("load retried projection WorkItem: %v", err)
	}
	if selectedEventID != firstEventID || workState != connector_workqueue.WorkStateRetry || !workRunAfter.Equal(retryWorkDue) || leaseToken != "" {
		t.Fatalf("retry state WorkItem=%s/%s/%v/%q, want selected-first/retry/backoff/no-lease", selectedEventID, workState, workRunAfter, leaseToken)
	}
	var successorStatus string
	var successorMirrorCount int
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, secondEventID).Scan(&successorStatus); err != nil {
		t.Fatalf("load retry successor Event: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE EventId = ?`, secondEventID).Scan(&successorMirrorCount); err != nil {
		t.Fatalf("count retry successor mirrors: %v", err)
	}
	readinessGap, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(ctx, connector_workqueue.LegacyEventMirrorReadinessInput{InstalledPackID: packInstall})
	if err != nil {
		t.Fatalf("check retry successor readiness: %v", err)
	}
	if successorStatus != statusPending || successorMirrorCount != 1 || readinessGap {
		t.Fatalf("retry successor status/queued-work/readiness = %s/%d/%v, want pending/1/false", successorStatus, successorMirrorCount, readinessGap)
	}
	claimNow := time.Now().UTC().Truncate(time.Second)
	if !claimNow.Before(runAfter) {
		t.Fatalf("retry backoff elapsed before inactive-successor proof: claim=%v retry=%v", claimNow, runAfter)
	}
	claimedWork, claimed, err := repo.ClaimNextWorkItem(
		connector_workqueue.WithDistributedClaim(ctx),
		connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: packInstall,
			Generation:      0,
			OwnerID:         "projection-retry-probe",
			FencingToken:    "projection-retry-probe-fence",
			LeaseToken:      "projection-retry-probe-token",
			LeaseUntil:      claimNow.Add(time.Minute),
			Now:             claimNow,
		},
	)
	if err != nil || claimed || claimedWork != nil {
		t.Fatalf("claim while projection retry owns inactive successor = (%+v, %v, %v), want no claim", claimedWork, claimed, err)
	}
	t.Log("evidence post-retry enqueue preserved immutable Event provenance while WorkItem backoff and successor readiness remained authoritative")
}

func TestService_SettledProjectionRejectsTargetMutationLane(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-target-lane"
		packInstall = "install-token-projection-target-lane"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	_, _, err := NewService(db, nil, nil).EnqueueBatch(context.Background(), namespace, packInstall, "projection-target-lane-1", []BatchEvent{{
		EventKind:                    eventKind,
		IdempotencyKey:               "projection-target-lane-event",
		Payload:                      json.RawMessage(`{"objectKind":"contact","sourceID":"contact-target-lane"}`),
		RouteVersion:                 1,
		TargetMutationLaneObjectKind: "account",
		TargetMutationLaneID:         "account-target-lane",
	}})
	if err == nil {
		t.Fatal("enqueue settled source projection with TargetMutationLane succeeded, want rejection")
	}

	var eventCount, workItemCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorEvent`).Scan(&eventCount); err != nil {
		t.Fatalf("count rejected projection events: %v", err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorWorkItem`).Scan(&workItemCount); err != nil {
		t.Fatalf("count rejected projection WorkItems: %v", err)
	}
	if eventCount != 0 || workItemCount != 0 {
		t.Fatalf("rejected settled source projection persisted events=%d workitems=%d", eventCount, workItemCount)
	}
}

func TestService_SettledProjectionWithoutSourceLanePreservesTargetMutationLane(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-target-lane-fallback"
		packInstall = "install-token-projection-target-lane-fallback"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	targetIDs := []string{"account-target-lane-a", "account-target-lane-b"}
	batch := make([]BatchEvent, 0, len(targetIDs))
	for _, targetID := range targetIDs {
		batch = append(batch, BatchEvent{
			EventKind:                    eventKind,
			IdempotencyKey:               "projection-target-lane-fallback-" + targetID,
			Payload:                      json.RawMessage(`{"property":"email"}`),
			RouteVersion:                 1,
			TargetMutationLaneObjectKind: "account",
			TargetMutationLaneID:         targetID,
		})
	}
	ids, _, err := NewService(db, nil, nil).EnqueueBatch(ctx, namespace, packInstall, "projection-target-lane-fallback-1", batch)
	if err != nil {
		t.Fatalf("enqueue settled projection events without source lane: %v", err)
	}
	if len(ids) != len(targetIDs) {
		t.Fatalf("enqueue ids=%d, want %d", len(ids), len(targetIDs))
	}

	for i, eventID := range ids {
		var workID, targetLaneKey string
		if err := db.QueryRowContext(ctx, `SELECT WorkId, COALESCE(TargetMutationLaneKey, '') FROM _System_ConnectorWorkItem WHERE EventId = ?`, eventID).Scan(&workID, &targetLaneKey); err != nil {
			t.Fatalf("load fallback WorkItem for event %s: %v", eventID, err)
		}
		if connector_workqueue.IsSourceProjectionWorkID(workID) {
			t.Fatalf("fallback event %s materialized source projection WorkItem %s", eventID, workID)
		}
		if want := sourceLaneKeyFor("account", targetIDs[i]); targetLaneKey != want {
			t.Fatalf("fallback WorkItem target lane=%s, want %s", targetLaneKey, want)
		}
	}
}

func TestService_SettledProjectionReplayCreatesDistinctDurableGeneration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay"
		packInstall = "install-token-projection-replay"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-1", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue settled projection event: %v", err)
	}
	originalID := ids[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalID); err != nil {
		t.Fatalf("mark projection event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalID); err != nil {
		t.Fatalf("mark projection WorkItem dead: %v", err)
	}

	replayID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalID, packInstall, 1)
	if err != nil {
		t.Fatalf("replay settled projection event: %v", err)
	}
	var originalWorkID, projectionKey, activeWorkID, selectedEventID, replayOfWorkID string
	var originalReplayedAt sql.NullTime
	var workItemCount, standaloneCount int
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, COALESCE(ProjectionKey, WorkId), ReplayedAt
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, originalID).Scan(&originalWorkID, &projectionKey, &originalReplayedAt); err != nil {
		t.Fatalf("load replay projection identity: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, EventId, COALESCE(ReplayOfWorkId, '')
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey = ?`, projectionKey).Scan(&activeWorkID, &selectedEventID, &replayOfWorkID); err != nil {
		t.Fatalf("load active replay WorkItem: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ?`, packInstall).Scan(&workItemCount); err != nil {
		t.Fatalf("count replay WorkItems: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND WorkId LIKE 'cwi:%'`, packInstall).Scan(&standaloneCount); err != nil {
		t.Fatalf("count replay standalone WorkItems: %v", err)
	}
	if !connector_workqueue.IsSourceProjectionWorkID(activeWorkID) || activeWorkID == projectionKey ||
		selectedEventID != replayID || replayOfWorkID != originalWorkID || !originalReplayedAt.Valid ||
		workItemCount != 2 || standaloneCount != 0 {
		t.Fatalf("replay materialization original=%s projection=%s active=%s event=%s replayOf=%s originalReplayed=%v total=%d standalone=%d, want historical + distinct csp replay=%s",
			originalWorkID, projectionKey, activeWorkID, selectedEventID, replayOfWorkID, originalReplayedAt.Valid, workItemCount, standaloneCount, replayID)
	}
}

func TestService_SettledProjectionReplayIgnoresSupersededEventWithoutDirectWorkItem(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-superseded"
		packInstall = "install-token-projection-replay-superseded"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-superseded-original", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-superseded-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-superseded"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection Event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("terminalize original Event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6),
       ProjectionActiveKey = NULL
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("terminalize original WorkItem: %v", err)
	}

	successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-superseded-successors", []BatchEvent{
		{
			EventKind: eventKind, IdempotencyKey: "projection-replay-superseded-a",
			Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-superseded","value":"a"}`), RouteVersion: 1,
		},
		{
			EventKind: eventKind, IdempotencyKey: "projection-replay-superseded-b",
			Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-superseded","value":"b"}`), RouteVersion: 1,
		},
	})
	if err != nil {
		t.Fatalf("enqueue coalesced successor Events: %v", err)
	}
	if len(successorIDs) != 2 {
		t.Fatalf("successor Event count = %d, want 2", len(successorIDs))
	}

	var successorWorkID, selectedEventID string
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, EventId
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey IS NOT NULL`).Scan(&successorWorkID, &selectedEventID); err != nil {
		t.Fatalf("load active successor projection: %v", err)
	}
	if selectedEventID != successorIDs[1] {
		t.Fatalf("selected successor Event = %s, want latest %s", selectedEventID, successorIDs[1])
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET EnqueueOrdinal = 0,
       CreatedDate = DATE_ADD(CreatedDate, INTERVAL 1 SECOND)
 WHERE Id = ?`, selectedEventID); err != nil {
		t.Fatalf("seed mixed historical ordering for selected successor: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'processed',
       Attempt = 1,
       TerminalAt = NOW(6),
       ProjectionActiveKey = NULL
 WHERE WorkId = ?`, successorWorkID); err != nil {
		t.Fatalf("terminalize coalesced successor WorkItem: %v", err)
	}

	var supersededOwners int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, successorIDs[0]).Scan(&supersededOwners); err != nil {
		t.Fatalf("count superseded direct WorkItems: %v", err)
	}
	if supersededOwners != 0 {
		t.Fatalf("superseded Event has %d direct WorkItems, want 0", supersededOwners)
	}

	replayID, err := svc.ReplayDeadEvent(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		originalEventID,
		packInstall,
		1,
	)
	if err != nil || replayID == "" {
		t.Fatalf("replay after terminal coalescing = (%q, %v), want replay child", replayID, err)
	}
}

func TestService_ZeroAttemptSourceProjectionReplayRollsBackWhenReplaySchemaIncomplete(t *testing.T) {
	for _, missingColumn := range []string{"TerminalAt", "ProjectionActiveKey"} {
		t.Run(missingColumn, func(t *testing.T) {
			assertZeroAttemptSourceProjectionReplayRollsBackWhenReplaySchemaIncomplete(t, missingColumn)
		})
	}
}

func assertZeroAttemptSourceProjectionReplayRollsBackWhenReplaySchemaIncomplete(t *testing.T, missingColumn string) {
	t.Helper()
	db, cleanup := openTiDB(t)
	defer cleanup()

	namespace := "projection-replay-incomplete-" + strings.ToLower(missingColumn)
	packInstall := "install-token-" + namespace
	const eventKind = "contact.propertychange"
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, namespace+"-1", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: namespace + "-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-incomplete-schema"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue settled projection event: %v", err)
	}
	originalEventID := ids[0]
	var originalWorkID, originalWorkState, originalProjectionActiveKey string
	var originalAttempt int
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, State, Attempt, ProjectionActiveKey
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, originalEventID).Scan(
		&originalWorkID, &originalWorkState, &originalAttempt, &originalProjectionActiveKey,
	); err != nil {
		t.Fatalf("load zero-attempt source projection mirror: %v", err)
	}
	if originalAttempt != 0 {
		t.Fatalf("source projection mirror attempt = %d, want zero-attempt compatibility mirror", originalAttempt)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark source projection Event dead: %v", err)
	}
	indexName := "idx_connector_work_item_projection_history"
	if missingColumn == "ProjectionActiveKey" {
		indexName = "uniq_connector_work_item_projection_active"
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX `+indexName); err != nil {
		t.Fatalf("drop %s index: %v", missingColumn, err)
	}
	if missingColumn == "TerminalAt" {
		if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX idx_connector_work_item_dlq`); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorWorkItem DROP COLUMN `+missingColumn); err != nil {
		t.Fatalf("drop %s column: %v", missingColumn, err)
	}

	replayEventID, err := svc.ReplayDeadEvent(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		originalEventID,
		packInstall,
		1,
	)
	if err == nil {
		t.Fatal("ReplayDeadEvent error = nil, want incomplete replay WorkItem schema error")
	}
	if replayEventID != "" {
		t.Fatalf("ReplayDeadEvent id = %q, want empty after transaction rollback", replayEventID)
	}

	var eventStatus, workState string
	var eventReplayedAt, workReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `
SELECT Status, ReplayedAt
  FROM _System_ConnectorEvent
 WHERE Id = ?`, originalEventID).Scan(&eventStatus, &eventReplayedAt); err != nil {
		t.Fatalf("load original Event after failed replay: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT State, ReplayedAt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, originalWorkID).Scan(&workState, &workReplayedAt); err != nil {
		t.Fatalf("load original WorkItem after failed replay: %v", err)
	}
	projectionActiveRestored := true
	if missingColumn != "ProjectionActiveKey" {
		var projectionActiveKey string
		if err := db.QueryRowContext(ctx, `
SELECT ProjectionActiveKey
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, originalWorkID).Scan(&projectionActiveKey); err != nil {
			t.Fatalf("load original projection activity after failed replay: %v", err)
		}
		projectionActiveRestored = projectionActiveKey == originalProjectionActiveKey
	}
	var replayEventCount, replayWorkItemCount int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE ReplayOfId = ?`, originalEventID).Scan(&replayEventCount); err != nil {
		t.Fatalf("count rolled-back replay Events: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem
 WHERE ReplayOfWorkId = ?`, originalWorkID).Scan(&replayWorkItemCount); err != nil {
		t.Fatalf("count rolled-back replay WorkItems: %v", err)
	}
	if eventStatus != statusDead || eventReplayedAt.Valid ||
		workState != originalWorkState || workReplayedAt.Valid ||
		!projectionActiveRestored ||
		replayEventCount != 0 || replayWorkItemCount != 0 {
		t.Fatalf(
			"failed replay with missing %s left Event=(%s,replayed=%v) WorkItem=(%s,replayed=%v,active-restored=%v) children=(%d,%d), want original dead/%s reservation restored and no children",
			missingColumn,
			eventStatus, eventReplayedAt.Valid,
			workState, workReplayedAt.Valid, projectionActiveRestored,
			replayEventCount, replayWorkItemCount, originalWorkState,
		)
	}
}

func TestService_WorkItemOwnedReplayCreatesDistinctLineageGeneration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "workitem-replay"
		packInstall = "install-token-workitem-replay"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(ctx, namespace, "webhook.event", "workitem-replay-original", json.RawMessage(`{"value":"original"}`), packInstall, 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("enqueue original event: %v", err)
	}
	var originalWorkID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&originalWorkID); err != nil {
		t.Fatalf("load original WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = 'processing',
       ClaimedBy = 'stale-event-worker',
       ClaimToken = 'stale-event-token'
 WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("seed drifted Event mirror: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE WorkId = ?`, originalWorkID); err != nil {
		t.Fatalf("mark original WorkItem dead: %v", err)
	}

	replayEventID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}

	var originalState, originalEventStatus, replayWorkID, replayState, replayOfWorkID, replayOfEventID string
	var originalReplayedAt sql.NullTime
	var originalEventClaimToken sql.NullString
	if err := db.QueryRowContext(ctx, `
SELECT State, ReplayedAt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, originalWorkID).Scan(&originalState, &originalReplayedAt); err != nil {
		t.Fatalf("load original terminal WorkItem: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, State, COALESCE(ReplayOfWorkId, '')
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, replayEventID).Scan(&replayWorkID, &replayState, &replayOfWorkID); err != nil {
		t.Fatalf("load replay WorkItem: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COALESCE(ReplayOfId, '') FROM _System_ConnectorEvent WHERE Id = ?`, replayEventID).Scan(&replayOfEventID); err != nil {
		t.Fatalf("load replay Event lineage: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT Status, ClaimToken
  FROM _System_ConnectorEvent
 WHERE Id = ?`, originalEventID).Scan(&originalEventStatus, &originalEventClaimToken); err != nil {
		t.Fatalf("load repaired Event mirror: %v", err)
	}
	if originalState != connector_workqueue.WorkStateDead || !originalReplayedAt.Valid ||
		replayWorkID == originalWorkID || replayState != connector_workqueue.WorkStatePending ||
		replayOfWorkID != originalWorkID || replayOfEventID != originalEventID ||
		originalEventStatus != statusProcessing || !originalEventClaimToken.Valid {
		t.Fatalf("replay lineage original=(%s,%s,%v,event=%s,claim=%v) replay=(%s,%s,%s,event=%s), want immutable terminal WorkItem, unchanged Event provenance, and distinct pending child",
			originalWorkID, originalState, originalReplayedAt.Valid, originalEventStatus, originalEventClaimToken.Valid,
			replayWorkID, replayState, replayOfWorkID, replayOfEventID)
	}
}

func TestService_WorkItemSourceProjectionCompletionLeavesEventImmutableAndReplayable(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-complete-replay-lock-order"
		packInstall = "install-token-projection-complete-replay-lock-order"
		eventKind   = "contact.propertychange"
		claimToken  = "projection-complete-replay-lease"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	now := time.Now().UTC().Truncate(time.Second)
	svc := NewService(db, nil, func() time.Time { return now })
	ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-lock-order", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-lock-order-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"projection-lock-order-contact"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue source projection: %v", err)
	}
	originalEventID := ids[0]
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET RunAfter = ?
 WHERE EventId = ?`, now, originalEventID); err != nil {
		t.Fatalf("make source projection WorkItem due: %v", err)
	}
	work, claimed, err := svc.WorkItemRepository().ClaimNextWorkItem(
		connector_workqueue.WithDistributedClaim(ctx),
		connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: packInstall,
			Generation:      0,
			OwnerID:         "projection-owner",
			FencingToken:    "projection-fence",
			LeaseToken:      claimToken,
			LeaseUntil:      now.Add(time.Minute),
			Now:             now,
		},
	)
	if err != nil || !claimed || work == nil || work.EventID != originalEventID {
		t.Fatalf("claim source projection WorkItem = (%+v, %v, %v)", work, claimed, err)
	}
	eventBefore, err := svc.loadByID(ctx, originalEventID)
	if err != nil {
		t.Fatalf("load transferred Event: %v", err)
	}
	completed, err := svc.CompleteSourceProjectionEventForWorkItem(
		ctx,
		originalEventID,
		work.WorkID,
		claimToken,
		statusDead,
		"terminal failure",
		now,
	)
	if err != nil || !completed {
		t.Fatalf("source projection completion = (%v, %v)", completed, err)
	}
	eventAfter, err := svc.loadByID(ctx, originalEventID)
	if err != nil {
		t.Fatalf("reload Event after WorkItem completion: %v", err)
	}
	if !reflect.DeepEqual(eventBefore, eventAfter) {
		t.Fatalf("source projection completion mutated Event:\nbefore=%+v\nafter=%+v", eventBefore, eventAfter)
	}
	replayEventID, err := svc.ReplayDeadEvent(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		originalEventID,
		packInstall,
		1,
	)
	if err != nil || replayEventID == "" {
		t.Fatalf("replay after WorkItem completion = (%q, %v), want a replay child", replayEventID, err)
	}
}

func TestService_ReplayLocksActivePackBeforeEventAndWorkItem(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "replay-pack-lock-order"
		packInstall = "install-token-replay-pack-lock-order"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(
		ctx,
		namespace,
		"webhook.event",
		"replay-pack-lock-order-original",
		nil,
		packInstall,
		1,
		"",
		"",
		0,
		"",
	)
	if err != nil {
		t.Fatalf("enqueue original event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("mark original WorkItem dead: %v", err)
	}

	packTx, packCleanup, err := sqltx.BeginPessimisticTx(ctx, db)
	if err != nil {
		t.Fatalf("begin pack-lock transaction: %v", err)
	}
	defer packCleanup()
	defer packTx.Rollback()
	if err := lockActivePackIncarnation(ctx, packTx, namespace, packInstall); err != nil {
		t.Fatalf("lock active pack: %v", err)
	}

	replayBeforeEventLock := make(chan struct{})
	svc.beforeReplayEventLock = func() {
		close(replayBeforeEventLock)
	}
	type replayResult struct {
		eventID string
		err     error
	}
	replayDone := make(chan replayResult, 1)
	go func() {
		replayEventID, replayErr := svc.ReplayDeadEvent(
			ctx,
			allowPerm(string(constants.PermReplayConnectorEvents)),
			originalEventID,
			packInstall,
			1,
		)
		replayDone <- replayResult{eventID: replayEventID, err: replayErr}
	}()

	select {
	case <-replayBeforeEventLock:
		t.Fatal("replay reached the Event lock while another transaction held the active-pack row")
	case <-time.After(100 * time.Millisecond):
	}
	if err := packTx.Rollback(); err != nil {
		t.Fatalf("release active-pack lock: %v", err)
	}
	<-replayBeforeEventLock
	result := <-replayDone
	if result.err != nil || result.eventID == "" {
		t.Fatalf("replay after active-pack release = (%q, %v), want a replay child", result.eventID, result.err)
	}
}

func TestService_SourceProjectionRetryWinsWorkItemCASAndKeepsEventImmutable(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-retry-complete-lock-order"
		packInstall = "install-token-projection-retry-complete-lock-order"
		eventKind   = "contact.propertychange"
		claimToken  = "projection-retry-complete-lease"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	now := time.Now().UTC().Truncate(time.Second)
	runAfter := now.Add(30 * time.Second)
	svc := NewService(db, nil, func() time.Time { return now })
	ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-retry-complete-lock-order", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-retry-complete-lock-order-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"projection-retry-complete-contact"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue source projection: %v", err)
	}
	eventID := ids[0]
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET RunAfter = ?
 WHERE EventId = ?`, now, eventID); err != nil {
		t.Fatalf("make source projection WorkItem due: %v", err)
	}
	work, claimed, err := svc.WorkItemRepository().ClaimNextWorkItem(
		connector_workqueue.WithDistributedClaim(ctx),
		connector_workqueue.ClaimWorkItemInput{
			InstalledPackID: packInstall,
			Generation:      0,
			OwnerID:         "projection-owner",
			FencingToken:    "projection-fence",
			LeaseToken:      claimToken,
			LeaseUntil:      now.Add(time.Minute),
			Now:             now,
		},
	)
	if err != nil || !claimed || work == nil || work.EventID != eventID {
		t.Fatalf("claim source projection WorkItem = (%+v, %v, %v)", work, claimed, err)
	}
	eventBefore, err := svc.loadByID(ctx, eventID)
	if err != nil {
		t.Fatalf("load Event after ownership transfer: %v", err)
	}
	retried, err := svc.RetrySourceProjectionEventForWorkItem(
		ctx,
		eventID,
		work.WorkID,
		claimToken,
		"retry evidence",
		runAfter,
		now,
		1,
	)
	if err != nil || !retried {
		t.Fatalf("source projection retry = (%v, %v), want true, nil", retried, err)
	}
	completed, err := svc.CompleteSourceProjectionEventForWorkItem(
		ctx,
		eventID,
		work.WorkID,
		claimToken,
		statusProcessed,
		"",
		now,
	)
	if err != nil || completed {
		t.Fatalf("completion after retry = (%v, %v), want false, nil", completed, err)
	}

	eventAfter, err := svc.loadByID(ctx, eventID)
	if err != nil {
		t.Fatalf("reload Event after WorkItem retry: %v", err)
	}
	if !reflect.DeepEqual(eventBefore, eventAfter) {
		t.Fatalf("source projection retry mutated Event:\nbefore=%+v\nafter=%+v", eventBefore, eventAfter)
	}
	var workState string
	if err := db.QueryRowContext(ctx, `SELECT State FROM _System_ConnectorWorkItem WHERE WorkId = ?`, work.WorkID).Scan(&workState); err != nil {
		t.Fatalf("load retried WorkItem: %v", err)
	}
	if workState != connector_workqueue.WorkStateRetry {
		t.Fatalf("retry winner WorkItem state = %s, want retry", workState)
	}
}

func TestService_WorkItemOwnedConcurrentReplayHasOneWinner(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "workitem-replay-race"
		packInstall = "install-token-workitem-replay-race"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(ctx, namespace, "webhook.event", "workitem-replay-race-original", nil, packInstall, 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("enqueue original event: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("mark original WorkItem dead: %v", err)
	}

	type replayResult struct {
		id  string
		err error
	}
	results := make(chan replayResult, 2)
	var start sync.WaitGroup
	start.Add(2)
	for i := 0; i < 2; i++ {
		go func() {
			start.Done()
			start.Wait()
			id, replayErr := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
			results <- replayResult{id: id, err: replayErr}
		}()
	}

	successes := 0
	conflicts := 0
	for i := 0; i < 2; i++ {
		result := <-results
		switch {
		case result.err == nil && result.id != "":
			successes++
		case errors.Is(result.err, ErrEventNotDead):
			conflicts++
		default:
			t.Fatalf("unexpected replay result: id=%q err=%v", result.id, result.err)
		}
	}
	var replayChildren int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE ReplayOfWorkId IS NOT NULL`).Scan(&replayChildren); err != nil {
		t.Fatalf("count replay children: %v", err)
	}
	if successes != 1 || conflicts != 1 || replayChildren != 1 {
		t.Fatalf("concurrent replay successes=%d conflicts=%d children=%d, want 1/1/1", successes, conflicts, replayChildren)
	}
}

func TestService_ProjectionReplayBlocksOlderRouteGenerationInSameSourceScope(t *testing.T) {
	for _, tt := range []struct {
		name                 string
		removeSuccessorOwner bool
	}{
		{name: "WorkItem-owned successor"},
		{name: "Event-only successor", removeSuccessorOwner: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()

			const (
				namespace   = "projection-replay-route-rotation"
				packInstall = "install-token-projection-replay-route-rotation"
				eventKind   = "contact.propertychange"
			)
			createInstalledPackTable(t, db, namespace)
			createConnectorEventTable(t, db)
			createConnectorInboundRouteTable(t, db, namespace)
			makeSettledProjectionRoute(t, db, namespace, eventKind)

			ctx := context.Background()
			svc := NewService(db, nil, nil)
			originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-route-original", []BatchEvent{{
				EventKind: eventKind, IdempotencyKey: "projection-replay-route-original",
				Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-route"}`), RouteVersion: 1,
			}})
			if err != nil {
				t.Fatalf("enqueue original projection Event: %v", err)
			}
			originalEventID := originalIDs[0]
			if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
				t.Fatalf("mark original Event dead: %v", err)
			}
			if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6),
       ProjectionActiveKey = NULL
 WHERE EventId = ?`, originalEventID); err != nil {
				t.Fatalf("mark original WorkItem dead: %v", err)
			}

			successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-route-successor", []BatchEvent{{
				EventKind: eventKind, IdempotencyKey: "projection-replay-route-successor",
				Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-route","value":"newer"}`), RouteVersion: 1,
			}})
			if err != nil {
				t.Fatalf("enqueue route-v1 successor: %v", err)
			}
			successorEventID := successorIDs[0]
			if tt.removeSuccessorOwner {
				if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorWorkItem WHERE EventId = ?`, successorEventID); err != nil {
					t.Fatalf("remove successor WorkItem: %v", err)
				}
			}

			if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version = 1`, packInstall, eventKind); err != nil {
				t.Fatalf("retire route v1: %v", err)
			}
			if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorInboundRoute
       (Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName,
        InvocationMode, QueueClass, IsCurrent, IsDeleted)
VALUES ('cir-projection-replay-route-v2', ?, ?, ?, 2, 'TestHandlerV2', ?, 'bulk', 1, 0)`,
				namespace, packInstall, eventKind,
				connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
			); err != nil {
				t.Fatalf("publish route v2: %v", err)
			}

			_, err = svc.ReplayDeadEvent(
				ctx,
				allowPerm(string(constants.PermReplayConnectorEvents)),
				originalEventID,
				packInstall,
				2,
			)
			if !errors.Is(err, ErrReplayBlocked) {
				t.Fatalf("ReplayDeadEvent across route identity = %v, want ErrReplayBlocked", err)
			}

			var eventReplayedAt, workReplayedAt sql.NullTime
			if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
				t.Fatalf("load original Event reservation: %v", err)
			}
			if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&workReplayedAt); err != nil {
				t.Fatalf("load original WorkItem reservation: %v", err)
			}
			var replayChildren int
			if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, originalEventID).Scan(&replayChildren); err != nil {
				t.Fatalf("count replay children: %v", err)
			}
			var successorStatus string
			if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, successorEventID).Scan(&successorStatus); err != nil {
				t.Fatalf("load successor Event: %v", err)
			}
			if eventReplayedAt.Valid || workReplayedAt.Valid || replayChildren != 0 || successorStatus != statusPending {
				t.Fatalf(
					"cross-identity replay mutated reservation/successor: event=%v work=%v children=%d successor=%s",
					eventReplayedAt.Valid, workReplayedAt.Valid, replayChildren, successorStatus,
				)
			}
		})
	}
}

func TestService_SourceProjectionReplayPreservesExecutionClass(t *testing.T) {
	for _, tt := range []struct {
		name                string
		eventOnly           bool
		deleteOriginalRoute bool
	}{
		{name: "WorkItem-owned source projection"},
		{name: "Event-only source projection", eventOnly: true},
		{name: "Event-only missing original route proof", eventOnly: true, deleteOriginalRoute: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()

			const (
				namespace   = "projection-replay-class"
				packInstall = "install-token-projection-replay-class"
				eventKind   = "contact.propertychange"
			)
			createInstalledPackTable(t, db, namespace)
			createConnectorEventTable(t, db)
			createConnectorInboundRouteTable(t, db, namespace)
			makeSettledProjectionRoute(t, db, namespace, eventKind)

			ctx := context.Background()
			svc := NewService(db, nil, nil)
			originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-class-original", []BatchEvent{{
				EventKind: eventKind, IdempotencyKey: "projection-replay-class-original",
				Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-class"}`), RouteVersion: 1,
			}})
			if err != nil {
				t.Fatalf("enqueue original source projection: %v", err)
			}
			originalEventID := originalIDs[0]
			if tt.eventOnly {
				if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID); err != nil {
					t.Fatalf("remove original WorkItem: %v", err)
				}
			} else if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6),
       ProjectionActiveKey = NULL
 WHERE EventId = ?`, originalEventID); err != nil {
				t.Fatalf("terminalize original WorkItem: %v", err)
			}
			if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
				t.Fatalf("terminalize original Event: %v", err)
			}

			if tt.deleteOriginalRoute {
				if _, err := db.ExecContext(ctx, `
DELETE FROM _System_ConnectorInboundRoute
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version = 1`, packInstall, eventKind); err != nil {
					t.Fatalf("delete original route proof: %v", err)
				}
			} else if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version = 1`, packInstall, eventKind); err != nil {
				t.Fatalf("retire original route: %v", err)
			}
			if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorInboundRoute
       (Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName,
        InvocationMode, QueueClass, IsCurrent, IsDeleted)
VALUES ('cir-projection-replay-class-v2', ?, ?, ?, 2, 'PerEventHandler',
        'per_event', 'default', 1, 0)`,
				namespace, packInstall, eventKind,
			); err != nil {
				t.Fatalf("publish per-event route v2: %v", err)
			}

			_, err = svc.ReplayDeadEvent(
				ctx,
				allowPerm(string(constants.PermReplayConnectorEvents)),
				originalEventID,
				packInstall,
				2,
			)
			if !errors.Is(err, ErrReplayBlocked) {
				t.Fatalf("ReplayDeadEvent after route-mode change = %v, want ErrReplayBlocked", err)
			}

			var eventReplayedAt sql.NullTime
			if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
				t.Fatalf("load original Event reservation: %v", err)
			}
			if eventReplayedAt.Valid {
				t.Fatal("route-mode conflict consumed Event replay reservation")
			}
			if !tt.eventOnly {
				var workReplayedAt sql.NullTime
				if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&workReplayedAt); err != nil {
					t.Fatalf("load original WorkItem reservation: %v", err)
				}
				if workReplayedAt.Valid {
					t.Fatal("route-mode conflict consumed WorkItem replay reservation")
				}
			}
			var replayChildren int
			if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, originalEventID).Scan(&replayChildren); err != nil {
				t.Fatalf("count replay children: %v", err)
			}
			if replayChildren != 0 {
				t.Fatalf("route-mode conflict created %d replay children, want 0", replayChildren)
			}
		})
	}
}

func TestService_WorkItemOwnedProjectionReplayBlockedByActiveSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-blocked"
		packInstall = "install-token-projection-replay-blocked"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-blocked-original", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-blocked-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-blocked"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("mark original WorkItem dead: %v", err)
	}
	successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-blocked-successor", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-blocked-successor",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-blocked","value":"newer"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue active successor: %v", err)
	}

	_, err = svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReplayDeadEvent error = %v, want ErrReplayBlocked", err)
	}
	var eventReplayedAt, workReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
		t.Fatalf("load original Event replay reservation: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&workReplayedAt); err != nil {
		t.Fatalf("load original WorkItem replay reservation: %v", err)
	}
	var activeEventID string
	if err := db.QueryRowContext(ctx, `SELECT EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey IS NOT NULL`).Scan(&activeEventID); err != nil {
		t.Fatalf("load active projection successor: %v", err)
	}
	if eventReplayedAt.Valid || workReplayedAt.Valid || activeEventID != successorIDs[0] {
		t.Fatalf("blocked replay consumed reservation event=%v work=%v active=%s, want unconsumed original and successor=%s",
			eventReplayedAt.Valid, workReplayedAt.Valid, activeEventID, successorIDs[0])
	}
}

func TestService_WorkItemOwnedProjectionReplayBlockedByEventOnlySuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-event-only-successor"
		packInstall = "install-token-projection-replay-event-only-successor"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-event-only-owner", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-event-only-owner",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-event-only-successor","value":"old"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection Event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("mark original WorkItem dead: %v", err)
	}
	successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-event-only-newer", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-event-only-newer",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-event-only-successor","value":"newer"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue newer projection Event: %v", err)
	}
	successorEventID := successorIDs[0]
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorWorkItem WHERE EventId = ?`, successorEventID); err != nil {
		t.Fatalf("remove newer Event WorkItem to emulate compatibility gap: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET EventKind = ' Contact.PropertyChange ' WHERE Id = ?`, successorEventID); err != nil {
		t.Fatalf("restore pre-canonical EventKind shape on newer Event: %v", err)
	}

	_, err = svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReplayDeadEvent error = %v, want ErrReplayBlocked", err)
	}
	var eventReplayedAt, workReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
		t.Fatalf("load original Event replay reservation: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&workReplayedAt); err != nil {
		t.Fatalf("load original WorkItem replay reservation: %v", err)
	}
	var successorStatus string
	var successorPayload []byte
	if err := db.QueryRowContext(ctx, `SELECT Status, Payload FROM _System_ConnectorEvent WHERE Id = ?`, successorEventID).Scan(&successorStatus, &successorPayload); err != nil {
		t.Fatalf("load newer Event after blocked replay: %v", err)
	}
	var replayChildren int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, originalEventID).Scan(&replayChildren); err != nil {
		t.Fatalf("count replay children: %v", err)
	}
	if eventReplayedAt.Valid || workReplayedAt.Valid || successorStatus != statusPending ||
		!strings.Contains(string(successorPayload), `"value": "newer"`) && !strings.Contains(string(successorPayload), `"value":"newer"`) ||
		replayChildren != 0 {
		t.Fatalf("blocked Event-only successor replay eventReserved=%v workReserved=%v successor=%s payload=%s children=%d",
			eventReplayedAt.Valid, workReplayedAt.Valid, successorStatus, successorPayload, replayChildren)
	}
}

func TestService_ZeroAttemptProjectionReplayBlockedByActiveSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-zero-attempt"
		packInstall = "install-token-projection-replay-zero-attempt"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-zero-original", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-zero-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-zero"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 0,
       TerminalAt = NOW(6)
 WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("mark zero-attempt WorkItem dead: %v", err)
	}
	successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-zero-successor", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-zero-successor",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-zero","value":"newer"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue active successor: %v", err)
	}

	_, err = svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReplayDeadEvent error = %v, want ErrReplayBlocked", err)
	}
	var eventReplayedAt, workReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
		t.Fatalf("load original Event replay reservation: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&workReplayedAt); err != nil {
		t.Fatalf("load original WorkItem replay reservation: %v", err)
	}
	var activeEventID string
	if err := db.QueryRowContext(ctx, `SELECT EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey IS NOT NULL`).Scan(&activeEventID); err != nil {
		t.Fatalf("load active projection successor: %v", err)
	}
	if eventReplayedAt.Valid || workReplayedAt.Valid || activeEventID != successorIDs[0] {
		t.Fatalf("blocked zero-attempt replay consumed reservation event=%v work=%v active=%s, want unconsumed original and successor=%s",
			eventReplayedAt.Valid, workReplayedAt.Valid, activeEventID, successorIDs[0])
	}
}

func TestService_ZeroAttemptReplayTerminalizesOriginalMirror(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "replay-zero-attempt-fence"
		packInstall = "install-token-replay-zero-attempt-fence"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(
		ctx,
		namespace,
		"webhook.event",
		"replay-zero-attempt-fence-original",
		json.RawMessage(`{"value":"original"}`),
		packInstall,
		1,
		"",
		"",
		0,
		"",
	)
	if err != nil {
		t.Fatalf("enqueue original Event: %v", err)
	}
	var originalWorkID, originalState string
	var originalAttempt int
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, State, Attempt
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, originalEventID).Scan(&originalWorkID, &originalState, &originalAttempt); err != nil {
		t.Fatalf("load zero-attempt mirror: %v", err)
	}
	if originalState != connector_workqueue.WorkStatePending || originalAttempt != 0 {
		t.Fatalf("original mirror = state %s attempt %d, want pending/0", originalState, originalAttempt)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark Event dead: %v", err)
	}

	replayEventID, err := svc.ReplayDeadEvent(
		ctx,
		allowPerm(string(constants.PermReplayConnectorEvents)),
		originalEventID,
		packInstall,
		1,
	)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}
	var originalReplayedAt, originalTerminalAt sql.NullTime
	if err := db.QueryRowContext(ctx, `
SELECT State, ReplayedAt, TerminalAt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?`, originalWorkID).Scan(&originalState, &originalReplayedAt, &originalTerminalAt); err != nil {
		t.Fatalf("load fenced original mirror: %v", err)
	}
	var replayOfWorkID, replayState string
	if err := db.QueryRowContext(ctx, `
SELECT COALESCE(ReplayOfWorkId, ''), State
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, replayEventID).Scan(&replayOfWorkID, &replayState); err != nil {
		t.Fatalf("load replay child: %v", err)
	}
	if originalState != connector_workqueue.WorkStateDead ||
		!originalReplayedAt.Valid ||
		!originalTerminalAt.Valid ||
		replayOfWorkID != originalWorkID ||
		replayState != connector_workqueue.WorkStatePending {
		t.Fatalf(
			"zero-attempt replay original=(%s,replayed=%v,terminal=%v) child=(%s,%s), want fenced dead original and pending lineage child",
			originalState, originalReplayedAt.Valid, originalTerminalAt.Valid, replayOfWorkID, replayState,
		)
	}
}

func TestService_EventOnlyProjectionReplayBlockedByActiveSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-event-only"
		packInstall = "install-token-projection-replay-event-only"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-event-only-original", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-event-only-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-event-only"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("remove original compatibility WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	successorIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-event-only-successor", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-event-only-successor",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-event-only","value":"newer"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue active successor: %v", err)
	}

	_, err = svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if !errors.Is(err, ErrReplayBlocked) {
		t.Fatalf("ReplayDeadEvent error = %v, want ErrReplayBlocked", err)
	}
	var eventReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
		t.Fatalf("load original Event replay reservation: %v", err)
	}
	var activeEventID string
	if err := db.QueryRowContext(ctx, `SELECT EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey IS NOT NULL`).Scan(&activeEventID); err != nil {
		t.Fatalf("load active projection successor: %v", err)
	}
	var replayChildren int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, originalEventID).Scan(&replayChildren); err != nil {
		t.Fatalf("count replay children: %v", err)
	}
	if eventReplayedAt.Valid || activeEventID != successorIDs[0] || replayChildren != 0 {
		t.Fatalf("blocked Event-only replay reservation=%v active=%s children=%d, want unconsumed original, successor=%s, no child",
			eventReplayedAt.Valid, activeEventID, replayChildren, successorIDs[0])
	}
}

func TestService_EventOnlyProjectionReplayCreatesFreshGeneration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-replay-event-only-success"
		packInstall = "install-token-projection-replay-event-only-success"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "projection-replay-event-only-success-original", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "projection-replay-event-only-success-original",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"contact-replay-event-only-success"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue original projection event: %v", err)
	}
	originalEventID := originalIDs[0]
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID); err != nil {
		t.Fatalf("remove original compatibility WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}

	replayEventID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}
	var eventReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorEvent WHERE Id = ?`, originalEventID).Scan(&eventReplayedAt); err != nil {
		t.Fatalf("load original Event replay reservation: %v", err)
	}
	var workID, projectionKey, projectionActiveKey, replayOfWorkID string
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, COALESCE(ProjectionKey, ''), COALESCE(ProjectionActiveKey, ''), COALESCE(ReplayOfWorkId, '')
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, replayEventID).Scan(&workID, &projectionKey, &projectionActiveKey, &replayOfWorkID); err != nil {
		t.Fatalf("load replay WorkItem: %v", err)
	}
	if !eventReplayedAt.Valid || workID == "" || projectionKey == "" ||
		projectionActiveKey != projectionKey || replayOfWorkID != "" {
		t.Fatalf("Event-only replay reserved=%v work=%s projection=%s active=%s lineage=%s, want reserved fresh active generation without WorkItem lineage",
			eventReplayedAt.Valid, workID, projectionKey, projectionActiveKey, replayOfWorkID)
	}
}

func TestService_ReplayFindsWorkflowActionStoredUnderCompatibilityWorkType(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "replay-compat-worktype"
		packInstall = "install-token-replay-compat-worktype"
		eventKind   = "workflow.action.execute"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	if _, err := db.ExecContext(context.Background(), `
UPDATE _System_ConnectorInboundRoute
   SET EventKind = ?,
       FlowDeveloperName = 'CompatibilityActionHandler'
 WHERE Namespace = ?`, eventKind, namespace); err != nil {
		t.Fatalf("configure workflow action route: %v", err)
	}

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(ctx, namespace, eventKind, "replay-compat-action-original", json.RawMessage(`{"action":"apply"}`), packInstall, 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("enqueue workflow action: %v", err)
	}
	var originalWorkID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&originalWorkID); err != nil {
		t.Fatalf("load original WorkItem: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, originalEventID); err != nil {
		t.Fatalf("mark original Event dead: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET WorkType = ?,
       State = 'dead',
       Attempt = 1,
       TerminalAt = NOW(6)
 WHERE WorkId = ?`, connector_workqueue.WorkTypeConnectorEvent, originalWorkID); err != nil {
		t.Fatalf("seed compatibility WorkType: %v", err)
	}

	replayEventID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}
	var originalReplayedAt sql.NullTime
	var replayOfWorkID, replayWorkType string
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE WorkId = ?`, originalWorkID).Scan(&originalReplayedAt); err != nil {
		t.Fatalf("load original replay reservation: %v", err)
	}
	if err := db.QueryRowContext(ctx, `
SELECT COALESCE(ReplayOfWorkId, ''), WorkType
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, replayEventID).Scan(&replayOfWorkID, &replayWorkType); err != nil {
		t.Fatalf("load replay WorkItem: %v", err)
	}
	if !originalReplayedAt.Valid || replayOfWorkID != originalWorkID || replayWorkType != connector_workqueue.WorkTypeWorkflowActionApply {
		t.Fatalf("compatibility replay originalReserved=%v lineage=%s workType=%s, want true/%s/%s",
			originalReplayedAt.Valid, replayOfWorkID, replayWorkType, originalWorkID, connector_workqueue.WorkTypeWorkflowActionApply)
	}
}

func TestService_LegacyEventReplayMarkerFencesWorkItemOwnedReplay(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "replay-legacy-event-fence"
		packInstall = "install-token-replay-legacy-event-fence"
		eventKind   = "webhook.event"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	ctx := context.Background()
	svc := NewService(db, nil, nil)
	originalEventID, _, err := svc.Enqueue(
		ctx,
		namespace,
		eventKind,
		"replay-legacy-event-fence-original",
		json.RawMessage(`{"objectKind":"contact","sourceID":"legacy-replay-fence"}`),
		packInstall,
		1,
		"",
		"",
		0,
		"",
	)
	if err != nil {
		t.Fatalf("enqueue original Event: %v", err)
	}
	var originalWorkID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, originalEventID).Scan(&originalWorkID); err != nil {
		t.Fatalf("load original WorkItem: %v", err)
	}
	legacyReplayAt := time.Now().UTC().Truncate(time.Second)
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = 'processed',
       ReplayedAt = ?
 WHERE Id = ?`, legacyReplayAt, originalEventID); err != nil {
		t.Fatalf("seed legacy Event replay marker: %v", err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = 'dead',
       Attempt = 1,
       TerminalAt = ?,
       ReplayedAt = NULL
 WHERE WorkId = ?`, legacyReplayAt, originalWorkID); err != nil {
		t.Fatalf("seed WorkItem-owned terminal generation: %v", err)
	}

	_, err = svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), originalEventID, packInstall, 1)
	if !errors.Is(err, ErrEventNotDead) {
		t.Fatalf("ReplayDeadEvent error = %v, want ErrEventNotDead from legacy replay fence", err)
	}
	var workReplayedAt sql.NullTime
	if err := db.QueryRowContext(ctx, `SELECT ReplayedAt FROM _System_ConnectorWorkItem WHERE WorkId = ?`, originalWorkID).Scan(&workReplayedAt); err != nil {
		t.Fatalf("load WorkItem replay reservation: %v", err)
	}
	var replayChildren int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, originalEventID).Scan(&replayChildren); err != nil {
		t.Fatalf("count replay children: %v", err)
	}
	if workReplayedAt.Valid || replayChildren != 0 {
		t.Fatalf("legacy replay fence left WorkItem reservation=%v children=%d, want unconsumed WorkItem and no duplicate child",
			workReplayedAt.Valid, replayChildren)
	}
}

func TestService_CompleteEventForWorkItemDoesNotCommitTerminalParentWithoutMirror(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createConnectorEventTable(t, db)
	ctx := context.Background()
	now := time.Date(2026, 7, 16, 2, 0, 0, 0, time.UTC)
	const (
		eventID    = "per-event-terminal-parent"
		workID     = "cwi:per-event-terminal-parent"
		leaseToken = "per-event-terminal-lease"
	)
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
       (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, ClaimToken, CreatedDate, LastModifiedDate)
VALUES (?, 'terminal-parent', 'terminal-parent-pack', 'object.changed', 'terminal-parent-key', 'processing', 'default', ?, ?, ?)`,
		eventID, leaseToken, now, now); err != nil {
		t.Fatalf("seed processing Event: %v", err)
	}
	// This is the precise former two-step failure shape: the Event is eligible
	// for terminalization but its mirror can no longer satisfy the lease CAS.
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, CreatedDate, LastModifiedDate)
VALUES (?, ?, 'terminal-parent-pack', 'connector_event', 'default', 'blocked', ?, ?, ?)`,
		workID, eventID, now, now, now); err != nil {
		t.Fatalf("seed blocked WorkItem: %v", err)
	}

	completed, err := NewService(db, nil, nil).CompleteEventForWorkItem(ctx, eventID, workID, leaseToken, statusProcessed, "", now)
	if err != nil || completed {
		t.Fatalf("atomic per-event completion = (%v, %v), want false, nil", completed, err)
	}

	var eventStatus string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&eventStatus); err != nil {
		t.Fatalf("read Event after rejected completion: %v", err)
	}
	if eventStatus != statusProcessing {
		t.Fatalf("Event status after rejected mirror completion = %s, want processing rollback", eventStatus)
	}
	var stranded int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e ON e.Id = w.EventId
 WHERE e.Status IN ('processed', 'skipped', 'dead')
   AND w.State IN ('pending', 'processing', 'retry', 'blocked', 'blocked_retry')`).Scan(&stranded); err != nil {
		t.Fatalf("count terminal-parent WorkItems: %v", err)
	}
	if stranded != 0 {
		t.Fatalf("terminal-parent nonterminal WorkItems = %d, want 0", stranded)
	}
}

func TestService_SettledProjectionConcurrentCompleteAndEnqueueNeverLoseSuccessor(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-concurrent"
		packInstall = "install-token-projection-concurrent"
		eventKind   = "contact.propertychange"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	svc := NewService(db, nil, nil)

	for i := 0; i < 12; i++ {
		sourceID := fmt.Sprintf("concurrent-contact-%02d", i)
		firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, fmt.Sprintf("concurrent-first-%02d", i), []BatchEvent{{
			EventKind: eventKind, IdempotencyKey: fmt.Sprintf("concurrent-first-%02d", i),
			Payload: json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"%s","property":"email"}`, sourceID)), RouteVersion: 1,
		}})
		if err != nil {
			t.Fatalf("round %d enqueue first: %v", i, err)
		}
		firstEventID := firstIDs[0]
		var workID string
		if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId = ?`, firstEventID).Scan(&workID); err != nil {
			t.Fatalf("round %d load work item: %v", i, err)
		}
		claimToken := fmt.Sprintf("concurrent-token-%02d", i)
		if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='processing', ClaimToken=? WHERE Id=?`, claimToken, firstEventID); err != nil {
			t.Fatalf("round %d mark event processing: %v", i, err)
		}
		if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State='processing', LeaseToken=?, LeaseOwner='worker', LeaseUntil=DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE WorkId=?`, claimToken, workID); err != nil {
			t.Fatalf("round %d mark work item processing: %v", i, err)
		}

		start := make(chan struct{})
		errCh := make(chan error, 2)
		secondIDCh := make(chan string, 1)
		go func(round int) {
			<-start
			ids, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, fmt.Sprintf("concurrent-second-%02d", round), []BatchEvent{
				{
					EventKind: eventKind, IdempotencyKey: fmt.Sprintf("concurrent-second-%02d", round),
					Payload: json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"%s","property":"phone"}`, sourceID)), RouteVersion: 1,
				},
				{
					EventKind: eventKind, IdempotencyKey: fmt.Sprintf("concurrent-filler-%02d", round),
					Payload: json.RawMessage(fmt.Sprintf(`{"objectKind":"contact","sourceID":"concurrent-filler-%02d","property":"phone"}`, round)), RouteVersion: 1,
				},
			})
			if err == nil {
				secondIDCh <- ids[0]
			}
			errCh <- err
		}(i)
		go func() {
			<-start
			now := time.Now().UTC().Truncate(time.Second)
			completed, err := svc.CompleteSourceProjectionEventForWorkItem(ctx, firstEventID, workID, claimToken, statusProcessed, "", now)
			if err == nil && !completed {
				err = fmt.Errorf("projection completion lost lease")
			}
			errCh <- err
		}()
		close(start)
		for j := 0; j < 2; j++ {
			if err := <-errCh; err != nil {
				t.Fatalf("round %d concurrent operation: %v", i, err)
			}
		}
		secondEventID := <-secondIDCh
		var selectedEventID, state string
		var activeWorkID string
		if err := db.QueryRowContext(ctx, `SELECT WorkId, EventId, State FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey = ?`, sourceProjectionKeyFromWorkID(workID)).Scan(&activeWorkID, &selectedEventID, &state); err != nil {
			t.Fatalf("round %d load final work item: %v", i, err)
		}
		if activeWorkID == workID || selectedEventID != secondEventID ||
			(state != connector_workqueue.WorkStatePending && state != connector_workqueue.WorkStateBlocked) {
			t.Fatalf("round %d lost successor: active=%s selected=%s state=%s want distinct work for %s pending/blocked",
				i, activeWorkID, selectedEventID, state, secondEventID)
		}
	}

	const committedSourceID = "committed-after-completion-begin"
	firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "committed-after-begin-first", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "committed-after-begin-first",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"committed-after-completion-begin","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue committed-after-begin first Event: %v", err)
	}
	firstEventID := firstIDs[0]
	var workID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE EventId=?`, firstEventID).Scan(&workID); err != nil {
		t.Fatalf("load committed-after-begin WorkItem: %v", err)
	}
	claimToken := "committed-after-begin-token"
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='processing', ClaimToken=? WHERE Id=?`, claimToken, firstEventID); err != nil {
		t.Fatalf("mark committed-after-begin Event processing: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State='processing', LeaseToken=?, LeaseOwner='worker', LeaseUntil=DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE WorkId=?`, claimToken, workID); err != nil {
		t.Fatalf("mark committed-after-begin WorkItem processing: %v", err)
	}
	completionTx, completionCleanup, err := sqltx.BeginPessimisticReadCommittedTx(ctx, db)
	if err != nil {
		t.Fatalf("begin committed-successor completion transaction: %v", err)
	}
	defer completionCleanup()
	defer completionTx.Rollback()
	var snapshotCount int
	if err := completionTx.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorEvent`).Scan(&snapshotCount); err != nil {
		t.Fatalf("establish pre-successor completion snapshot: %v", err)
	}
	secondIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "committed-after-begin-second", []BatchEvent{
		{
			EventKind: eventKind, IdempotencyKey: "committed-after-begin-second",
			Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"committed-after-completion-begin","property":"phone"}`), RouteVersion: 1,
		},
		{
			EventKind: eventKind, IdempotencyKey: "committed-after-begin-filler",
			Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"committed-after-begin-filler","property":"phone"}`), RouteVersion: 1,
		},
	})
	if err != nil {
		t.Fatalf("enqueue successor after completion transaction began: %v", err)
	}
	completed, err := connector_workqueue.NewRepository(db, nil, nil).WithTx(completionTx).CompleteWorkItem(
		ctx, workID, claimToken, connector_workqueue.WorkStateProcessed, time.Now().UTC().Truncate(time.Second),
	)
	if err != nil || !completed {
		t.Fatalf("complete after successor commit = (%v, %v), want true, nil", completed, err)
	}
	if err := completionTx.Commit(); err != nil {
		t.Fatalf("commit completion with visible successor: %v", err)
	}
	var selectedEventID string
	var activeWorkID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId, EventId FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey=?`, sourceProjectionKeyFromWorkID(workID)).Scan(&activeWorkID, &selectedEventID); err != nil {
		t.Fatalf("load committed successor selection: %v", err)
	}
	if activeWorkID == workID || selectedEventID != secondIDs[0] {
		t.Fatalf("selected generation after completion snapshot = %s/%s, want distinct work for successor %s for %s",
			activeWorkID, selectedEventID, secondIDs[0], committedSourceID)
	}
	var stranded int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e ON e.Id = w.EventId
 WHERE w.WorkId LIKE 'csp:%'
   AND e.Status IN ('processed', 'skipped', 'dead')
   AND w.State IN ('pending', 'processing', 'retry', 'blocked', 'blocked_retry')`).Scan(&stranded); err != nil {
		t.Fatalf("count source-projection terminal-parent WorkItems: %v", err)
	}
	if stranded != 0 {
		t.Fatalf("source-projection terminal-parent nonterminal WorkItems = %d, want 0", stranded)
	}
	t.Log("evidence 12/12 concurrent fresh-batch races plus a successor committed after completion began were preserved as terminal history plus one active generation")
}

func TestRepository_SourceProjectionCompletionDoesNotInvertEnqueueLocks(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace   = "projection-lock-order"
		packInstall = "install-token-projection-lock-order"
		eventKind   = "contact.propertychange"
		sourceID    = "lock-order-contact"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	makeSettledProjectionRoute(t, db, namespace, eventKind)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	svc := NewService(db, nil, nil)
	firstIDs, _, err := svc.EnqueueBatch(ctx, namespace, packInstall, "lock-order-first", []BatchEvent{{
		EventKind: eventKind, IdempotencyKey: "lock-order-first",
		Payload: json.RawMessage(`{"objectKind":"contact","sourceID":"lock-order-contact","property":"email"}`), RouteVersion: 1,
	}})
	if err != nil {
		t.Fatalf("enqueue first projection: %v", err)
	}
	firstEventID := firstIDs[0]
	claimToken := "lock-order-token"
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='processing', ClaimToken=? WHERE Id=?`, claimToken, firstEventID); err != nil {
		t.Fatalf("mark first Event processing: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET State='processing', LeaseToken=?, LeaseOwner='worker', LeaseUntil=DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE EventId=?`, claimToken, firstEventID); err != nil {
		t.Fatalf("mark projection WorkItem processing: %v", err)
	}

	var workID, workType, queueClass, sourceLaneKey, targetLaneKey string
	var laneOrdinal, generation int64
	if err := db.QueryRowContext(ctx, `
SELECT WorkId, WorkType, QueueClass, COALESCE(SourceLaneKey, ''),
       COALESCE(TargetMutationLaneKey, ''), LaneOrdinal, Generation
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?`, firstEventID).Scan(
		&workID, &workType, &queueClass, &sourceLaneKey, &targetLaneKey, &laneOrdinal, &generation,
	); err != nil {
		t.Fatalf("load projection WorkItem: %v", err)
	}

	completionTx, completionCleanup, err := sqltx.BeginPessimisticReadCommittedTx(ctx, db)
	if err != nil {
		t.Fatalf("begin completion transaction: %v", err)
	}
	defer completionCleanup()
	defer completionTx.Rollback()
	var lockedWorkID string
	if err := completionTx.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem WHERE WorkId=? FOR UPDATE`, workID).Scan(&lockedWorkID); err != nil {
		t.Fatalf("lock projection WorkItem: %v", err)
	}

	enqueueTx, enqueueCleanup, err := sqltx.BeginPessimisticTx(ctx, db)
	if err != nil {
		t.Fatalf("begin enqueue transaction: %v", err)
	}
	defer enqueueCleanup()
	defer enqueueTx.Rollback()
	secondEventID := "lock-order-successor"
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := enqueueTx.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, Payload, RetryCount,
   NextAttemptAt, SourceObjectKind, SourceObjectID, SourceLaneKey, RouteVersion,
   EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, 0, ?, 'contact', ?, ?, 1, ?, ?, ?)`,
		secondEventID, namespace, packInstall, eventKind, secondEventID, queueClass,
		json.RawMessage(`{"objectKind":"contact","sourceID":"lock-order-contact","property":"phone"}`),
		now, sourceID, sourceLaneKey, laneOrdinal+1, now, now,
	); err != nil {
		t.Fatalf("insert uncommitted successor Event: %v", err)
	}

	ensureStarted := make(chan struct{})
	ensureDone := make(chan error, 1)
	go func() {
		close(ensureStarted)
		_, _, ensureErr := connector_workqueue.NewRepository(db, nil, nil).WithTx(enqueueTx).AdmitSourceProjectionWorkItem(ctx, connector_workqueue.SourceProjectionWorkItemInput{InputStartOrdinal: laneOrdinal + 1, WorkItem: connector_workqueue.CreateWorkItemInput{
			WorkID: sourceProjectionKeyFromWorkID(workID), EventID: secondEventID, InstalledPackID: packInstall,
			WorkType: workType, QueueClass: queueClass, State: connector_workqueue.WorkStatePending,
			RunAfter: now, LaneOrdinal: laneOrdinal + 1, SourceLaneKey: sourceLaneKey,
			TargetMutationLaneKey: targetLaneKey, Generation: generation, Now: now,
		}}, true)
		ensureDone <- ensureErr
	}()
	<-ensureStarted
	select {
	case err := <-ensureDone:
		t.Fatalf("enqueue WorkItem reconcile did not wait for completion lock: %v", err)
	case <-time.After(150 * time.Millisecond):
	}

	completed, err := connector_workqueue.NewRepository(db, nil, nil).WithTx(completionTx).CompleteWorkItem(
		ctx, workID, claimToken, connector_workqueue.WorkStateProcessed, now,
	)
	if err != nil || !completed {
		t.Fatalf("complete while enqueue holds successor Event = (%v, %v), want true, nil", completed, err)
	}
	if err := completionTx.Commit(); err != nil {
		t.Fatalf("commit completion transaction: %v", err)
	}
	if err := <-ensureDone; err != nil {
		t.Fatalf("reconcile successor after completion: %v", err)
	}
	if err := enqueueTx.Commit(); err != nil {
		t.Fatalf("commit successor enqueue: %v", err)
	}

	var selectedEventID, state string
	var activeWorkID string
	if err := db.QueryRowContext(ctx, `SELECT WorkId, EventId, State FROM _System_ConnectorWorkItem WHERE ProjectionActiveKey=?`, sourceProjectionKeyFromWorkID(workID)).Scan(&activeWorkID, &selectedEventID, &state); err != nil {
		t.Fatalf("load final projection WorkItem: %v", err)
	}
	if activeWorkID == workID || selectedEventID != secondEventID ||
		(state != connector_workqueue.WorkStatePending && state != connector_workqueue.WorkStateBlocked) {
		t.Fatalf("final projection WorkItem = work %s event %s state %s, want distinct work for successor %s pending/blocked",
			activeWorkID, selectedEventID, state, secondEventID)
	}
	t.Log("evidence completion did not lock the uncommitted successor Event; enqueue created the next durable generation after the completion lock released")
}

// createInstalledPackTable creates a minimal _System_InstalledPack table and
// seeds it with the given namespaces so that ResolveConnectorPackByNamespace
// returns active rows for the test fixtures.
func createInstalledPackTable(t *testing.T, db *sql.DB, namespaces ...string) {
	t.Helper()
	_, err := db.Exec(`CREATE TABLE _System_InstalledPack (
		Id            VARCHAR(255) PRIMARY KEY,
		PackInstallId VARCHAR(255) NOT NULL DEFAULT '',
		Name          VARCHAR(255) NOT NULL,
		Category      VARCHAR(128) NOT NULL,
		Status        VARCHAR(64)  NOT NULL,
		PackVersion   VARCHAR(255),
		IsDeleted     TINYINT      NOT NULL DEFAULT 0
	)`)
	if err != nil {
		t.Fatalf("create _System_InstalledPack: %v", err)
	}
	for _, ns := range namespaces {
		installToken := fmt.Sprintf("install-token-%s", ns)
		if _, err := db.Exec(`INSERT INTO _System_InstalledPack (Id, PackInstallId, Name, Category, Status, IsDeleted)
			VALUES (?, ?, ?, 'connector-pack', 'active', 0)`,
			fmt.Sprintf("pack-%s", ns), installToken, ns,
		); err != nil {
			t.Fatalf("seed _System_InstalledPack %s: %v", ns, err)
		}
	}
	connectorfixture.CreateSystemTables(t, db, "_System_ConnectorProjectionProtocol")
	if _, err := db.Exec(`INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal)
 SELECT PackInstallId,1,0 FROM _System_InstalledPack`); err != nil {
		t.Fatal(err)
	}

}

// createConnectorInboundRouteTable creates _System_ConnectorInboundRoute and
// seeds one live route row per namespace so Enqueue's B20 JOIN (routeVersion>=1)
// can find a matching live route row in integration tests.
//
// Each namespace gets a single route (Version=1, IsCurrent=1, IsDeleted=0)
// for the eventKind "webhook.event". The InstallId matches createInstalledPackTable's
// "install-token-<ns>" convention.
func createConnectorInboundRouteTable(t *testing.T, db *sql.DB, namespaces ...string) {
	t.Helper()
	_, err := db.Exec(`CREATE TABLE _System_ConnectorInboundRoute (
		Id                 VARCHAR(255) PRIMARY KEY,
		Namespace          VARCHAR(255) NOT NULL,
		InstalledPackId    VARCHAR(255) NOT NULL,
			EventKind             VARCHAR(255) NOT NULL,
			Version               INT          NOT NULL DEFAULT 1,
			FlowDeveloperName     VARCHAR(255) NOT NULL,
			InvocationMode        VARCHAR(64)  NOT NULL DEFAULT 'per_event',
			BatchInputName        VARCHAR(255) NOT NULL DEFAULT 'events',
			BatchResultOutputName VARCHAR(255) NOT NULL DEFAULT 'eventResults',
			MaxBatchSize          INT          NOT NULL DEFAULT 100,
			QueueClass            VARCHAR(32)  NOT NULL DEFAULT 'default',
			FlowExecutionFilter   JSON,
			IsCurrent             TINYINT(1)   NOT NULL DEFAULT 1,
			IsDeleted             TINYINT(1)   NOT NULL DEFAULT 0,
			CreatedDate           DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
			LastModifiedDate      DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
		UNIQUE KEY uniq_cir_pack_kind_version (InstalledPackId, EventKind, Version)
	)`)
	if err != nil {
		t.Fatalf("create _System_ConnectorInboundRoute: %v", err)
	}
	for _, ns := range namespaces {
		installToken := fmt.Sprintf("install-token-%s", ns)
		routeID := fmt.Sprintf("cir-test-%s", ns)
		if _, err := db.Exec(`INSERT INTO _System_ConnectorInboundRoute
			(Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName, IsCurrent, IsDeleted)
			VALUES (?, ?, ?, 'webhook.event', 1, 'TestHandler', 1, 0)`,
			routeID, ns, installToken,
		); err != nil {
			t.Fatalf("seed _System_ConnectorInboundRoute %s: %v", ns, err)
		}
	}
}

func createMigrationTable(t *testing.T, db *sql.DB) {
	t.Helper()
	_, err := db.Exec(`CREATE TABLE _System_Migration (
		Id               VARCHAR(255) PRIMARY KEY,
		Version          INT NOT NULL,
		Description      TEXT,
		AppliedAt        DATETIME NOT NULL,
		CreatedDate      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
		LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
	)`)
	if err != nil {
		t.Fatalf("create _System_Migration: %v", err)
	}
}

const migrationIDConnectorEventEnqueueOrdinalBackfill = "20260619_connector_event_enqueue_ordinal_backfill"

func migrateConnectorEventEnqueueOrdinalBackfill(ctx context.Context, db *sql.DB) (int, error) {
	if db == nil {
		return 0, fmt.Errorf("connector event enqueue ordinal backfill requires a database")
	}
	applied, err := connectorEventMigrationApplied(ctx, db, migrationIDConnectorEventEnqueueOrdinalBackfill)
	if err != nil {
		return 0, err
	}
	if applied {
		remaining, err := connectorEventEnqueueOrdinalLiveZeroCount(ctx, db)
		if err != nil {
			return 0, err
		}
		if remaining == 0 {
			return 0, nil
		}
	}
	updated, err := backfillConnectorEventEnqueueOrdinals(ctx, db)
	if err != nil {
		return updated, err
	}
	if err := recordConnectorEventMigrationApplied(ctx, db, migrationIDConnectorEventEnqueueOrdinalBackfill); err != nil {
		return updated, err
	}
	return updated, nil
}

func connectorEventMigrationApplied(ctx context.Context, db *sql.DB, id string) (bool, error) {
	var one int
	err := db.QueryRowContext(ctx, `SELECT 1 FROM _System_Migration WHERE Id = ? LIMIT 1`, id).Scan(&one)
	if err == sql.ErrNoRows {
		return false, nil
	}
	return err == nil, err
}

func recordConnectorEventMigrationApplied(ctx context.Context, db *sql.DB, id string) error {
	_, err := db.ExecContext(ctx, `
INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
VALUES (?, 1, 'Backfill _System_ConnectorEvent.EnqueueOrdinal for warm-upgrade rows.', NOW(), NOW(), NOW())
ON DUPLICATE KEY UPDATE AppliedAt = VALUES(AppliedAt), LastModifiedDate = VALUES(LastModifiedDate)`, id)
	return err
}

func connectorEventEnqueueOrdinalLiveZeroCount(ctx context.Context, db *sql.DB) (int, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE EnqueueOrdinal = 0
   AND Status NOT IN ('processed', 'dead')`).Scan(&count); err != nil {
		return 0, err
	}
	return count, nil
}

func backfillConnectorEventEnqueueOrdinals(ctx context.Context, db *sql.DB) (int, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	rows, err := tx.QueryContext(ctx, `
SELECT DISTINCT InstalledPackId
  FROM _System_ConnectorEvent
 WHERE EnqueueOrdinal = 0
   AND Status NOT IN ('processed', 'dead')
 ORDER BY InstalledPackId ASC`)
	if err != nil {
		return 0, err
	}
	packOrder := make([]string, 0)
	for rows.Next() {
		var packInstallID string
		if err := rows.Scan(&packInstallID); err != nil {
			_ = rows.Close()
			return 0, err
		}
		packOrder = append(packOrder, packInstallID)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return 0, err
	}
	if err := rows.Close(); err != nil {
		return 0, err
	}

	updated := 0
	for _, packInstallID := range packOrder {
		eventRows, err := tx.QueryContext(ctx, `
SELECT Id, EnqueueOrdinal
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND Status NOT IN ('processed', 'dead')
 ORDER BY CreatedDate ASC, DeliverySequence ASC, Id ASC`, packInstallID)
		if err != nil {
			return updated, err
		}
		type eventOrdinal struct {
			id      string
			ordinal int64
		}
		events := make([]eventOrdinal, 0)
		for eventRows.Next() {
			var ev eventOrdinal
			if err := eventRows.Scan(&ev.id, &ev.ordinal); err != nil {
				_ = eventRows.Close()
				return updated, err
			}
			events = append(events, ev)
		}
		if err := eventRows.Err(); err != nil {
			_ = eventRows.Close()
			return updated, err
		}
		if err := eventRows.Close(); err != nil {
			return updated, err
		}

		for i, ev := range events {
			ordinal := int64(i + 1)
			if ev.ordinal == ordinal {
				continue
			}
			res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET EnqueueOrdinal = ?, LastModifiedDate = NOW()
 WHERE Id = ?`, ordinal, ev.id)
			if err != nil {
				return updated, err
			}
			if n, err := res.RowsAffected(); err != nil {
				return updated, err
			} else if n > 0 {
				updated += int(n)
			}
		}
	}
	var remaining int
	if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE EnqueueOrdinal = 0
   AND Status NOT IN ('processed', 'dead')`).Scan(&remaining); err != nil {
		return updated, err
	}
	if remaining != 0 {
		return updated, fmt.Errorf("enqueue ordinal backfill: %d legacy row(s) remain with EnqueueOrdinal=0", remaining)
	}
	if err := tx.Commit(); err != nil {
		return updated, err
	}
	committed = true
	return updated, nil
}

// ─────────────────────────────────────────────────────────────────────────────
// Test 15 — 50 events / 50 goroutines — exactly one winner per event.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ConcurrentClaim_SingleWinnerPerEvent(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "noop-connector")
	createConnectorEventTable(t, db)
	// B20: Enqueue with routeVersion=1 JOINs _System_ConnectorInboundRoute.
	createConnectorInboundRouteTable(t, db, "noop-connector")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const eventCount = 50

	// Enqueue 50 distinct events.
	eventIDs := make([]string, eventCount)
	for i := 0; i < eventCount; i++ {
		ikey := fmt.Sprintf("ikey-%d", i)
		deliveryID := fmt.Sprintf("dlv-single-%d", i) // each event is its own delivery
		id, isDup, err := svc.Enqueue(ctx, "noop-connector", "webhook.event", ikey, nil, "install-token-noop-connector", 1, "", "", 0, deliveryID)
		if err != nil {
			t.Fatalf("Enqueue[%d]: %v", i, err)
		}
		if isDup {
			t.Fatalf("Enqueue[%d]: unexpected duplicate", i)
		}
		eventIDs[i] = id
	}

	// 50 goroutines race to claim — all start at the same moment.
	type result struct {
		workerID string
		eventID  string
	}
	results := make(chan result, eventCount*2)
	start := make(chan struct{})
	var wg sync.WaitGroup

	for w := 0; w < eventCount; w++ {
		wg.Add(1)
		workerID := fmt.Sprintf("worker-%d", w)
		go func(wid string) {
			defer wg.Done()
			<-start
			// Each goroutine polls until it claims exactly one event or the
			// context expires.  This mirrors a real worker's retry loop and
			// ensures all 50 events are eventually claimed (single winner each).
			for {
				// ClaimNext takes installedPackID; the seed loop sets PackInstallId = "install-token-<ns>".
				ev, claimed, err := svc.ClaimNext(ctx, "install-token-noop-connector", wid)
				if err != nil || ctx.Err() != nil {
					return
				}
				if claimed && ev != nil {
					results <- result{workerID: wid, eventID: ev.ID}
					return
				}
				// Brief yield before retrying so we don't busy-spin.
				time.Sleep(time.Millisecond)
			}
		}(workerID)
	}

	close(start) // release all goroutines simultaneously
	wg.Wait()
	close(results)

	// Collect claims.
	claimedBy := map[string]string{}
	for r := range results {
		if prev, dup := claimedBy[r.eventID]; dup {
			t.Errorf("event %s claimed by both %s and %s", r.eventID, prev, r.workerID)
		}
		claimedBy[r.eventID] = r.workerID
	}

	if len(claimedBy) != eventCount {
		t.Errorf("expected %d claims, got %d", eventCount, len(claimedBy))
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test #2161 repro — retired-incarnation duplicate must return
// ErrPackIncarnationChanged, not a false isDuplicate=true.
//
// Setup:
//   - install-B is the ACTIVE incarnation in _System_InstalledPack.
//   - install-A is the RETIRED incarnation (no row or Status≠active).
//   - A pre-existing _System_ConnectorEvent row with InstalledPackId=install-A
//     and IdempotencyKey="dup-key" exists (e.g. left over from the previous
//     incarnation's activity).
//
// Before the fix: the pre-tx dedup SELECT ran first, found the install-A row,
// and returned (existingID, true, nil) — silently dropping the delivery.
// After the fix: the active-incarnation FOR UPDATE lock fails for install-A
// (no active row) and returns ErrPackIncarnationChanged — the caller retries
// with the current active install-B.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_StaleIncarnationDuplicate_ReturnsErrPackIncarnationChanged(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	// Seed _System_InstalledPack: ONLY install-B is active; install-A has no row.
	_, err := db.Exec(`CREATE TABLE _System_InstalledPack (
		Id            VARCHAR(255) PRIMARY KEY,
		PackInstallId VARCHAR(255) NOT NULL DEFAULT '',
		Name          VARCHAR(255) NOT NULL,
		Category      VARCHAR(128) NOT NULL,
		Status        VARCHAR(64)  NOT NULL,
		PackVersion   VARCHAR(255),
		IsDeleted     TINYINT      NOT NULL DEFAULT 0
	)`)
	if err != nil {
		t.Fatalf("create _System_InstalledPack: %v", err)
	}
	// insert-B = active incarnation
	if _, err := db.Exec(`INSERT INTO _System_InstalledPack (Id, PackInstallId, Name, Category, Status, IsDeleted)
		VALUES ('pack-b', 'install-B', 'noop-connector', 'connector-pack', 'active', 0)`); err != nil {
		t.Fatalf("seed install-B: %v", err)
	}
	// install-A intentionally NOT inserted — retired incarnation has no active row.

	createConnectorEventTable(t, db)

	// Seed a pre-existing ConnectorEvent row for the RETIRED incarnation (install-A).
	// This simulates a leftover event from before the pack was reinstalled.
	if _, err := db.Exec(`INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RetryCount, RouteVersion, DeliverySequence, CreatedDate, LastModifiedDate)
		VALUES ('stale-evt-001', 'noop-connector', 'install-A', 'webhook.event', 'dup-key', 'dead', 5, 0, 0, NOW(), NOW())`); err != nil {
		t.Fatalf("seed stale ConnectorEvent: %v", err)
	}

	svc := NewService(db, nil, nil)
	ctx := context.Background()

	// Call Enqueue with the RETIRED incarnation's packInstallID and the same
	// IdempotencyKey that already exists under install-A.
	//
	// Expected (post-fix): ErrPackIncarnationChanged — the FOR UPDATE lock finds
	// no active row for install-A and the caller is prompted to retry with install-B.
	// Forbidden (pre-fix): (existingID, isDuplicate=true, nil) — false duplicate
	// caused by the pre-tx dedup SELECT short-circuiting before the lock.
	_, isDup, err := svc.Enqueue(ctx, "noop-connector", "webhook.event", "dup-key", nil, "install-A", 0, "", "", 0, "")
	if err == nil {
		t.Fatalf("expected ErrPackIncarnationChanged, got nil (isDup=%v) — retired incarnation must not return false duplicate", isDup)
	}
	if isDup {
		t.Fatalf("got isDuplicate=true for a RETIRED incarnation — this is the bug described in issue #2161")
	}
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged, got %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test 16 — noop-connector and webhook-test-connector use the same code path.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_NoopConnectorReusesSamePath(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "noop-connector", "webhook-test-connector")
	createConnectorEventTable(t, db)
	// B20: Enqueue with routeVersion=1 JOINs _System_ConnectorInboundRoute.
	createConnectorInboundRouteTable(t, db, "noop-connector", "webhook-test-connector")

	svc := NewService(db, nil, nil)
	ctx := context.Background()

	// Enqueue one event per namespace.
	id1, _, err := svc.Enqueue(ctx, "noop-connector", "webhook.event", "ikey-noop", nil, "install-token-noop-connector", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue noop-connector: %v", err)
	}
	id2, _, err := svc.Enqueue(ctx, "webhook-test-connector", "webhook.event", "ikey-wt", nil, "install-token-webhook-test-connector", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue webhook-test-connector: %v", err)
	}

	// Claim each event through the identical ClaimNext path.
	// The seed loop assigns PackInstallId = "install-token-<ns>" for each namespace.
	ev1, ok1, err := svc.ClaimNext(ctx, "install-token-noop-connector", "w-noop")
	if err != nil || !ok1 || ev1 == nil {
		t.Fatalf("ClaimNext noop-connector: ev=%v ok=%v err=%v", ev1, ok1, err)
	}
	if ev1.ID != id1 {
		t.Errorf("ClaimNext noop-connector: want %s, got %s", id1, ev1.ID)
	}

	ev2, ok2, err := svc.ClaimNext(ctx, "install-token-webhook-test-connector", "w-wt")
	if err != nil || !ok2 || ev2 == nil {
		t.Fatalf("ClaimNext webhook-test-connector: ev=%v ok=%v err=%v", ev2, ok2, err)
	}
	if ev2.ID != id2 {
		t.Errorf("ClaimNext webhook-test-connector: want %s, got %s", id2, ev2.ID)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: strict per-delivery ordering — 8 events, one DeliveryId, 8 workers.
// Workers must observe the events in DeliverySequence order and at most one
// is in-flight at a time within the delivery.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ConcurrentClaim_StrictPerDeliveryOrder(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "order-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "order-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const count = 8
	const packInstall = "install-token-order-test"
	const dlvID = "dlv-order-test-001"

	// Enqueue 8 events under the same DeliveryId with sequential DeliverySequence.
	for i := 0; i < count; i++ {
		ikey := fmt.Sprintf("order-ikey-%d", i)
		if _, _, err := svc.Enqueue(ctx, "order-test", "webhook.event", ikey, nil, packInstall, 1, "", "", i, dlvID); err != nil {
			t.Fatalf("Enqueue[%d]: %v", i, err)
		}
	}

	// 8 workers race. Each HOLDS its claim briefly (widening any out-of-order
	// window) before MarkProcessed, then loops until the whole delivery drains.
	// Strict per-delivery ordering requires BOTH: claims happen in exact
	// DeliverySequence order AND at most one event is ever in flight at a time.
	// Without the ClaimNext NOT EXISTS guard, a later sibling is claimed while an
	// earlier one is still processing → maxInFlight>1 and claimOrder out of order →
	// this test fails (verified by removing the guard).
	var (
		mu          sync.Mutex
		claimOrder  []int
		inFlight    int
		maxInFlight int
		processed   int32
		wg          sync.WaitGroup
	)
	start := make(chan struct{})
	for w := 0; w < count; w++ {
		wg.Add(1)
		workerID := fmt.Sprintf("order-worker-%d", w)
		go func(wid string) {
			defer wg.Done()
			<-start
			for {
				if ctx.Err() != nil {
					return
				}
				ev, claimed, err := svc.ClaimNext(ctx, packInstall, wid)
				if err != nil {
					return
				}
				if !claimed || ev == nil {
					if atomic.LoadInt32(&processed) >= count {
						return
					}
					time.Sleep(time.Millisecond)
					continue
				}
				mu.Lock()
				claimOrder = append(claimOrder, ev.DeliverySequence)
				inFlight++
				if inFlight > maxInFlight {
					maxInFlight = inFlight
				}
				mu.Unlock()

				time.Sleep(5 * time.Millisecond) // hold the lease to widen the race window

				mu.Lock()
				inFlight--
				mu.Unlock()
				if err := svc.MarkProcessed(ctx, ev.ID, ev.ClaimToken); err != nil {
					t.Errorf("MarkProcessed(%s): %v", ev.ID, err)
					return
				}
				atomic.AddInt32(&processed, 1)
			}
		}(workerID)
	}
	close(start)
	wg.Wait()

	if got := atomic.LoadInt32(&processed); got != count {
		t.Fatalf("processed=%d, want %d (claimOrder=%v)", got, count, claimOrder)
	}
	if maxInFlight != 1 {
		t.Errorf("maxInFlight=%d, want 1 — a later same-delivery event started before an earlier one finished (per-delivery serialization violated)", maxInFlight)
	}
	if len(claimOrder) != count {
		t.Fatalf("claimOrder len=%d, want %d: %v", len(claimOrder), count, claimOrder)
	}
	for i := 0; i < count; i++ {
		if claimOrder[i] != i {
			t.Fatalf("claimOrder=%v, want [0..%d] — DeliverySequence must be claimed/executed strictly in order", claimOrder, count-1)
		}
	}
}

// Test: same-source rows across different DeliveryIds must still drain
// single-file. This covers workloads that emit one delivery per mutation while
// still targeting the same underlying source record.
func TestService_ClaimNext_SerializesSameSourceAcrossDeliveries(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-order-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "source-order-test")

	baseNow := time.Date(2026, 6, 12, 16, 0, 0, 123000000, time.UTC)
	nowSeq := []time.Time{
		baseNow,
		baseNow.Add(200 * time.Second),
		baseNow.Add(400 * time.Second),
	}
	nowIdx := 0
	svc := NewService(db, deterministicID(
		"evt-source-z-head",
		"evt-source-a-tail",
		"evt-source-m-other",
	), func() time.Time {
		if nowIdx >= len(nowSeq) {
			return nowSeq[len(nowSeq)-1]
		}
		v := nowSeq[nowIdx]
		nowIdx++
		return v
	})
	ctx := context.Background()

	const packInstall = "install-token-source-order-test"
	payloadA1 := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Sent"}}`)
	payloadA2 := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Responded"}}`)
	payloadB1 := []byte(`{"objectKind":"workflow-actions","sourceID":"other-source","payload":{"status":"Sent"}}`)

	if _, _, err := svc.Enqueue(ctx, "source-order-test", "webhook.event", "ikey-source-a-1", payloadA1, packInstall, 1, "", "", 0, "dlv-source-a-1"); err != nil {
		t.Fatalf("Enqueue shared-source first: %v", err)
	}
	if _, _, err := svc.Enqueue(ctx, "source-order-test", "webhook.event", "ikey-source-a-2", payloadA2, packInstall, 1, "", "", 0, "dlv-source-a-2"); err != nil {
		t.Fatalf("Enqueue shared-source second: %v", err)
	}
	if _, _, err := svc.Enqueue(ctx, "source-order-test", "webhook.event", "ikey-source-b-1", payloadB1, packInstall, 1, "", "", 0, "dlv-source-b-1"); err != nil {
		t.Fatalf("Enqueue other-source: %v", err)
	}

	ev1, ok1, err := svc.ClaimNext(ctx, packInstall, "worker-source-1")
	if err != nil || !ok1 || ev1 == nil {
		t.Fatalf("ClaimNext shared-source head: ev=%v ok=%v err=%v", ev1, ok1, err)
	}
	if ev1.ID != "evt-source-z-head" {
		t.Fatalf("first claim = %s, want evt-source-z-head", ev1.ID)
	}

	ev2, ok2, err := svc.ClaimNext(ctx, packInstall, "worker-source-2")
	if err != nil || !ok2 || ev2 == nil {
		t.Fatalf("ClaimNext parallel different source: ev=%v ok=%v err=%v", ev2, ok2, err)
	}
	if ev2.ID != "evt-source-m-other" {
		t.Fatalf("second claim = %s, want evt-source-m-other; same-source sibling was not blocked", ev2.ID)
	}

	evBlocked, okBlocked, err := svc.ClaimNext(ctx, packInstall, "worker-source-3")
	if err != nil {
		t.Fatalf("ClaimNext blocked sibling check: %v", err)
	}
	if okBlocked || evBlocked != nil {
		t.Fatalf("same-source later sibling must stay blocked while first is processing; got ev=%v ok=%v", evBlocked, okBlocked)
	}

	if err := svc.MarkProcessed(ctx, ev1.ID, ev1.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed first shared-source row: %v", err)
	}

	ev3, ok3, err := svc.ClaimNext(ctx, packInstall, "worker-source-4")
	if err != nil || !ok3 || ev3 == nil {
		t.Fatalf("ClaimNext released sibling: ev=%v ok=%v err=%v", ev3, ok3, err)
	}
	if ev3.ID != "evt-source-a-tail" {
		t.Fatalf("released claim = %s, want evt-source-a-tail", ev3.ID)
	}
}

// Test: same-source rows with equal EnqueueOrdinal must still drain single-file.
// Id acts as the stable tiebreak when the source lane ordinal collides.
func TestService_ClaimNext_SerializesSameSourceAcrossEqualEnqueueOrdinals(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-equal-ordinal-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "source-equal-ordinal-test")

	svc := NewService(db, deterministicID(
		"evt-source-equal-a",
		"evt-source-equal-b",
	), nil)
	ctx := context.Background()

	const packInstall = "install-token-source-equal-ordinal-test"
	payloadA := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Sent"}}`)
	payloadB := []byte(`{"objectKind":"workflow-actions","sourceID":"shared-source","payload":{"status":"Responded"}}`)

	idA, _, err := svc.Enqueue(ctx, "source-equal-ordinal-test", "webhook.event", "ikey-source-equal-a", payloadA, packInstall, 1, "", "", 0, "dlv-source-equal-a")
	if err != nil {
		t.Fatalf("Enqueue shared-source first: %v", err)
	}
	idB, _, err := svc.Enqueue(ctx, "source-equal-ordinal-test", "webhook.event", "ikey-source-equal-b", payloadB, packInstall, 1, "", "", 0, "dlv-source-equal-b")
	if err != nil {
		t.Fatalf("Enqueue shared-source second: %v", err)
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent
SET EnqueueOrdinal = 7
WHERE Id IN (?, ?)`, idA, idB); err != nil {
		t.Fatalf("force equal EnqueueOrdinal: %v", err)
	}

	evA, okA, err := svc.ClaimNext(ctx, packInstall, "worker-source-equal-a")
	if err != nil || !okA || evA == nil {
		t.Fatalf("ClaimNext equal-ordinal shared-source head: ev=%v ok=%v err=%v", evA, okA, err)
	}
	if evA.ID != idA {
		t.Fatalf("first claim = %s, want %s", evA.ID, idA)
	}

	evBlocked, okBlocked, err := svc.ClaimNext(ctx, packInstall, "worker-source-equal-b")
	if err != nil {
		t.Fatalf("ClaimNext blocked sibling check: %v", err)
	}
	if okBlocked || evBlocked != nil {
		t.Fatalf("same-source later sibling must stay blocked while first is processing; got ev=%v ok=%v", evBlocked, okBlocked)
	}

	if err := svc.MarkProcessed(ctx, evA.ID, evA.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed first equal-ordinal shared-source row: %v", err)
	}

	evB, okB, err := svc.ClaimNext(ctx, packInstall, "worker-source-equal-b")
	if err != nil || !okB || evB == nil {
		t.Fatalf("ClaimNext released equal-ordinal sibling: ev=%v ok=%v err=%v", evB, okB, err)
	}
	if evB.ID != idB {
		t.Fatalf("released claim = %s, want %s", evB.ID, idB)
	}
}

func TestConnectorEventEnqueueOrdinalBackfillPreservesLegacySameSourceOrder(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-legacy-ordinal-test")
	createConnectorEventTable(t, db)
	createMigrationTable(t, db)

	ctx := context.Background()
	const packInstall = "install-token-source-legacy-ordinal-test"
	payloadHead := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-shared-source","payload":{"status":"Created"}}`)
	payloadTail := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-shared-source","payload":{"status":"Updated"}}`)
	headCreated := time.Date(2026, 6, 19, 9, 0, 0, 0, time.UTC)
	tailCreated := headCreated.Add(time.Minute)

	// Simulate warm-upgrade rows: EnqueueOrdinal inherited the column default 0.
	// The older row deliberately has a lexicographically larger Id than the later
	// row, proving the repair preserves historical queue order instead of letting
	// the equal-ordinal Id tiebreak invert the same-source lane.
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
		 RouteVersion, DeliverySequence, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, ?, ?)`,
		"evt-legacy-z-head", "source-legacy-ordinal-test", packInstall, "webhook.event", "legacy-head", payloadHead, headCreated, headCreated); err != nil {
		t.Fatalf("insert legacy head row: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
		 RouteVersion, DeliverySequence, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, ?, ?)`,
		"evt-legacy-a-tail", "source-legacy-ordinal-test", packInstall, "webhook.event", "legacy-tail", payloadTail, tailCreated, tailCreated); err != nil {
		t.Fatalf("insert legacy tail row: %v", err)
	}

	updated, err := migrateConnectorEventEnqueueOrdinalBackfill(ctx, db)
	if err != nil {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill: %v", err)
	}
	if updated != 2 {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill updated %d rows, want 2", updated)
	}

	svc := NewService(db, deterministicID("claim-token-head", "claim-token-tail"), func() time.Time {
		return tailCreated.Add(time.Hour)
	})
	evHead, ok, err := svc.ClaimNext(ctx, packInstall, "worker-legacy-head")
	if err != nil || !ok || evHead == nil {
		t.Fatalf("ClaimNext legacy head: ev=%v ok=%v err=%v", evHead, ok, err)
	}
	if evHead.ID != "evt-legacy-z-head" {
		t.Fatalf("first claim = %s, want evt-legacy-z-head", evHead.ID)
	}

	evBlocked, okBlocked, err := svc.ClaimNext(ctx, packInstall, "worker-legacy-blocked")
	if err != nil {
		t.Fatalf("ClaimNext blocked legacy tail: %v", err)
	}
	if okBlocked || evBlocked != nil {
		t.Fatalf("same-source legacy tail must stay blocked while head is processing; got ev=%v ok=%v", evBlocked, okBlocked)
	}

	if err := svc.MarkProcessed(ctx, evHead.ID, evHead.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed legacy head: %v", err)
	}

	evTail, ok, err := svc.ClaimNext(ctx, packInstall, "worker-legacy-tail")
	if err != nil || !ok || evTail == nil {
		t.Fatalf("ClaimNext legacy tail: ev=%v ok=%v err=%v", evTail, ok, err)
	}
	if evTail.ID != "evt-legacy-a-tail" {
		t.Fatalf("released claim = %s, want evt-legacy-a-tail", evTail.ID)
	}
}

func TestConnectorEventEnqueueOrdinalBackfillRerunsWhenReceiptExistsButLiveZerosRemain(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-legacy-ordinal-rerun-test")
	createConnectorEventTable(t, db)
	createMigrationTable(t, db)

	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `
	INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
	VALUES ('20260619_connector_event_enqueue_ordinal_backfill', 1, 'already stamped', NOW(), NOW(), NOW())`); err != nil {
		t.Fatalf("stamp migration receipt: %v", err)
	}

	const packInstall = "install-token-source-legacy-ordinal-rerun-test"
	created := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC)
	payload := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-shared-source","payload":{"status":"Created"}}`)
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
		 RouteVersion, DeliverySequence, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, ?, ?)`,
		"evt-legacy-rerun-zero", "source-legacy-ordinal-rerun-test", packInstall, "webhook.event", "legacy-rerun", payload, created, created); err != nil {
		t.Fatalf("insert post-receipt zero-ordinal row: %v", err)
	}

	updated, err := migrateConnectorEventEnqueueOrdinalBackfill(ctx, db)
	if err != nil {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill: %v", err)
	}
	if updated != 1 {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill updated %d rows, want 1", updated)
	}

	var ordinal int64
	if err := db.QueryRowContext(ctx, `SELECT EnqueueOrdinal FROM _System_ConnectorEvent WHERE Id = 'evt-legacy-rerun-zero'`).Scan(&ordinal); err != nil {
		t.Fatalf("query repaired ordinal: %v", err)
	}
	if ordinal == 0 {
		t.Fatal("post-receipt live row still has EnqueueOrdinal=0; want repaired non-zero ordinal")
	}
}

func TestConnectorEventEnqueueOrdinalBackfillRerunReordersMixedOrdinalSpace(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-legacy-ordinal-mixed-test")
	createConnectorEventTable(t, db)
	createMigrationTable(t, db)

	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `
	INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
	VALUES ('20260619_connector_event_enqueue_ordinal_backfill', 1, 'already stamped', NOW(), NOW(), NOW())`); err != nil {
		t.Fatalf("stamp migration receipt: %v", err)
	}

	const packInstall = "install-token-source-legacy-ordinal-mixed-test"
	payload := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-shared-source","payload":{"status":"Created"}}`)
	headCreated := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC)
	middleCreated := headCreated.Add(time.Minute)
	tailCreated := headCreated.Add(2 * time.Minute)

	for _, row := range []struct {
		id             string
		idempotency    string
		enqueueOrdinal int64
		created        time.Time
		lastModified   time.Time
	}{
		{id: "evt-mixed-z-head", idempotency: "mixed-head", enqueueOrdinal: 7, created: headCreated, lastModified: headCreated},
		{id: "evt-mixed-m-middle", idempotency: "mixed-middle", enqueueOrdinal: 8, created: middleCreated, lastModified: middleCreated},
		// Simulates an old pod writing after the first repair. Its Id is
		// lexicographically first, so a zero-subset-only rerun would move it ahead
		// of the already ordered rows.
		{id: "evt-mixed-a-tail", idempotency: "mixed-tail", enqueueOrdinal: 0, created: tailCreated, lastModified: tailCreated},
	} {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
			(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
			 RouteVersion, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, ?, ?, ?)`,
			row.id, "source-legacy-ordinal-mixed-test", packInstall, "webhook.event", row.idempotency, payload, row.enqueueOrdinal, row.created, row.lastModified); err != nil {
			t.Fatalf("insert mixed ordinal row %s: %v", row.id, err)
		}
	}

	updated, err := migrateConnectorEventEnqueueOrdinalBackfill(ctx, db)
	if err != nil {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill: %v", err)
	}
	if updated != 3 {
		t.Fatalf("MigrateConnectorEventEnqueueOrdinalBackfill updated %d rows, want 3", updated)
	}

	wantOrdinals := map[string]int64{
		"evt-mixed-z-head":   1,
		"evt-mixed-m-middle": 2,
		"evt-mixed-a-tail":   3,
	}
	for id, want := range wantOrdinals {
		var got int64
		if err := db.QueryRowContext(ctx, `SELECT EnqueueOrdinal FROM _System_ConnectorEvent WHERE Id = ?`, id).Scan(&got); err != nil {
			t.Fatalf("query ordinal for %s: %v", id, err)
		}
		if got != want {
			t.Fatalf("%s EnqueueOrdinal = %d, want %d", id, got, want)
		}
	}

	svc := NewService(db, deterministicID("claim-token-mixed-head", "claim-token-mixed-middle", "claim-token-mixed-tail"), func() time.Time {
		return tailCreated.Add(time.Hour)
	})
	for _, wantID := range []string{"evt-mixed-z-head", "evt-mixed-m-middle", "evt-mixed-a-tail"} {
		ev, ok, err := svc.ClaimNext(ctx, packInstall, "worker-"+wantID)
		if err != nil || !ok || ev == nil {
			t.Fatalf("ClaimNext %s: ev=%v ok=%v err=%v", wantID, ev, ok, err)
		}
		if ev.ID != wantID {
			t.Fatalf("ClaimNext = %s, want %s", ev.ID, wantID)
		}
		if err := svc.MarkProcessed(ctx, ev.ID, ev.ClaimToken); err != nil {
			t.Fatalf("MarkProcessed %s: %v", ev.ID, err)
		}
	}
}

func TestServiceClaimNextZeroOrdinalFallsBackToHistoricalOrder(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "source-zero-fallback-test")
	createConnectorEventTable(t, db)

	ctx := context.Background()
	const packInstall = "install-token-source-zero-fallback-test"
	payload := []byte(`{"objectKind":"workflow-actions","sourceID":"legacy-shared-source","payload":{"status":"Created"}}`)
	headCreated := time.Date(2026, 6, 19, 11, 0, 0, 0, time.UTC)
	tailCreated := headCreated.Add(time.Minute)

	for _, row := range []struct {
		id             string
		idempotency    string
		enqueueOrdinal int64
		created        time.Time
	}{
		{id: "evt-positive-z-head", idempotency: "positive-head", enqueueOrdinal: 1, created: headCreated},
		// Simulates an old binary writing after the new binary has already
		// emitted positive ordinals. The lexicographically smaller Id would win
		// the old equal/zero tiebreak unless the guard falls back to history.
		{id: "evt-zero-a-tail", idempotency: "zero-tail", enqueueOrdinal: 0, created: tailCreated},
	} {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
			(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, Payload, RetryCount,
			 RouteVersion, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
			VALUES (?, ?, ?, ?, ?, 'pending', ?, 0, 1, 0, ?, ?, ?)`,
			row.id, "source-zero-fallback-test", packInstall, "webhook.event", row.idempotency, payload, row.enqueueOrdinal, row.created, row.created); err != nil {
			t.Fatalf("insert zero fallback row %s: %v", row.id, err)
		}
	}

	svc := NewService(db, deterministicID("claim-token-zero-head", "claim-token-zero-tail"), func() time.Time {
		return tailCreated.Add(time.Hour)
	})
	evHead, ok, err := svc.ClaimNext(ctx, packInstall, "worker-zero-head")
	if err != nil || !ok || evHead == nil {
		t.Fatalf("ClaimNext positive head: ev=%v ok=%v err=%v", evHead, ok, err)
	}
	if evHead.ID != "evt-positive-z-head" {
		t.Fatalf("first claim = %s, want evt-positive-z-head", evHead.ID)
	}

	evBlocked, okBlocked, err := svc.ClaimNext(ctx, packInstall, "worker-zero-blocked")
	if err != nil {
		t.Fatalf("ClaimNext blocked zero tail: %v", err)
	}
	if okBlocked || evBlocked != nil {
		t.Fatalf("zero-ordinal tail must stay blocked while historical head is processing; got ev=%v ok=%v", evBlocked, okBlocked)
	}

	if err := svc.MarkProcessed(ctx, evHead.ID, evHead.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed positive head: %v", err)
	}
	evTail, ok, err := svc.ClaimNext(ctx, packInstall, "worker-zero-tail")
	if err != nil || !ok || evTail == nil {
		t.Fatalf("ClaimNext zero tail: ev=%v ok=%v err=%v", evTail, ok, err)
	}
	if evTail.ID != "evt-zero-a-tail" {
		t.Fatalf("released claim = %s, want evt-zero-a-tail", evTail.ID)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: cross-delivery parallelism — delivery A seq=0 is stuck in-flight;
// delivery B events must still drain without waiting.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ConcurrentClaim_CrossDeliveryParallel(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "cross-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "cross-test")

	// Deterministic sequential ids so delivery A's events sort before delivery B's:
	// A0 and B0 share CreatedDate AND DeliverySequence (both seq 0), so the row Id is
	// the only tiebreaker in (CreatedDate, DeliverySequence, Id). With random UUIDs the
	// "A is claimed first" assertions were ~50% flaky; sequential ids make them
	// deterministic without changing what the test proves (cross-delivery parallelism).
	var idMu sync.Mutex
	idn := 0
	idgen := func() string {
		idMu.Lock()
		defer idMu.Unlock()
		idn++
		return fmt.Sprintf("cross-ev-%04d", idn)
	}
	svc := NewService(db, idgen, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const packInstall = "install-token-cross-test"
	const dlvA = "dlv-cross-A"
	const dlvB = "dlv-cross-B"

	// Delivery A: seq=0, seq=1
	idA0, _, err := svc.Enqueue(ctx, "cross-test", "webhook.event", "a0", nil, packInstall, 1, "", "", 0, dlvA)
	if err != nil {
		t.Fatalf("Enqueue A0: %v", err)
	}
	_, _, err = svc.Enqueue(ctx, "cross-test", "webhook.event", "a1", nil, packInstall, 1, "", "", 1, dlvA)
	if err != nil {
		t.Fatalf("Enqueue A1: %v", err)
	}

	// Delivery B: seq=0, seq=1
	idB0, _, err := svc.Enqueue(ctx, "cross-test", "webhook.event", "b0", nil, packInstall, 1, "", "", 0, dlvB)
	if err != nil {
		t.Fatalf("Enqueue B0: %v", err)
	}
	idB1, _, err := svc.Enqueue(ctx, "cross-test", "webhook.event", "b1", nil, packInstall, 1, "", "", 1, dlvB)
	if err != nil {
		t.Fatalf("Enqueue B1: %v", err)
	}

	// Claim A0 and intentionally leave it in-flight (no MarkProcessed).
	evA0, okA0, err := svc.ClaimNext(ctx, packInstall, "worker-a")
	if err != nil || !okA0 || evA0 == nil || evA0.ID != idA0 {
		t.Fatalf("expected to claim A0 first: ev=%v ok=%v err=%v", evA0, okA0, err)
	}
	// A0 is now in "processing" — A1 must be blocked by NOT EXISTS guard.
	// B0 (different DeliveryId) must be claimable.
	evB0, okB0, err := svc.ClaimNext(ctx, packInstall, "worker-b0")
	if err != nil || !okB0 || evB0 == nil || evB0.ID != idB0 {
		t.Fatalf("expected to claim B0 while A is stuck: ev=%v ok=%v err=%v", evB0, okB0, err)
	}

	// Mark B0 processed; B1 should now unblock.
	if err := svc.MarkProcessed(ctx, evB0.ID, evB0.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed B0: %v", err)
	}
	evB1, okB1, err := svc.ClaimNext(ctx, packInstall, "worker-b1")
	if err != nil || !okB1 || evB1 == nil || evB1.ID != idB1 {
		t.Fatalf("expected to claim B1 after B0 processed: ev=%v ok=%v err=%v", evB1, okB1, err)
	}

	// A1 must still be blocked (A0 still processing).
	evCheck, okCheck, err := svc.ClaimNext(ctx, packInstall, "worker-check")
	if err != nil {
		t.Fatalf("ClaimNext for A1 check: %v", err)
	}
	if okCheck {
		t.Errorf("expected no claimable event while A0 in-flight, got %v", evCheck)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: NULL DeliveryId rows are independent singletons — each claimable in
// any order, no cross-row blocking.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ConcurrentClaim_NullDeliveryIdSingleton(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "null-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "null-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const packInstall = "install-token-null-test"

	// Two NULL-DeliveryId rows (legacy: pass empty deliveryID).
	idN0, _, err := svc.Enqueue(ctx, "null-test", "webhook.event", "n0", nil, packInstall, 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue N0: %v", err)
	}
	idN1, _, err := svc.Enqueue(ctx, "null-test", "webhook.event", "n1", nil, packInstall, 1, "", "", 1, "")
	if err != nil {
		t.Fatalf("Enqueue N1: %v", err)
	}
	// One row with a real DeliveryId.
	idD0, _, err := svc.Enqueue(ctx, "null-test", "webhook.event", "d0", nil, packInstall, 1, "", "", 0, "dlv-null-real")
	if err != nil {
		t.Fatalf("Enqueue D0: %v", err)
	}

	// All three must be independently claimable WITHOUT processing in between:
	// claim each and LEAVE it in-flight (processing). If the guard wrongly treated
	// NULL=NULL as a delivery match, the second NULL row would be blocked by the
	// first while it is processing and this loop would stall. The three claims must
	// all succeed and form the exact set {idN0, idN1, idD0}.
	claimed := map[string]bool{}
	for i := 0; i < 3; i++ {
		ev, ok, err := svc.ClaimNext(ctx, packInstall, fmt.Sprintf("null-worker-%d", i))
		if err != nil {
			t.Fatalf("ClaimNext %d: %v", i, err)
		}
		if !ok || ev == nil {
			t.Fatalf("ClaimNext %d: expected a claimable event while others are in-flight, got ok=%v ev=%v", i, ok, ev)
		}
		if claimed[ev.ID] {
			t.Errorf("event %s claimed twice", ev.ID)
		}
		claimed[ev.ID] = true
	}

	for _, wantID := range []string{idN0, idN1, idD0} {
		if !claimed[wantID] {
			t.Errorf("event %s was never claimed", wantID)
		}
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: a replayed dead event REJOINS its original delivery's ordering group, so
// it cannot run out-of-order relative to still-pending siblings (#2154 replay path).
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ReplayDeadEvent_RejoinsDeliveryOrdering(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "replay-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "replay-test")

	svc := NewService(db, nil, nil)
	ctx := context.Background()

	const packInstall = "install-token-replay-test"
	const dlv = "dlv-replay-001"

	// Delivery D: seq0 (will dead-letter) and seq1 (its dependent, still pending).
	id0, _, err := svc.Enqueue(ctx, "replay-test", "webhook.event", "rk0", nil, packInstall, 1, "", "", 0, dlv)
	if err != nil {
		t.Fatalf("Enqueue seq0: %v", err)
	}
	if _, _, err := svc.Enqueue(ctx, "replay-test", "webhook.event", "rk1", nil, packInstall, 1, "", "", 1, dlv); err != nil {
		t.Fatalf("Enqueue seq1: %v", err)
	}

	// seq0 dead-letters (simulate exhausted retries).
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='dead' WHERE Id=?`, id0); err != nil {
		t.Fatalf("mark seq0 dead: %v", err)
	}

	// Replay seq0 → a new pending row. It MUST inherit DeliveryId=D and seq=0 so it
	// rejoins the delivery ordering rather than becoming an independent NULL singleton.
	newID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), id0, packInstall, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}

	// ClaimNext must hand out the REPLAYED seq0 (the delivery head), not seq1.
	ev, ok, err := svc.ClaimNext(ctx, packInstall, "replay-worker")
	if err != nil || !ok || ev == nil {
		t.Fatalf("ClaimNext after replay: ev=%v ok=%v err=%v", ev, ok, err)
	}
	if ev.ID != newID {
		t.Fatalf("ClaimNext returned %s, want the replayed seq0 %s — replay did not rejoin the delivery ordering (seq1 was claimable out of order)", ev.ID, newID)
	}

	// While the replayed seq0 is in flight, seq1 must NOT be claimable.
	ev2, ok2, err := svc.ClaimNext(ctx, packInstall, "replay-worker-2")
	if err != nil {
		t.Fatalf("ClaimNext for seq1 block check: %v", err)
	}
	if ok2 {
		t.Errorf("seq1 was claimable while the replayed seq0 is in flight (got %v); a replayed event must block its later siblings", ev2)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: EnqueueBatch dedups within and across batches inside the tx — a batch
// event reusing an already-enqueued idempotency key returns dup=true with the
// existing id, and both events still commit (the dup is not an error).
// ─────────────────────────────────────────────────────────────────────────────

func TestService_EnqueueBatch_DedupWithinAndAcrossBatches(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "atom-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "atom-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "atom-test"
	const packInstall = "install-token-atom-test"
	const dlv = "dlv-atom-001"

	// First, insert one event so its idempotency key is known to exist.
	events1 := []BatchEvent{
		{EventKind: "webhook.event", IdempotencyKey: "atom-ikey-seed", Payload: nil, RouteVersion: 1, DeliverySequence: 0},
	}
	ids1, dups1, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, events1)
	if err != nil {
		t.Fatalf("seed EnqueueBatch: %v", err)
	}
	if len(ids1) != 1 || dups1[0] {
		t.Fatalf("seed: want 1 fresh id, got ids=%v dups=%v", ids1, dups1)
	}

	// Now enqueue a batch: first event is fresh, second re-uses the seed key (dup).
	// EnqueueBatch must handle the dup gracefully (return dup=true) and commit both.
	events2 := []BatchEvent{
		{EventKind: "webhook.event", IdempotencyKey: "atom-ikey-new", Payload: nil, RouteVersion: 1, DeliverySequence: 1},
		{EventKind: "webhook.event", IdempotencyKey: "atom-ikey-seed", Payload: nil, RouteVersion: 1, DeliverySequence: 2},
	}
	ids2, dups2, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, events2)
	if err != nil {
		t.Fatalf("second EnqueueBatch: %v", err)
	}
	if len(ids2) != 2 {
		t.Fatalf("want 2 ids, got %v", ids2)
	}
	if dups2[0] {
		t.Errorf("first event (fresh key) should not be dup")
	}
	if !dups2[1] {
		t.Errorf("second event (seed key) should be dup")
	}
	// The dup's returned id should be the seed's id, not a new row.
	if ids2[1] != ids1[0] {
		t.Errorf("dup id = %q, want seed id %q", ids2[1], ids1[0])
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: EnqueueBatch is ATOMIC — when one event in a batch fails its route/pack
// guard (INSERT…SELECT writes 0 rows), the WHOLE batch rolls back, so NO earlier
// event of the batch persists. This all-or-nothing property is what removes the
// partial-enqueue → reordered-retry DeliverySequence collision at the source:
// a delivery is never half-enqueued, so a retry sees all-or-none. Without the
// single-tx rollback, event 0 (which has a live route) would persist.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_EnqueueBatch_AtomicAllOrNothing(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "atomroll-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "atomroll-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "atomroll-test"
	const packInstall = "install-token-atomroll-test"
	const dlv = "dlv-atomroll-001"

	// event 0: webhook.event @ RouteVersion 1 — the seeded live route exists, so on
	//          its own this INSERT…SELECT would write 1 row.
	// event 1: RouteVersion 2 — no such route row → its INSERT…SELECT writes 0 rows →
	//          ErrPackIncarnationChanged → the whole batch (incl. event 0) rolls back.
	batch := []BatchEvent{
		{EventKind: "webhook.event", IdempotencyKey: "atomroll-ok", Payload: nil, RouteVersion: 1, DeliverySequence: 0},
		{EventKind: "webhook.event", IdempotencyKey: "atomroll-bad", Payload: nil, RouteVersion: 2, DeliverySequence: 1},
	}
	ids, dups, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, batch)
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("EnqueueBatch err = %v (ids=%v dups=%v), want ErrPackIncarnationChanged", err, ids, dups)
	}

	// Atomicity: NEITHER event persisted. event 0 would have succeeded alone, but a
	// failed batch must roll back ALL events — so its row must be absent too.
	for _, key := range []string{"atomroll-ok", "atomroll-bad"} {
		var count int
		if qerr := db.QueryRowContext(ctx,
			`SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId = ? AND IdempotencyKey = ?`,
			packInstall, key).Scan(&count); qerr != nil {
			t.Fatalf("count %q: %v", key, qerr)
		}
		if count != 0 {
			t.Errorf("event %q persisted (count=%d) — batch was NOT atomic; a failed batch must roll back ALL events", key, count)
		}
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: concurrent EnqueueBatch calls for the same DeliveryId do not corrupt
// DeliverySequence ordering — each call acquires the pack row lock, so the
// batches serialize and all events land with distinct sequences.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ConcurrentClaim_EnqueueBatchParallel(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "par-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "par-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "par-test"
	const packInstall = "install-token-par-test"
	const goroutines = 4
	const eventsPerBatch = 3

	var wg sync.WaitGroup
	errs := make(chan error, goroutines)
	start := make(chan struct{})

	for g := 0; g < goroutines; g++ {
		g := g
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			dlv := fmt.Sprintf("dlv-par-%d", g)
			batch := make([]BatchEvent, eventsPerBatch)
			for i := range batch {
				batch[i] = BatchEvent{
					EventKind:        "webhook.event",
					IdempotencyKey:   fmt.Sprintf("par-ikey-%d-%d", g, i),
					Payload:          nil,
					RouteVersion:     1,
					DeliverySequence: i,
				}
			}
			_, _, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, batch)
			if err != nil {
				errs <- fmt.Errorf("goroutine %d: %w", g, err)
			}
		}()
	}
	close(start)
	wg.Wait()
	close(errs)

	for err := range errs {
		t.Errorf("EnqueueBatch parallel: %v", err)
	}

	// All goroutines × events must be claimable and processable without deadlock.
	const total = goroutines * eventsPerBatch
	var processed int32
	workerCtx, workerCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer workerCancel()

	var wg2 sync.WaitGroup
	for w := 0; w < 4; w++ {
		wg2.Add(1)
		wid := fmt.Sprintf("par-worker-%d", w)
		go func(workerID string) {
			defer wg2.Done()
			for {
				if workerCtx.Err() != nil {
					return
				}
				if atomic.LoadInt32(&processed) >= int32(total) {
					return
				}
				ev, claimed, err := svc.ClaimNext(workerCtx, packInstall, workerID)
				if err != nil || !claimed || ev == nil {
					time.Sleep(time.Millisecond)
					continue
				}
				if err := svc.MarkProcessed(workerCtx, ev.ID, ev.ClaimToken); err != nil {
					t.Errorf("MarkProcessed: %v", err)
					return
				}
				atomic.AddInt32(&processed, 1)
			}
		}(wid)
	}
	wg2.Wait()

	if got := atomic.LoadInt32(&processed); got != int32(total) {
		t.Errorf("processed=%d, want %d", got, total)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: with atomic enqueue, strict DeliverySequence < is sufficient to order
// events within a DeliveryId — equal sequences no longer arise, and replay
// ordering is deterministic (seq 0 claimed before seq 1 before seq 2).
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ReplayOrdering_StrictSequence(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "replay-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "replay-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "replay-test"
	const packInstall = "install-token-replay-test"
	const dlv = "dlv-replay-001"
	const count = 5

	// EnqueueBatch with distinct DeliverySequences 0..count-1 in one atomic tx.
	batch := make([]BatchEvent, count)
	for i := range batch {
		batch[i] = BatchEvent{
			EventKind:        "webhook.event",
			IdempotencyKey:   fmt.Sprintf("replay-ikey-%d", i),
			Payload:          nil,
			RouteVersion:     1,
			DeliverySequence: i,
		}
	}
	ids, dups, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, batch)
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	for i, d := range dups {
		if d {
			t.Errorf("event[%d] unexpectedly dup (id=%s)", i, ids[i])
		}
	}

	// Claim and process events; verify they come out in sequence order.
	var gotSeqs []int
	workerID := "replay-worker"
	for len(gotSeqs) < count {
		ev, claimed, err := svc.ClaimNext(ctx, packInstall, workerID)
		if err != nil {
			t.Fatalf("ClaimNext: %v", err)
		}
		if !claimed || ev == nil {
			time.Sleep(time.Millisecond)
			continue
		}
		gotSeqs = append(gotSeqs, ev.DeliverySequence)
		if err := svc.MarkProcessed(ctx, ev.ID, ev.ClaimToken); err != nil {
			t.Fatalf("MarkProcessed: %v", err)
		}
	}

	for i, seq := range gotSeqs {
		if seq != i {
			t.Errorf("gotSeqs[%d]=%d, want %d — replay ordering not strict", i, seq, i)
		}
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: the claim is atomic with the per-delivery ordering check. ClaimNext's
// candidate SELECT picks seq1 because seq0 is dead; a concurrent ReplayDeadEvent
// then re-introduces a pending seq0 row in the window BEFORE the claim UPDATE runs.
// The UPDATE's ordering re-check (the NOT EXISTS) must reject the seq1 claim, and
// the loop must re-select and claim the freshly-replayed seq0 head instead — so a
// replayed prerequisite never executes after its later sibling. The race is made
// deterministic by the afterClaimSelect test seam.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ClaimNext_ReplayRaceDoesNotClaimLaterSibling(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "claimrace-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "claimrace-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "claimrace-test"
	const packInstall = "install-token-claimrace-test"
	const dlv = "dlv-claimrace-001"

	// Delivery [E0 seq0, E1 seq1]; drive E0 to dead so the claim SELECT (which skips
	// terminal siblings) sees E1 as the only claimable candidate.
	ids, _, err := svc.EnqueueBatch(ctx, ns, packInstall, dlv, []BatchEvent{
		{EventKind: "webhook.event", IdempotencyKey: "claimrace-e0", RouteVersion: 1, DeliverySequence: 0},
		{EventKind: "webhook.event", IdempotencyKey: "claimrace-e1", RouteVersion: 1, DeliverySequence: 1},
	})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	e0ID := ids[0]
	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='dead' WHERE Id=?`, e0ID); err != nil {
		t.Fatalf("mark E0 dead: %v", err)
	}

	// Seam: on the FIRST claim attempt (after the SELECT picked seq1), a concurrent
	// admin replays E0, re-introducing a pending seq0 head. once.Do keeps the replay
	// to the first iteration so the retry can claim the new seq0 row.
	var once sync.Once
	svc.afterClaimSelect = func() {
		once.Do(func() {
			if _, rerr := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), e0ID, packInstall, 1); rerr != nil {
				t.Errorf("ReplayDeadEvent in seam: %v", rerr)
			}
		})
	}

	ev, claimed, err := svc.ClaimNext(ctx, packInstall, "claimrace-worker")
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if !claimed || ev == nil {
		t.Fatalf("expected a claim, got claimed=%v ev=%v", claimed, ev)
	}
	if ev.DeliverySequence != 0 {
		t.Errorf("ClaimNext claimed DeliverySequence=%d (id=%s), want 0 — a concurrent replay re-introduced the seq0 head between the SELECT and the UPDATE; the claim UPDATE's ordering re-check must reject the later seq1 sibling (TOCTOU)", ev.DeliverySequence, ev.ID)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Test: a CROSS-incarnation replay does not falsely rejoin a delivery group. When
// the dead row was enqueued under a superseded PackInstallId and the admin replay
// resolves the CURRENT active incarnation, the replay row must be a NULL-DeliveryId
// singleton — the original delivery's other members live under a different
// InstalledPackId that ClaimNext (incarnation-scoped) cannot co-order with, and they
// are stranded by the rebirth lifecycle. Copying the DeliveryId would advertise a
// group this row can never serialize against. (Same-incarnation replay still rejoins —
// see TestService_ReplayDeadEvent_RejoinsDeliveryOrdering.)
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ReplayDeadEvent_CrossIncarnation_DropsDeliveryGroup(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "xinc-test") // active incarnation = install-token-xinc-test
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "xinc-test")

	svc := NewService(db, nil, nil)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	const ns = "xinc-test"
	const packB = "install-token-xinc-test"   // current active incarnation (replay target)
	const packAOld = "install-token-xinc-OLD" // superseded incarnation the dead row was under
	const dlv = "dlv-xinc-001"

	// A dead event from a SUPERSEDED incarnation, carrying a DeliveryId from its delivery.
	deadID := "dead-xinc-0"
	if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RouteVersion, DeliveryId, DeliverySequence, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'webhook.event', 'xinc-ikey-0', 'dead', 1, ?, 0, NOW(), NOW())`,
		deadID, ns, packAOld, dlv); err != nil {
		t.Fatalf("insert dead row: %v", err)
	}

	// Admin replay resolves the CURRENT active incarnation (packB) — cross-incarnation.
	newID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), deadID, packB, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}

	var deliveryID sql.NullString
	var seq int
	if err := db.QueryRowContext(ctx,
		`SELECT DeliveryId, DeliverySequence FROM _System_ConnectorEvent WHERE Id = ?`, newID).Scan(&deliveryID, &seq); err != nil {
		t.Fatalf("load replay row: %v", err)
	}
	if deliveryID.Valid {
		t.Errorf("cross-incarnation replay row has DeliveryId=%q, want NULL — it must NOT copy the superseded incarnation's delivery group", deliveryID.String)
	}
	if seq != 0 {
		t.Errorf("cross-incarnation replay row has DeliverySequence=%d, want 0 (singleton)", seq)
	}
}

func TestService_ReplayDeadEvent_SameSourceCrossDeliveryGetsFreshEnqueueOrdinal(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	createInstalledPackTable(t, db, "replay-test")
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, "replay-test")

	svc := NewService(db, nil, nil)
	ctx := context.Background()

	const packInstall = "install-token-replay-test"

	payload0 := json.RawMessage(`{"objectKind":"Contact","sourceID":"src-123","phase":"sent"}`)
	payload1 := json.RawMessage(`{"objectKind":"Contact","sourceID":"src-123","phase":"responded"}`)

	id0, _, err := svc.Enqueue(ctx, "replay-test", "webhook.event", "replay-src-0", payload0, packInstall, 1, "", "", 0, "dlv-replay-src-0")
	if err != nil {
		t.Fatalf("Enqueue dead row: %v", err)
	}
	liveID, _, err := svc.Enqueue(ctx, "replay-test", "webhook.event", "replay-src-1", payload1, packInstall, 1, "", "", 0, "dlv-replay-src-1")
	if err != nil {
		t.Fatalf("Enqueue live row: %v", err)
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorEvent SET Status='dead' WHERE Id=?`, id0); err != nil {
		t.Fatalf("mark dead row dead: %v", err)
	}

	replayID, err := svc.ReplayDeadEvent(ctx, allowPerm(string(constants.PermReplayConnectorEvents)), id0, packInstall, 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}

	ev, ok, err := svc.ClaimNext(ctx, packInstall, "replay-same-source-worker")
	if err != nil || !ok || ev == nil {
		t.Fatalf("ClaimNext after replay: ev=%v ok=%v err=%v", ev, ok, err)
	}
	if ev.ID != liveID {
		t.Fatalf("ClaimNext returned %s, want existing live same-source row %s before replay %s", ev.ID, liveID, replayID)
	}
}

// WorkId names a generation; ProjectionKey names its source identity.
func sourceProjectionKeyFromWorkID(workID string) string {
	parts := strings.Split(workID, ":")
	return strings.Join(parts[:2], ":")
}