package connector_events

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
)

// deterministicID returns a factory that cycles through the provided ids.
func deterministicID(ids ...string) func() string {
	i := 0
	return func() string {
		id := ids[i%len(ids)]
		i++
		return id
	}
}

// fixedNow returns a time source that always returns t.
func fixedNow(t time.Time) func() time.Time {
	return func() time.Time { return t }
}

func TestClaimOrderingGuardUsesMaterializedSourceLane(t *testing.T) {
	guardSQL, _ := buildClaimOrderingGuardSQL(time.Now())
	for _, want := range []string{
		"b.SourceLaneKey = e.SourceLaneKey",
		"b.EnqueueOrdinal < e.EnqueueOrdinal",
		"NULLIF(e.SourceLaneKey, '') IS NULL",
		"JSON_EXTRACT(e.Payload, '$.objectKind')",
		// DELIVERY branch stays strict FIFO with the bare non-terminal status check.
		"b.Status NOT IN ('processed', 'skipped', 'dead')",
		// SOURCE-LANE branches carry the NextAttemptAt-gated blocker (NCRM #3139).
		"b.Status NOT IN ('processed', 'skipped', 'dead', 'pending', 'failed')",
		"b.NextAttemptAt IS NULL OR b.NextAttemptAt <= ?",
	} {
		if !strings.Contains(guardSQL, want) {
			t.Fatalf("claim ordering guard missing %q:\n%s", want, guardSQL)
		}
	}
}

func TestClaimCandidateStatusGatesPendingNotBefore(t *testing.T) {
	if !strings.Contains(claimCandidateStatusSQL, "e.Status = 'pending' AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?)") {
		t.Fatalf("claim candidate predicate does not gate pending NextAttemptAt:\n%s", claimCandidateStatusSQL)
	}
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	settleAt := now.Add(connector_inbound_routing.DefaultSourceSettleDelay)
	state, runAfter, ok := workItemInitialState(statusPending, sql.NullTime{Time: settleAt, Valid: true}, now)
	if !ok || state != connector_workqueue.WorkStatePending || !runAfter.Equal(settleAt) {
		t.Fatalf("pending initial state = (%q, %s, %v), want pending at %s", state, runAfter, ok, settleAt)
	}
}

func TestLoadEventsForWorkItemsHydratesCandidateWindowInTwoQueries(t *testing.T) {
	svc, mock := newMock(t)
	now := time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC)
	nextAttempt := now.Add(time.Minute)
	columns := []string{
		"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
		"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
		"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
		"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey",
		"DeliveryId", "DeliverySequence", "CreatedDate", "LastModifiedDate",
	}
	rows := sqlmock.NewRows(columns).
		AddRow("event-1", "test", "pack-1", "noop.skip", "key-1", statusPending, "bulk", []byte(`{}`), 0, nil, nil, nil, nil, nil, nil, nil, nil, int64(1), nil, nil, "lane-1", nil, 0, now, now).
		AddRow("event-2", "test", "pack-1", "noop.skip", "key-2", statusFailed, "bulk", []byte(`{}`), 1, nil, nil, nil, nil, nil, nil, nil, nil, int64(1), nil, nil, "lane-2", nil, 0, now, now)
	mock.ExpectQuery(`(?s)SELECT Id, Namespace.*FROM _System_ConnectorEvent.*WHERE Id IN \(\?,\?\)`).
		WithArgs("event-1", "event-2").
		WillReturnRows(rows)
	mock.ExpectQuery(`(?s)SELECT Id, NextAttemptAt.*WHERE Id IN \(\?,\?\).*NextAttemptAt IS NOT NULL`).
		WithArgs("event-1", "event-2").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "NextAttemptAt"}).AddRow("event-2", nextAttempt))

	events, err := svc.LoadEventsForWorkItems(context.Background(), []string{"event-1", "event-2", "event-1"})
	if err != nil {
		t.Fatalf("LoadEventsForWorkItems: %v", err)
	}
	if len(events) != 2 || events["event-1"] == nil || events["event-2"] == nil {
		t.Fatalf("events = %#v, want two hydrated rows", events)
	}
	if events["event-1"].NextAttemptAt != nil || events["event-2"].NextAttemptAt == nil || !events["event-2"].NextAttemptAt.Equal(nextAttempt) {
		t.Fatalf("next attempts = event-1:%v event-2:%v", events["event-1"].NextAttemptAt, events["event-2"].NextAttemptAt)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

// TestClaimOrderingGuardBindCount pins the (sql, binds) invariant: the number of
// returned binds exactly equals the number of `?` placeholders in the built SQL,
// for every guard builder. This catches positional-arg drift when a new bind is
// added to one source-lane branch but not threaded into a call site.
func TestClaimOrderingGuardBindCount(t *testing.T) {
	now := time.Now()
	cases := []struct {
		name string
		sql  string
		bind []interface{}
	}{}
	g, gb := buildClaimOrderingGuard(now)
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"guard", g, gb})
	gs, gsb := buildClaimOrderingGuardSQL(now)
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"guardSQL", gs, gsb})
	rs, rsb := buildClaimOrderingResidualSQL(now)
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"residual", rs, rsb})
	lh, lhb := buildLaneHeadJoin("pack", now)
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"laneHead", lh, lhb})
	dh, dhb := buildDeliveryHeadJoin("pack")
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"deliveryHead", dh, dhb})
	ec, ecb := buildClaimOrderingGuardExceptCurrentSQL("cur", now, false)
	cases = append(cases, struct {
		name string
		sql  string
		bind []interface{}
	}{"exceptCurrent", ec, ecb})

	for _, c := range cases {
		if got, want := len(c.bind), strings.Count(c.sql, "?"); got != want {
			t.Fatalf("%s: %d binds but %d `?` placeholders:\n%s", c.name, got, want, c.sql)
		}
	}
}

// TestClaimOrderingGuardExceptCurrentBindOrder pins the EXACT bind ORDER for the
// coalesce guard (count alone cannot catch a now/currentEventID swap). Both
// source-lane branches are gated with (currentEventID, now). The versioned
// source-scoped handoff omits the legacy delivery-wide predecessor branch.
func TestClaimOrderingGuardExceptCurrentBindOrder(t *testing.T) {
	now := time.Unix(1000, 0)
	_, binds := buildClaimOrderingGuardExceptCurrentSQL("CUR", now, true)
	want := []interface{}{"CUR", now, "CUR", now}
	if len(binds) != len(want) {
		t.Fatalf("got %d binds, want %d: %v", len(binds), len(want), binds)
	}
	for i := range want {
		if binds[i] != want[i] {
			t.Fatalf("bind[%d] = %v, want %v (full: %v)", i, binds[i], want[i], binds)
		}
	}
}

func TestClaimOrderingGuardExceptCurrentIgnoresCrossSourceDeliveryPredecessors(t *testing.T) {
	guard, _ := buildClaimOrderingGuardExceptCurrentSQL("CUR", time.Unix(1000, 0), true)
	if strings.Contains(guard, "b.DeliveryId = e.DeliveryId") {
		t.Fatalf("source-scoped coalesce guard still contains delivery-wide FIFO:\n%s", guard)
	}
	if got := strings.Count(guard, "b.SourceLaneKey = e.SourceLaneKey"); got != 1 {
		t.Fatalf("source-lane guard count = %d, want 1:\n%s", got, guard)
	}
	if got := strings.Count(guard, "JSON_EXTRACT(b.Payload, '$.sourceID')"); got != 1 {
		t.Fatalf("JSON source fallback guard count = %d, want 1:\n%s", got, guard)
	}
}

func TestClaimOrderingGuardExceptCurrentLegacyModeKeepsDeliveryPredecessors(t *testing.T) {
	guard, _ := buildClaimOrderingGuardExceptCurrentSQL("CUR", time.Unix(1000, 0), false)
	if !strings.Contains(guard, "b.DeliveryId = e.DeliveryId") {
		t.Fatalf("legacy coalesce guard lost delivery-wide FIFO:\n%s", guard)
	}
}

func TestSourceLaneFromPayload(t *testing.T) {
	tests := []struct {
		name string
		in   string
		want sourceLane
	}{
		{
			name: "preserves exact envelope values",
			in:   `{"objectKind":" lead ","sourceID":" 00Q123 "}`,
			want: sourceLane{ObjectKind: " lead ", SourceID: " 00Q123 ", Key: sourceLaneKeyFor(" lead ", " 00Q123 ")},
		},
		{
			name: "missing source does not invent lane",
			in:   `{"objectKind":"lead"}`,
			want: sourceLane{ObjectKind: "lead"},
		},
		{
			name: "invalid json",
			in:   `{`,
			want: sourceLane{},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := sourceLaneFromPayload(json.RawMessage(tt.in)); got != tt.want {
				t.Fatalf("sourceLaneFromPayload() = %+v, want %+v", got, tt.want)
			}
		})
	}
}

func TestSourceLaneKeyForPreservesWhitespaceIdentity(t *testing.T) {
	if got, wantDifferentFrom := sourceLaneKeyFor("lead", " 00Q123 "), sourceLaneKeyFor("lead", "00Q123"); got == wantDifferentFrom {
		t.Fatalf("sourceLaneKeyFor must keep QueueEnvelope sourceID whitespace significant")
	}
}

func TestEnqueueBatch_SourceLaneOverridePreservesSourceIDWhitespaceIdentity(t *testing.T) {
	svc, mock := newMock(t, "id-lane-space")
	const (
		namespace    = "hubspot-crm-connector"
		installID    = "install-token-test"
		eventKind    = "workflow.action.execute"
		idempotency  = "callback-space"
		contactID    = "contact-123"
		parentID     = " parent-001 "
		routeVersion = int64(1)
		deliveryID   = "dlv-lane-space-test"
	)
	expectedLaneKey := sourceLaneKeyFor("shared-parent", parentID)
	payload := json.RawMessage(`{"objectKind":"workflow-actions","sourceID":"` + contactID + `","payload":{"parent_id":"` + parentID + `"}}`)

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, namespace, installID)
	expectNextEnqueueOrdinal(mock, installID, 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs(installID, idempotency).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute`).
		WithArgs(
			"id-lane-space",
			namespace,
			installID,
			eventKind,
			idempotency,
			statusPending,
			sqlmock.AnyArg(),
			connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
			installID,
			eventKind,
			routeVersion,
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			"workflow-actions",
			contactID,
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			"workflow-actions",
			contactID,
			expectedLaneKey,
			routeVersion,
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			deliveryID,
			0,
			int64(1),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			installID,
			eventKind,
			routeVersion,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectConnectorWorkItemMaterialization(mock, "id-lane-space", installID, "default", statusPending, expectedLaneKey, "")
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	_, _, err := svc.EnqueueBatch(context.Background(), namespace, installID, deliveryID, []BatchEvent{
		{
			EventKind:            eventKind,
			IdempotencyKey:       idempotency,
			Payload:              payload,
			RouteVersion:         routeVersion,
			SourceLaneObjectKind: "shared-parent",
			SourceLaneID:         parentID,
		},
	})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet expectations: %v", err)
	}
}

// packRow is the row returned by ResolveConnectorPackByNamespace.
// id is the _System_InstalledPack.Id (deterministic from Name).
// installToken is the _System_InstalledPack.PackInstallId (per-incarnation identity).
func packRow(id, installToken, namespace string) *sqlmock.Rows {
	return sqlmock.NewRows([]string{"Id", "PackInstallId", "Name", "Category", "Status", "COALESCE(AppliedPayloadHash, '')", "COALESCE(PackVersion, '')"}).
		AddRow(id, installToken, namespace, "connector-pack", "active", "", "")
}

// noPackRow returns an empty result — pack not found.
func noPackRow() *sqlmock.Rows {
	return sqlmock.NewRows([]string{"Id", "PackInstallId", "Name", "Category", "Status", "COALESCE(AppliedPayloadHash, '')", "COALESCE(PackVersion, '')"})
}

// eventRow returns all columns for a ConnectorEvent scan (includes InstalledPackId + ClaimToken + RouteVersion + companion/source-lane columns).
// packID is the _System_InstalledPack.Id for the incarnation that owns this event.
// routeVersion follows three-state semantics: nil=legacy NULL, 0=no-route sentinel, >=1=specific version.
func eventRow(id, ns, packID, kind, ikey, status string, retry int, replayOf string) *sqlmock.Rows {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	return sqlmock.NewRows([]string{
		"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
		"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
		"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
		"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey",
		"DeliveryId", "DeliverySequence", "CreatedDate", "LastModifiedDate",
	}).AddRow(
		id, ns, packID, kind, ikey, status,
		"default", []byte(`{}`), retry, nil, nil,
		nil, nil, nil, nil, nil, nullStrPtr(replayOf),
		int64(1),      // default to version=1 for happy-path event rows
		nil, nil, nil, // CompanionPackInstallId, CompanionFlowVersionId, SourceLaneKey (same-pack/legacy)
		nil, 0, now, now, // DeliveryId (nil=NULL legacy), DeliverySequence
	)
}

func deadWorkItemViewRow(workID, eventID, ns, packID, kind, ikey, state string, failureCount int, legacy bool) *sqlmock.Rows {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	failureClass := "inbound_execution_failed"
	failureSummary := "safe operator summary"
	legacyValue := 0
	if legacy {
		failureClass = ""
		failureSummary = ""
		legacyValue = 1
	}
	return sqlmock.NewRows([]string{
		"WorkId", "EventId", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey",
		"State", "FailureCount", "FailureClass", "FailureSummary", "LastFailureAt",
		"TerminalAt", "ReplayedAt", "ReplayOfWorkId", "ReplayOfId", "ProcessedAt",
		"DeliveryId", "DeliverySequence", "CreatedDate", "LastModifiedDate", "LegacyEvent",
	}).AddRow(
		workID, eventID, ns, packID, kind, ikey,
		state, failureCount, failureClass, failureSummary, now,
		now, nil, "", "", nil,
		"", 0, now, now, legacyValue,
	)
}

func expectLegacyReplayOwnershipProbe(mock sqlmock.Sqlmock, eventID string) {
	mock.ExpectQuery(`(?s)SELECT w.WorkId, COALESCE\(w.ProjectionKey, ''\), w.State, w.ReplayOfWorkId, w.ReplayedAt, w.Attempt.*FROM _System_ConnectorWorkItem w.*w.WorkType IN \(\?, \?\).*LIMIT 2.*FOR UPDATE`).
		WithArgs(eventID, connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, connector_workqueue.WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}))
}

func expectReplayCutoverReceiptMissing(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(`(?s)SELECT Version, Description.*FROM _System_Migration.*WHERE Id = \?.*LIMIT 1`).
		WithArgs(WorkItemOnlyExecutionCutoverMigrationID).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}))
}

func expectWorkItemReplayReservation(mock sqlmock.Sqlmock, eventID, workID, projectionKey string) {
	mock.ExpectQuery(`(?s)SELECT w.WorkId, COALESCE\(w.ProjectionKey, ''\), w.State, w.ReplayOfWorkId, w.ReplayedAt, w.Attempt.*FROM _System_ConnectorWorkItem w.*w.WorkType IN \(\?, \?\).*LIMIT 2.*FOR UPDATE`).
		WithArgs(eventID, connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, connector_workqueue.WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow(workID, projectionKey, connector_workqueue.WorkStateDead, nil, nil, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET ReplayedAt = \?.*WHERE WorkId = \?.*EventId = \?.*State = \?.*ReplayedAt IS NULL`).
		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg(), workID, eventID, connector_workqueue.WorkStateDead).
		WillReturnResult(sqlmock.NewResult(0, 1))
}

func expectZeroAttemptReplayReservation(mock sqlmock.Sqlmock, eventID, workID, state string) {
	mock.ExpectQuery(`(?s)SELECT w.WorkId, COALESCE\(w.ProjectionKey, ''\), w.State, w.ReplayOfWorkId, w.ReplayedAt, w.Attempt.*FROM _System_ConnectorWorkItem w.*w.WorkType IN \(\?, \?\).*LIMIT 2.*FOR UPDATE`).
		WithArgs(eventID, connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, connector_workqueue.WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "ProjectionKey", "State", "ReplayOfWorkId", "ReplayedAt", "Attempt"}).
			AddRow(workID, "", state, nil, nil, 0))
}

func expectZeroAttemptReplayFence(mock sqlmock.Sqlmock, eventID, workID string, rowsAffected int64) {
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*TerminalAt = COALESCE\(TerminalAt, \?\).*ReplayedAt = \?.*ProjectionActiveKey = NULL.*WHERE WorkId = \?.*EventId = \?.*Attempt = 0.*ReplayedAt IS NULL`).
		WithArgs(
			connector_workqueue.WorkStateDead,
			sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),
			workID, eventID,
			connector_workqueue.WorkStatePending,
			connector_workqueue.WorkStateRetry,
			connector_workqueue.WorkStateBlocked,
			connector_workqueue.WorkStateBlockedRetry,
			connector_workqueue.WorkStateDead,
		).
		WillReturnResult(sqlmock.NewResult(0, rowsAffected))
}

func expectReplayEventLock(mock sqlmock.Sqlmock, eventID, status string, replayedAt any) {
	mock.ExpectQuery(`(?s)SELECT Status, ReplayedAt.*FROM _System_ConnectorEvent.*WHERE Id = \?.*FOR UPDATE`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Status", "ReplayedAt"}).AddRow(status, replayedAt))
}

func expectReplayPackAndEventLocks(mock sqlmock.Sqlmock, namespace, installID, eventID, status string, replayedAt any) {
	expectActivePackLock(mock, namespace, installID)
	expectReplayCutoverReceiptMissing(mock)
	expectReplayEventLock(mock, eventID, status, replayedAt)
}

// eventRowWithToken is like eventRow but populates the ClaimToken column.
func eventRowWithToken(id, ns, packID, kind, ikey, status, claimToken string, retry int) *sqlmock.Rows {
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	return sqlmock.NewRows([]string{
		"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
		"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
		"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
		"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey",
		"DeliveryId", "DeliverySequence", "CreatedDate", "LastModifiedDate",
	}).AddRow(
		id, ns, packID, kind, ikey, status,
		"default", []byte(`{}`), retry, nil, nil,
		"worker-1", nil, claimToken, nil, nil, nil,
		int64(1),      // default to version=1
		nil, nil, nil, // CompanionPackInstallId, CompanionFlowVersionId, SourceLaneKey (same-pack/legacy)
		nil, 0, now, now, // DeliveryId (nil=NULL legacy), DeliverySequence
	)
}

func nullStrPtr(s string) interface{} {
	if s == "" {
		return nil
	}
	return s
}

// fakePerm is a simple PermissionChecker used in tests.
type fakePerm struct{ keys map[string]bool }

func (f fakePerm) Has(k string) bool { return f.keys[k] }

func allowPerm(keys ...string) fakePerm {
	m := map[string]bool{}
	for _, k := range keys {
		m[k] = true
	}
	return fakePerm{keys: m}
}

func denyPerm() fakePerm { return fakePerm{keys: map[string]bool{}} }

func TestService_EnqueueBatchMissingIngressSchemaRollsBackBeforeMaterialization(t *testing.T) {
	svc, mock := newMock(t, "event-1")
	svc.admissionGate = connector_admission.NewSQLGate(svc.db)
	const namespace = "hubspot-crm-connector"

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectExec(`INSERT INTO _System_ConnectorIngressControl`).
		WithArgs(connector_admission.NamespaceFenceID(namespace), namespace, connector_admission.ModeActive).
		WillReturnError(errors.New("Error 1146 (42S02): Table 'nexuscrm._System_ConnectorIngressControl' doesn't exist"))
	mock.ExpectRollback()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))

	_, _, err := svc.EnqueueBatch(context.Background(), namespace, "pack-1", "delivery-1", []BatchEvent{{
		EventKind: "contact.created", IdempotencyKey: "source-1", Payload: json.RawMessage(`{}`), RouteVersion: 1,
	}})
	if !errors.Is(err, connector_admission.ErrIngressSchemaUnavailable) {
		t.Fatalf("EnqueueBatch error = %v, want typed ingress-schema unavailable", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("materialization ran after missing ingress schema: %v", err)
	}
}

// newMock returns a Service backed by sqlmock.
func newMock(t *testing.T, idgens ...string) (*Service, sqlmock.Sqlmock) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	t.Cleanup(func() { db.Close() })
	idgen := deterministicID(idgens...)
	now := fixedNow(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC))
	svc := NewService(db, idgen, now)
	// Tests in this package may substitute a deterministic in-memory authority;
	// production has no public admission-gate replacement hook.
	svc.admissionGate = testTransactionGate{}
	return svc, mock
}

type testTransactionGate struct{}

func (testTransactionGate) RequireActiveTx(_ context.Context, _ *sql.Tx, namespace string) (connector_admission.Fence, error) {
	return connector_admission.Fence{
		Namespace:  strings.ToLower(strings.TrimSpace(namespace)),
		Mode:       connector_admission.ModeActive,
		Capability: connector_admission.CapabilityAvailable,
	}, nil
}

func expectActivePackLock(mock sqlmock.Sqlmock, namespace, installID string) {
	mock.ExpectQuery(`SELECT 1\s+FROM _System_InstalledPack`).
		WithArgs(namespace, installID).
		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
}

func expectActivePackLockMissing(mock sqlmock.Sqlmock, namespace, installID string) {
	mock.ExpectQuery(`SELECT 1\s+FROM _System_InstalledPack`).
		WithArgs(namespace, installID).
		WillReturnError(sql.ErrNoRows)
}

func expectActiveCompanionPackLock(mock sqlmock.Sqlmock, installID string) {
	mock.ExpectQuery(`SELECT 1\s+FROM _System_InstalledPack`).
		WithArgs(installID, string(packstate.AppPack)).
		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
}

func expectActiveCompanionPackLockMissing(mock sqlmock.Sqlmock, installID string) {
	mock.ExpectQuery(`SELECT 1\s+FROM _System_InstalledPack`).
		WithArgs(installID, string(packstate.AppPack)).
		WillReturnError(sql.ErrNoRows)
}

func expectPessimisticBegin(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectPessimisticCleanup(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectLegacyTerminalEventNoMirror(mock sqlmock.Sqlmock, eventID, status, claimToken string) {
	expectPessimisticBegin(mock)
	mock.ExpectQuery(`SELECT Status, ClaimToken`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Status", "ClaimToken"}).AddRow(status, claimToken))
}

func canonicalEventKind(kind string) string {
	return strings.ToLower(strings.TrimSpace(kind))
}

func expectNextEnqueueOrdinal(mock sqlmock.Sqlmock, installID string, next int64) {
	mock.ExpectQuery(`SELECT ProtocolVersion, LegacyMaxOrdinal`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"ProtocolVersion", "LegacyMaxOrdinal"}).AddRow(1, 0))
	mock.ExpectQuery(`SELECT COALESCE\(MAX\(EnqueueOrdinal\), 0\) \+ 1`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"next"}).AddRow(next))
}

func expectActiveRouteQueueClassLock(mock sqlmock.Sqlmock, installID, eventKind string, routeVersion int64, queueClass string) {
	expectActiveRouteQueueClassLockWithMode(mock, installID, eventKind, routeVersion, queueClass, connector_inbound_routing.InvocationModePerEvent)
}

func expectActiveRouteQueueClassLockWithMode(mock sqlmock.Sqlmock, installID, eventKind string, routeVersion int64, queueClass, invocationMode string) {
	mock.ExpectQuery(`(?s)SELECT COALESCE\(NULLIF\(QueueClass, ''\), 'default'\), COALESCE\(InvocationMode, ''\).*FROM _System_ConnectorInboundRoute.*LIMIT 1\s+FOR UPDATE`).
		WithArgs(installID, canonicalEventKind(eventKind), routeVersion).
		WillReturnRows(sqlmock.NewRows([]string{"QueueClass", "InvocationMode"}).AddRow(queueClass, invocationMode))
}

func expectActiveRouteQueueClassLockMissing(mock sqlmock.Sqlmock, installID, eventKind string, routeVersion int64) {
	mock.ExpectQuery(`(?s)SELECT COALESCE\(NULLIF\(QueueClass, ''\), 'default'\), COALESCE\(InvocationMode, ''\).*FROM _System_ConnectorInboundRoute.*LIMIT 1\s+FOR UPDATE`).
		WithArgs(installID, canonicalEventKind(eventKind), routeVersion).
		WillReturnError(sql.ErrNoRows)
}

func expectRouteBackedEnqueueInsert(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, idempotencyKey string, routeVersion int64, rowsAffected int64) {
	expectRouteBackedEnqueueInsertOnly(mock, eventID, namespace, installID, eventKind, idempotencyKey, routeVersion, rowsAffected)
	if rowsAffected > 0 {
		expectConnectorWorkItemMaterialization(mock, eventID, installID, "default", statusPending, "", "")
	}
}

func expectRouteBackedEnqueueInsertOnly(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, idempotencyKey string, routeVersion int64, rowsAffected int64) {
	canonicalKind := canonicalEventKind(eventKind)
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute`).
		WithArgs(
			eventID,
			namespace,
			installID,
			canonicalKind,
			idempotencyKey,
			statusPending,
			sqlmock.AnyArg(), // payload
			connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
			installID,
			canonicalKind,
			routeVersion,
			sqlmock.AnyArg(), // source-settle companionPackInstallId
			sqlmock.AnyArg(), // source-settle companionFlowVersionId
			sqlmock.AnyArg(), // source-settle SourceObjectKind
			sqlmock.AnyArg(), // source-settle SourceObjectID
			sqlmock.AnyArg(), // source-settle current time
			sqlmock.AnyArg(), // source-settle default deadline
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			routeVersion,
			sqlmock.AnyArg(), // companionPackInstallId (nil for same-pack)
			sqlmock.AnyArg(), // companionFlowVersionId (nil for same-pack)
			sqlmock.AnyArg(), // deliveryId (nil for legacy/empty)
			sqlmock.AnyArg(), // deliverySequence
			sqlmock.AnyArg(), // enqueueOrdinal
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
			installID,
			canonicalKind,
			routeVersion,
		).
		WillReturnResult(sqlmock.NewResult(0, rowsAffected))
}

func expectRouteZeroEnqueueInsert(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, idempotencyKey string, rowsAffected int64) {
	expectRouteZeroEnqueueInsertOnly(mock, eventID, namespace, installID, eventKind, idempotencyKey, rowsAffected)
	if rowsAffected > 0 {
		expectConnectorWorkItemMaterialization(mock, eventID, installID, "default", statusPending, "", "")
	}
}

func expectRouteZeroEnqueueInsertOnly(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, idempotencyKey string, rowsAffected int64) {
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*VALUES`).
		WithArgs(
			eventID,
			namespace,
			installID,
			canonicalEventKind(eventKind),
			idempotencyKey,
			statusPending,
			"default",
			sqlmock.AnyArg(), // payload
			sqlmock.AnyArg(), // NextAttemptAt
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			int64(0),
			sqlmock.AnyArg(), // companionPackInstallId (nil for same-pack)
			sqlmock.AnyArg(), // companionFlowVersionId (nil for same-pack)
			sqlmock.AnyArg(), // deliveryId (nil for legacy/empty)
			sqlmock.AnyArg(), // deliverySequence
			sqlmock.AnyArg(), // enqueueOrdinal
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
		).
		WillReturnResult(sqlmock.NewResult(0, rowsAffected))
}

func expectConnectorWorkItemMaterialization(mock sqlmock.Sqlmock, eventID, installID, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	expectConnectorWorkItemMaterializationWithKind(mock, eventID, installID, "webhook.event", queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey)
}

func expectConnectorWorkItemMaterializationWithKind(mock sqlmock.Sqlmock, eventID, installID, eventKind, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	expectConnectorWorkItemMaterializationWithKindAt(mock, eventID, installID, eventKind, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey, nil, sqlmock.AnyArg())
}

func expectConnectorWorkItemMaterializationWithKindAt(mock sqlmock.Sqlmock, eventID, installID, eventKind, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string, nextAttemptAt, runAfter interface{}) {
	workType := workTypeForEventKind(eventKind)
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow(eventID, installID, eventKind, queueClass, eventStatus, sourceLaneKey, int64(1), nextAttemptAt))
	workState := connector_workqueue.WorkStatePending
	if eventStatus == statusFailed {
		workState = connector_workqueue.WorkStateRetry
	}
	reconcileLane := strings.TrimSpace(sourceLaneKey) != "" || strings.TrimSpace(targetMutationLaneKey) != ""
	insertState := workState
	if reconcileLane {
		insertState = connector_workqueue.WorkStateBlocked
		if workState == connector_workqueue.WorkStateRetry {
			insertState = connector_workqueue.WorkStateBlockedRetry
		}
	}
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			connectorEventWorkIDForWorkType(workType, eventID),
			eventID,
			installID,
			workType,
			queueClass,
			insertState,
			runAfter,
			sourceLaneKey,
			targetMutationLaneKey,
			int64(1),
			int64(0),
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	if reconcileLane {
		laneKey := sourceLaneKey
		if strings.TrimSpace(targetMutationLaneKey) != "" {
			laneKey = targetMutationLaneKey
		}
		workID := connectorEventWorkIDForWorkType(workType, eventID)
		mock.ExpectQuery(laneReadyHeadWorkItemQueryPattern()).
			WithArgs(installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateRetry,
				connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateRetry,
				connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(),
			).
			WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow(workID))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
			WithArgs(
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), workID,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
			WithArgs(
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), workID,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
			WithArgs(
				connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateBlockedRetry, connector_workqueue.WorkStateRetry,
				sqlmock.AnyArg(), workID,
				connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStateBlockedRetry,
				installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
				installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
}

func expectSourceProjectionWorkItemMaterializationAt(mock sqlmock.Sqlmock, eventID, installID, eventKind string, routeVersion int64, queueClass, sourceLaneKey string, nextAttemptAt, runAfter interface{}) {
	workType := workTypeForEventKind(eventKind)
	workID := sourceProjectionWorkID(installID, eventKind, routeVersion, "", "", workType, sourceLaneKey)
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow(eventID, installID, eventKind, queueClass, statusPending, sourceLaneKey, int64(1), nextAttemptAt))
	mock.ExpectQuery(`(?s)SELECT COALESCE\(e.RouteVersion, 0\).*COALESCE\(r.InvocationMode, ''\)`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "InvocationMode"}).
			AddRow(routeVersion, "", "", connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled))
	expectNoActiveSourceProjectionWorkItem(mock, workID)
	expectUncoveredSourceProjectionEvent(mock, workID, eventID, installID, sourceLaneKey, 1)
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			workID, eventID, installID, workType, queueClass, connector_workqueue.WorkStateBlocked, runAfter,
			sourceLaneKey, "", int64(1), int64(0), workID, workID, sqlmock.AnyArg(), sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectWorkItemLaneReconcile(mock, installID, sourceLaneKey, workID)
}

func expectUncoveredSourceProjectionEvent(mock sqlmock.Sqlmock, projectionKey, eventID, installID, sourceLaneKey string, ordinal int64) {
	mock.ExpectQuery(`(?s)SELECT EventId, LaneOrdinal.*ProjectionKey = \?.*FOR UPDATE`).
		WithArgs(projectionKey, ordinal, ordinal, eventID).
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "LaneOrdinal"}))
	mock.ExpectQuery(`(?s)SELECT 1.*projection.LaneOrdinal = 0`).
		WithArgs(eventID, projectionKey, installID, sourceLaneKey).
		WillReturnRows(sqlmock.NewRows([]string{"1"}))
}

func expectNoActiveSourceProjectionWorkItem(mock sqlmock.Sqlmock, projectionKey string) {
	activeColumns := []string{
		"WorkId", "EventId", "InstalledPackId", "WorkType",
		"SourceLaneKey", "State", "ProjectionKey", "ProjectionActiveKey",
	}
	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE\(SourceLaneKey, ''\), State,.*WHERE ProjectionActiveKey = \?.*LIMIT 1`).
		WithArgs(projectionKey).
		WillReturnRows(sqlmock.NewRows(activeColumns))
	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE\(SourceLaneKey, ''\), State,.*WHERE WorkId = \?.*FOR UPDATE`).
		WithArgs(projectionKey).
		WillReturnRows(sqlmock.NewRows(activeColumns))
	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE\(SourceLaneKey, ''\), State,.*WHERE ProjectionActiveKey = \?.*LIMIT 1`).
		WithArgs(projectionKey).
		WillReturnRows(sqlmock.NewRows(activeColumns))
}

func expectNoExistingWorkItemMirror(mock sqlmock.Sqlmock, eventID string) {
	mock.ExpectQuery(`SELECT WorkType\s+FROM _System_ConnectorWorkItem`).
		WithArgs(eventID, connector_workqueue.WorkTypeConnectorEvent).
		WillReturnError(sql.ErrNoRows)
}

func expectConnectorWorkItemInsert(mock sqlmock.Sqlmock, eventID, installID, workType, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	workState := connector_workqueue.WorkStatePending
	if eventStatus == statusFailed {
		workState = connector_workqueue.WorkStateRetry
	}
	reconcileLane := strings.TrimSpace(sourceLaneKey) != "" || strings.TrimSpace(targetMutationLaneKey) != ""
	insertState := workState
	if reconcileLane {
		insertState = connector_workqueue.WorkStateBlocked
		if workState == connector_workqueue.WorkStateRetry {
			insertState = connector_workqueue.WorkStateBlockedRetry
		}
	}
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			connectorEventWorkIDForWorkType(workType, eventID),
			eventID,
			installID,
			workType,
			queueClass,
			insertState,
			sqlmock.AnyArg(), // runAfter
			sourceLaneKey,
			targetMutationLaneKey,
			int64(1),
			int64(0),
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	if reconcileLane {
		laneKey := sourceLaneKey
		if strings.TrimSpace(targetMutationLaneKey) != "" {
			laneKey = targetMutationLaneKey
		}
		workID := connectorEventWorkIDForWorkType(workType, eventID)
		mock.ExpectQuery(laneReadyHeadWorkItemQueryPattern()).
			WithArgs(installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateRetry,
				connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateRetry,
				connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(),
			).
			WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow(workID))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
			WithArgs(
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), workID,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
			WithArgs(
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
				connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
				sqlmock.AnyArg(), installID, int64(0), laneKey,
				connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), workID,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
			WithArgs(
				connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStatePending,
				connector_workqueue.WorkStateBlockedRetry, connector_workqueue.WorkStateRetry,
				sqlmock.AnyArg(), workID,
				connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStateBlockedRetry,
				installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
				installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
}

func expectConnectorWorkItemDuplicateMaterializationWithoutExistingMirror(mock sqlmock.Sqlmock, eventID, installID, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow(eventID, installID, "webhook.event", queueClass, eventStatus, sourceLaneKey, int64(1), nil))
	expectNoExistingWorkItemMirror(mock, eventID)
	expectConnectorWorkItemInsert(mock, eventID, installID, connector_workqueue.WorkTypeConnectorEvent, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey)
}

func expectConnectorWorkItemDuplicateMaterialization(mock sqlmock.Sqlmock, eventID, installID, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	expectConnectorWorkItemDuplicateMaterializationWithKindAndExistingType(mock, eventID, installID, "webhook.event", connector_workqueue.WorkTypeConnectorEvent, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey)
}

func expectConnectorWorkItemDuplicateMaterializationWithKindAndExistingType(mock sqlmock.Sqlmock, eventID, installID, eventKind, existingWorkType, queueClass, eventStatus, sourceLaneKey, targetMutationLaneKey string) {
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow(eventID, installID, eventKind, queueClass, eventStatus, sourceLaneKey, int64(1), nil))
	mock.ExpectQuery(`SELECT WorkType\s+FROM _System_ConnectorWorkItem`).
		WithArgs(eventID, connector_workqueue.WorkTypeConnectorEvent).
		WillReturnRows(sqlmock.NewRows([]string{"WorkType"}).AddRow(existingWorkType))
	insertState := connector_workqueue.WorkStatePending
	if strings.TrimSpace(sourceLaneKey) != "" || strings.TrimSpace(targetMutationLaneKey) != "" {
		insertState = connector_workqueue.WorkStateBlocked
	}
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			connectorEventWorkIDForWorkType(existingWorkType, eventID),
			eventID,
			installID,
			existingWorkType,
			queueClass,
			insertState,
			sqlmock.AnyArg(), // runAfter
			sourceLaneKey,
			targetMutationLaneKey,
			int64(1),
			int64(0),
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
		).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "Duplicate entry 'event-work' for key 'uniq_connector_work_event_type'"})
	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType, QueueClass.*ReplayOfWorkId.*ProjectionActiveKey.*WHERE WorkId = \?.*OR \(EventId = \? AND WorkType = \?\).*ORDER BY WorkId.*LIMIT 2.*FOR UPDATE`).
		WithArgs(connectorEventWorkIDForWorkType(existingWorkType, eventID), eventID, existingWorkType).
		WillReturnRows(sqlmock.NewRows([]string{
			"WorkId", "EventId", "InstalledPackId", "WorkType", "QueueClass",
			"SourceLaneKey", "TargetMutationLaneKey", "LaneOrdinal", "Generation",
			"ReplayOfWorkId", "ProjectionKey", "ProjectionActiveKey",
		}).AddRow(
			connectorEventWorkIDForWorkType(existingWorkType, eventID), eventID, installID, existingWorkType,
			queueClass, sourceLaneKey, targetMutationLaneKey, int64(1), int64(0), "", "", "",
		))
}

func laneReadyHeadWorkItemQueryPattern() string {
	return `(?s)SELECT w\.WorkId.*USE INDEX \(idx_connector_work_item_target_lane\).*State IN \(\?, \?, \?, \?\).*RunAfter <= \?.*UNION ALL.*USE INDEX \(idx_connector_work_item_source_lane_gen\).*LEFT JOIN _System_ConnectorEvent e.*ORDER BY w\.LaneOrdinal ASC,.*e\.EnqueueOrdinal.*e\.DeliverySequence.*w\.CreatedDate ASC.*w\.WorkId ASC.*LIMIT 1`
}

func TestLegacyEventWorkItemClaimGuardUsesDurableRows(t *testing.T) {
	svc := &Service{}
	guard, args := svc.legacyEventWorkItemClaimGuardSQL("e")
	if len(args) != 0 {
		t.Fatalf("durable claim guard binds = %v, want none", args)
	}
	for _, want := range []string{
		"e.ExecutionOwner = 'event'",
		"NULLIF(COALESCE(projection.ProjectionActiveKey, ''), '') IS NOT NULL",
		"projection.WorkId LIKE 'csp:%'",
		"COALESCE(projection.SourceLaneKey, '') = COALESCE(e.SourceLaneKey, '')",
		"LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(e.EventKind))",
	} {
		if !strings.Contains(guard, want) {
			t.Fatalf("durable claim guard missing %q:\n%s", want, guard)
		}
	}
	if strings.Contains(guard, "InstalledPackId NOT IN") {
		t.Fatalf("durable claim guard still depends on process-local pack list:\n%s", guard)
	}
	if strings.Contains(guard, "direct.EventId") {
		t.Fatalf("materialization alone must not fence the Event before WorkItem ownership transfers:\n%s", guard)
	}
	if strings.Contains(guard, "projection.WorkType") {
		t.Fatalf("durable claim guard must fail closed on mismatched WorkItem type:\n%s", guard)
	}
}

func expectRouteBackedReplayInsert(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, replayOfID string, routeVersion int64, rowsAffected int64) {
	expectRouteBackedReplayEventInsert(mock, eventID, namespace, installID, eventKind, replayOfID, routeVersion, rowsAffected)
	if rowsAffected > 0 {
		expectConnectorWorkItemMaterialization(mock, eventID, installID, "default", statusPending, "", "")
	}
}

func expectRouteBackedReplayEventInsert(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, replayOfID string, routeVersion int64, rowsAffected int64) {
	canonicalKind := canonicalEventKind(eventKind)
	mock.ExpectQuery(`SELECT ProtocolVersion, LegacyMaxOrdinal`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"ProtocolVersion", "LegacyMaxOrdinal"}).AddRow(1, 0))
	mock.ExpectQuery(`SELECT COALESCE\(MAX\(EnqueueOrdinal\), 0\) \+ 1`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"next"}).AddRow(int64(1)))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute`).
		WithArgs(
			eventID,
			namespace,
			installID,
			canonicalKind,
			"replay:"+eventID,
			statusPending,
			sqlmock.AnyArg(), // payload
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			routeVersion,
			sqlmock.AnyArg(), // companionPackInstallId (nil for same-pack replays)
			sqlmock.AnyArg(), // companionFlowVersionId (nil for same-pack replays)
			replayOfID,
			sqlmock.AnyArg(), // deliveryId (copied from dead row; nil for legacy)
			sqlmock.AnyArg(), // deliverySequence (copied from dead row)
			sqlmock.AnyArg(), // enqueueOrdinal
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
			installID,
			canonicalKind,
			routeVersion,
		).
		WillReturnResult(sqlmock.NewResult(0, rowsAffected))
}

func expectConnectorWorkItemReplayMaterialization(mock sqlmock.Sqlmock, eventID, installID, replayOfWorkID string, insertErr error) {
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs(eventID).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow(eventID, installID, "webhook.event", "default", statusPending, "", int64(1), nil))
	expectation := mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkItem.*ReplayOfWorkId.*VALUES`).
		WithArgs(
			connectorEventWorkIDForWorkType(connector_workqueue.WorkTypeConnectorEvent, eventID),
			eventID,
			installID,
			connector_workqueue.WorkTypeConnectorEvent,
			"default",
			connector_workqueue.WorkStatePending,
			sqlmock.AnyArg(),
			"",
			"",
			int64(1),
			int64(0),
			replayOfWorkID,
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		)
	if insertErr != nil {
		expectation.WillReturnError(insertErr)
		return
	}
	expectation.WillReturnResult(sqlmock.NewResult(0, 1))
}

func expectRouteBackedReplayInsertError(mock sqlmock.Sqlmock, eventID, namespace, installID, eventKind, replayOfID string, routeVersion int64, err error) {
	canonicalKind := canonicalEventKind(eventKind)
	mock.ExpectQuery(`SELECT ProtocolVersion, LegacyMaxOrdinal`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"ProtocolVersion", "LegacyMaxOrdinal"}).AddRow(1, 0))
	mock.ExpectQuery(`SELECT COALESCE\(MAX\(EnqueueOrdinal\), 0\) \+ 1`).
		WithArgs(installID).
		WillReturnRows(sqlmock.NewRows([]string{"next"}).AddRow(int64(1)))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute`).
		WithArgs(
			eventID,
			namespace,
			installID,
			canonicalKind,
			"replay:"+eventID,
			statusPending,
			sqlmock.AnyArg(), // payload
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			routeVersion,
			sqlmock.AnyArg(), // companionPackInstallId
			sqlmock.AnyArg(), // companionFlowVersionId
			replayOfID,
			sqlmock.AnyArg(), // deliveryId (copied from dead row; nil for legacy)
			sqlmock.AnyArg(), // deliverySequence (copied from dead row)
			sqlmock.AnyArg(), // enqueueOrdinal
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
			installID,
			canonicalKind,
			routeVersion,
		).
		WillReturnError(err)
}

func TestEnqueue_DuplicateMaterializationUsesWorkItemIdempotency(t *testing.T) {
	svc, mock := newMock(t)

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-dup-work").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("existing-event-id"))
	expectConnectorWorkItemDuplicateMaterialization(mock, "existing-event-id", "install-token-test", "default", statusPending, "", "")
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "ikey-dup-work", nil, "install-token-test", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !isDup {
		t.Fatal("expected isDuplicate=true")
	}
	if id != "existing-event-id" {
		t.Fatalf("want existing-event-id, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueue_DuplicateWorkflowActionReusesExistingLegacyWorkItemMirror(t *testing.T) {
	svc, mock := newMock(t)

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-dup-workflow-action").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("existing-workflow-action"))
	expectConnectorWorkItemDuplicateMaterializationWithKindAndExistingType(
		mock,
		"existing-workflow-action",
		"install-token-test",
		"workflow.action.execute",
		connector_workqueue.WorkTypeConnectorEvent,
		"default",
		statusPending,
		"",
		"",
	)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "workflow.action.execute", "ikey-dup-workflow-action", nil, "install-token-test", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !isDup {
		t.Fatal("expected isDuplicate=true")
	}
	if id != "existing-workflow-action" {
		t.Fatalf("want existing-workflow-action, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_WorkItemFailureRollsBackEventFact(t *testing.T) {
	svc, mock := newMock(t, "event-rollback")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-rollback").
		WillReturnError(sql.ErrNoRows)
	expectRouteBackedEnqueueInsertOnly(mock, "event-rollback", "noop-connector", "install-token-test", "webhook.event", "ikey-rollback", 1, 1)
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs("event-rollback").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow("event-rollback", "install-token-test", "webhook.event", "default", statusPending, "", int64(1), nil))
	workErr := errors.New("work item insert failed")
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(workErr)
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.EnqueueBatch(context.Background(), "noop-connector", "install-token-test", "dlv-rollback", []BatchEvent{{
		EventKind:        "webhook.event",
		IdempotencyKey:   "ikey-rollback",
		Payload:          json.RawMessage(`{}`),
		RouteVersion:     1,
		DeliverySequence: 0,
	}})
	if !errors.Is(err, workErr) {
		t.Fatalf("EnqueueBatch error = %v, want %v", err, workErr)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_WorkItemMissingSchemaRollsBack(t *testing.T) {
	svc, mock := newMock(t, "event-no-workitem-schema")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-no-workitem-schema").
		WillReturnError(sql.ErrNoRows)
	expectRouteBackedEnqueueInsertOnly(mock, "event-no-workitem-schema", "noop-connector", "install-token-test", "webhook.event", "ikey-no-workitem-schema", 1, 1)
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs("event-no-workitem-schema").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow("event-no-workitem-schema", "install-token-test", "webhook.event", "default", statusPending, "", int64(1), nil))
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(errors.New("Error 1146: Table 'nexuscrm._System_ConnectorWorkItem' doesn't exist"))
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "noop-connector", "install-token-test", "dlv-no-workitem-schema", []BatchEvent{{
		EventKind:        "webhook.event",
		IdempotencyKey:   "ikey-no-workitem-schema",
		Payload:          json.RawMessage(`{}`),
		RouteVersion:     1,
		DeliverySequence: 0,
	}})
	if err == nil {
		t.Fatalf("EnqueueBatch succeeded with ids=%v dups=%v, want WorkItem schema error", ids, dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_WorkItemMissingLaneOrdinalRollsBack(t *testing.T) {
	svc, mock := newMock(t, "event-no-workitem-laneordinal")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-no-workitem-laneordinal").
		WillReturnError(sql.ErrNoRows)
	expectRouteBackedEnqueueInsertOnly(mock, "event-no-workitem-laneordinal", "noop-connector", "install-token-test", "webhook.event", "ikey-no-workitem-laneordinal", 1, 1)
	mock.ExpectQuery(`SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE\(SourceLaneKey, ''\), EnqueueOrdinal, NextAttemptAt`).
		WithArgs("event-no-workitem-laneordinal").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackId", "EventKind", "QueueClass", "Status", "SourceLaneKey", "EnqueueOrdinal", "NextAttemptAt"}).
			AddRow("event-no-workitem-laneordinal", "install-token-test", "webhook.event", "default", statusPending, "", int64(1), nil))
	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WillReturnError(errors.New("Error 1054: Unknown column 'LaneOrdinal' in 'field list'"))
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "noop-connector", "install-token-test", "dlv-no-workitem-laneordinal", []BatchEvent{{
		EventKind:        "webhook.event",
		IdempotencyKey:   "ikey-no-workitem-laneordinal",
		Payload:          json.RawMessage(`{}`),
		RouteVersion:     1,
		DeliverySequence: 0,
	}})
	if err == nil {
		t.Fatalf("EnqueueBatch succeeded with ids=%v dups=%v, want WorkItem schema error", ids, dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestMissingConnectorWorkItemSchemaRecognizesProjectionGenerationColumns(t *testing.T) {
	for _, column := range []string{"ProjectionKey", "ProjectionActiveKey", "TerminalAt"} {
		err := errors.New("Error 1054 (42S22): Unknown column '" + column + "' in 'field list'")
		if !isMissingConnectorWorkItemSchema(err) {
			t.Fatalf("missing %s was not recognized as an additive WorkItem schema gap", column)
		}
	}
	if isMissingConnectorWorkItemSchema(errors.New("driver: bad connection")) {
		t.Fatal("unrelated connector error was classified as a WorkItem schema gap")
	}
}

func TestEnqueueBatch_WorkItemCarriesTargetMutationLaneWhenPresent(t *testing.T) {
	svc, mock := newMock(t, "event-target-lane")
	targetLaneKey := sourceLaneKeyFor("campaign", "campaign-001")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-target-lane").
		WillReturnError(sql.ErrNoRows)
	expectRouteBackedEnqueueInsertOnly(mock, "event-target-lane", "noop-connector", "install-token-test", "workflow.action.execute", "ikey-target-lane", 1, 1)
	expectConnectorWorkItemMaterializationWithKind(mock, "event-target-lane", "install-token-test", "workflow.action.execute", "default", statusPending, "", targetLaneKey)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "noop-connector", "install-token-test", "dlv-target", []BatchEvent{{
		EventKind:                    "workflow.action.execute",
		IdempotencyKey:               "ikey-target-lane",
		Payload:                      json.RawMessage(`{}`),
		RouteVersion:                 1,
		TargetMutationLaneObjectKind: "campaign",
		TargetMutationLaneID:         "campaign-001",
		DeliverySequence:             0,
	}})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if len(ids) != 1 || ids[0] != "event-target-lane" {
		t.Fatalf("ids = %v, want [event-target-lane]", ids)
	}
	if len(dups) != 1 || dups[0] {
		t.Fatalf("dups = %v, want [false]", dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_RouteBackedHubSpotInboundMaterializesConnectorEventWorkItem(t *testing.T) {
	svc, mock := newMock(t, "event-hubspot-contact")
	sourceLaneKey := sourceLaneKeyFor("contact", "hubspot-contact-001")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", "install-token-test")
	expectNextEnqueueOrdinal(mock, "install-token-test", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-test", "ikey-hubspot-contact").
		WillReturnError(sql.ErrNoRows)
	expectRouteBackedEnqueueInsertOnly(mock, "event-hubspot-contact", "hubspot-connector", "install-token-test", "contact.propertychange", "ikey-hubspot-contact", 1, 1)
	expectConnectorWorkItemMaterializationWithKind(mock, "event-hubspot-contact", "install-token-test", "contact.propertychange", "default", statusPending, sourceLaneKey, "")
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", "install-token-test", "dlv-hubspot-contact", []BatchEvent{{
		EventKind:        "contact.propertychange",
		IdempotencyKey:   "ikey-hubspot-contact",
		Payload:          json.RawMessage(`{"objectKind":"contact","sourceID":"hubspot-contact-001","payload":{"propertyName":"lifecyclestage"}}`),
		RouteVersion:     1,
		DeliverySequence: 0,
	}})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if len(ids) != 1 || ids[0] != "event-hubspot-contact" {
		t.Fatalf("ids = %v, want [event-hubspot-contact]", ids)
	}
	if len(dups) != 1 || dups[0] {
		t.Fatalf("dups = %v, want [false]", dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_FreshRouteFastPathReconcilesWorkItemLaneHeads(t *testing.T) {
	svc, mock := newMock(t, "event-batch-1", "event-batch-2")
	const installID = "install-token-test"
	laneKey := sourceLaneKeyFor("contact", "shared-contact")
	headWorkID := connectorEventWorkIDForWorkType(connector_workqueue.WorkTypeConnectorEvent, "event-batch-1")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", installID)
	expectNextEnqueueOrdinal(mock, installID, 10)
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(installID, sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*VALUES`).
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkItem.*VALUES`).
		WithArgs(
			headWorkID,
			"event-batch-1",
			installID,
			connector_workqueue.WorkTypeConnectorEvent,
			"default",
			connector_workqueue.WorkStatePending,
			sqlmock.AnyArg(),
			laneKey,
			"",
			int64(10),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			connectorEventWorkIDForWorkType(connector_workqueue.WorkTypeConnectorEvent, "event-batch-2"),
			"event-batch-2",
			installID,
			connector_workqueue.WorkTypeConnectorEvent,
			"default",
			connector_workqueue.WorkStateBlocked,
			sqlmock.AnyArg(),
			laneKey,
			"",
			int64(11),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 2))
	expectWorkItemLaneReconcile(mock, installID, laneKey, headWorkID)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", installID, "dlv-batch", []BatchEvent{
		{
			EventKind:        "contact.propertychange",
			IdempotencyKey:   "ikey-batch-1",
			Payload:          json.RawMessage(`{"objectKind":"contact","sourceID":"shared-contact"}`),
			RouteVersion:     0,
			DeliverySequence: 0,
		},
		{
			EventKind:        "contact.propertychange",
			IdempotencyKey:   "ikey-batch-2",
			Payload:          json.RawMessage(`{"objectKind":"contact","sourceID":"shared-contact"}`),
			RouteVersion:     0,
			DeliverySequence: 1,
		},
	})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if got, want := strings.Join(ids, ","), "event-batch-1,event-batch-2"; got != want {
		t.Fatalf("ids = %v, want %s", ids, want)
	}
	if len(dups) != 2 || dups[0] || dups[1] {
		t.Fatalf("dups = %v, want [false false]", dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestSourceProjectionWorkIDIsStablePerSourceAndIsolatesRouteIdentity(t *testing.T) {
	const installID = "install-settled"
	laneKey := sourceLaneKeyFor("contact", "shared-contact")
	first := sourceProjectionWorkID(installID, "contact.propertychange", 1, "", "", connector_workqueue.WorkTypeConnectorEvent, laneKey)
	second := sourceProjectionWorkID(installID, "contact.propertychange", 1, "", "", connector_workqueue.WorkTypeConnectorEvent, laneKey)
	if first != second || !connector_workqueue.IsSourceProjectionWorkID(first) {
		t.Fatalf("projection work id is not stable: first=%q second=%q", first, second)
	}
	otherRoute := sourceProjectionWorkID(installID, "contact.propertychange", 2, "", "", connector_workqueue.WorkTypeConnectorEvent, laneKey)
	otherSource := sourceProjectionWorkID(installID, "contact.propertychange", 1, "", "", connector_workqueue.WorkTypeConnectorEvent, sourceLaneKeyFor("contact", "other-contact"))
	if first == otherRoute || first == otherSource {
		t.Fatalf("projection identity conflated route/source: first=%q route=%q source=%q", first, otherRoute, otherSource)
	}
}

func TestPartitionFreshRouteBatchWorkItemsSC400KeepsCommandsAndProjectionHeads(t *testing.T) {
	rows := makeSC400PartitionRows()
	ordinary, projectionHeads := partitionFreshRouteBatchWorkItems(rows)
	if len(ordinary) != 400 || len(projectionHeads) != 400 {
		t.Fatalf("partition ordinary=%d projection_heads=%d, want 400/400", len(ordinary), len(projectionHeads))
	}
	for _, row := range ordinary {
		if row.sourceProjection || row.workType != connector_workqueue.WorkTypeWorkflowActionApply {
			t.Fatalf("ordinary command was altered: %+v", row)
		}
	}
	for _, row := range projectionHeads {
		if !row.sourceProjection || row.enqueueOrdinal%10 != 9 {
			t.Fatalf("projection head is not the newest source event: %+v", row)
		}
	}
}

func BenchmarkPartitionFreshRouteBatchWorkItemsSC400(b *testing.B) {
	rows := makeSC400PartitionRows()
	b.ReportAllocs()
	for i := 0; i < b.N; i++ {
		ordinary, projectionHeads := partitionFreshRouteBatchWorkItems(rows)
		if len(ordinary) != 400 || len(projectionHeads) != 400 {
			b.Fatalf("partition ordinary=%d projection_heads=%d", len(ordinary), len(projectionHeads))
		}
	}
}

func makeSC400PartitionRows() []freshRouteBatchEvent {
	rows := make([]freshRouteBatchEvent, 0, 4400)
	for source := 0; source < 400; source++ {
		workID := fmt.Sprintf("csp:%03d", source)
		for change := 0; change < 10; change++ {
			rows = append(rows, freshRouteBatchEvent{
				workID: workID, workType: connector_workqueue.WorkTypeConnectorEvent,
				sourceProjection: true, enqueueOrdinal: int64(source*10 + change),
			})
		}
	}
	for action := 0; action < 400; action++ {
		rows = append(rows, freshRouteBatchEvent{
			workID: fmt.Sprintf("cwi:%03d", action), workType: connector_workqueue.WorkTypeWorkflowActionApply,
			enqueueOrdinal: int64(4000 + action),
		})
	}
	return rows
}

func TestEnqueueBatch_FreshRouteFastPathBatchesCompanionEvents(t *testing.T) {
	svc, mock := newMock(t, "event-companion-1", "event-companion-2")
	const (
		providerInstallID  = "install-provider"
		companionInstallID = "install-companion"
	)

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", providerInstallID)
	expectNextEnqueueOrdinal(mock, providerInstallID, 20)
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(providerInstallID, sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}))
	expectActiveCompanionPackLock(mock, companionInstallID)
	expectActiveRouteQueueClassLock(mock, providerInstallID, "contact.propertychange", 1, "default")
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*VALUES`).
		WithArgs(
			"event-companion-1",
			"hubspot-connector",
			providerInstallID,
			"contact.propertychange",
			"ikey-companion-1",
			statusPending,
			"default",
			sqlmock.AnyArg(),
			nil,
			nil,
			nil,
			nil,
			int64(1),
			companionInstallID,
			"companion-flow-v1",
			"delivery-companion",
			0,
			int64(20),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			"event-companion-2",
			"hubspot-connector",
			providerInstallID,
			"contact.propertychange",
			"ikey-companion-2",
			statusPending,
			"default",
			sqlmock.AnyArg(),
			nil,
			nil,
			nil,
			nil,
			int64(1),
			companionInstallID,
			"companion-flow-v2",
			"delivery-companion",
			1,
			int64(21),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkItem.*VALUES`).
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	ids, dups, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", providerInstallID, "delivery-companion", []BatchEvent{
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-1",
			Payload:                json.RawMessage(`{}`),
			RouteVersion:           1,
			CompanionPackInstallID: companionInstallID,
			CompanionFlowVersionID: "companion-flow-v1",
			DeliverySequence:       0,
		},
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-2",
			Payload:                json.RawMessage(`{}`),
			RouteVersion:           1,
			CompanionPackInstallID: companionInstallID,
			CompanionFlowVersionID: "companion-flow-v2",
			DeliverySequence:       1,
		},
	})
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if got, want := strings.Join(ids, ","), "event-companion-1,event-companion-2"; got != want {
		t.Fatalf("ids = %v, want %s", ids, want)
	}
	if len(dups) != 2 || dups[0] || dups[1] {
		t.Fatalf("dups = %v, want [false false]", dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_FreshRouteFastPathChecksDuplicatesBeforeRouteLock(t *testing.T) {
	svc, mock := newMock(t, "unused-event-1", "unused-event-2")
	const providerInstallID = "install-provider"

	mock.ExpectBegin()
	tx, err := svc.db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatalf("BeginTx: %v", err)
	}
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(providerInstallID, sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}).AddRow("ikey-companion-1", "existing-event"))
	mock.ExpectRollback()

	ids, dups, fastPath, err := svc.enqueueFreshRouteBatchFastPath(
		context.Background(),
		tx,
		"hubspot-connector",
		providerInstallID,
		"delivery-companion",
		[]BatchEvent{
			{
				EventKind:              "contact.propertychange",
				IdempotencyKey:         "ikey-companion-1",
				Payload:                json.RawMessage(`{}`),
				RouteVersion:           7,
				CompanionPackInstallID: "install-companion",
				CompanionFlowVersionID: "flow-v7",
			},
			{
				EventKind:              "contact.propertychange",
				IdempotencyKey:         "ikey-companion-2",
				Payload:                json.RawMessage(`{}`),
				RouteVersion:           7,
				CompanionPackInstallID: "install-companion",
				CompanionFlowVersionID: "flow-v7",
			},
		},
		50,
		time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
	)
	_ = tx.Rollback()
	if err != nil {
		t.Fatalf("enqueueFreshRouteBatchFastPath: %v", err)
	}
	if fastPath || ids != nil || dups != nil {
		t.Fatalf("fastPath = %v, ids = %v, dups = %v; want slow-path fallback", fastPath, ids, dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_FreshRouteFastPathRejectsSweptCompanionRoute(t *testing.T) {
	svc, mock := newMock(t, "unused-event-1", "unused-event-2")
	const providerInstallID = "install-provider"

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", providerInstallID)
	expectNextEnqueueOrdinal(mock, providerInstallID, 25)
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(providerInstallID, sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}))
	expectActiveCompanionPackLock(mock, "install-companion")
	expectActiveRouteQueueClassLockMissing(mock, providerInstallID, "contact.propertychange", 7)
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", providerInstallID, "delivery-companion", []BatchEvent{
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-1",
			Payload:                json.RawMessage(`{}`),
			RouteVersion:           7,
			CompanionPackInstallID: "install-companion",
			CompanionFlowVersionID: "flow-v7",
		},
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-2",
			Payload:                json.RawMessage(`{}`),
			RouteVersion:           7,
			CompanionPackInstallID: "install-companion",
			CompanionFlowVersionID: "flow-v7",
		},
	})
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_FreshRouteFastPathLocksDistinctCompanionsOnceInStableOrder(t *testing.T) {
	svc, mock := newMock(t, "event-companion-z-1", "event-companion-a", "event-companion-z-2")
	const providerInstallID = "install-provider"

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", providerInstallID)
	expectNextEnqueueOrdinal(mock, providerInstallID, 30)
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(providerInstallID, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}))
	expectActiveCompanionPackLock(mock, "companion-a")
	expectActiveCompanionPackLock(mock, "companion-z")
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*VALUES`).
		WillReturnResult(sqlmock.NewResult(0, 3))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkItem.*VALUES`).
		WillReturnResult(sqlmock.NewResult(0, 3))
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	events := []BatchEvent{
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-z-1",
			Payload:                json.RawMessage(`{}`),
			CompanionPackInstallID: "companion-z",
			CompanionFlowVersionID: "flow-z-1",
		},
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-a",
			Payload:                json.RawMessage(`{}`),
			CompanionPackInstallID: "companion-a",
			CompanionFlowVersionID: "flow-a",
		},
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-z-2",
			Payload:                json.RawMessage(`{}`),
			CompanionPackInstallID: "companion-z",
			CompanionFlowVersionID: "flow-z-2",
		},
	}
	ids, dups, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", providerInstallID, "delivery-companions", events)
	if err != nil {
		t.Fatalf("EnqueueBatch: %v", err)
	}
	if len(ids) != 3 || len(dups) != 3 || dups[0] || dups[1] || dups[2] {
		t.Fatalf("ids = %v, dups = %v", ids, dups)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueueBatch_FreshRouteFastPathRejectsInactiveCompanion(t *testing.T) {
	svc, mock := newMock(t, "event-companion-1", "event-companion-2")
	const providerInstallID = "install-provider"

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "hubspot-connector", providerInstallID)
	expectNextEnqueueOrdinal(mock, providerInstallID, 40)
	mock.ExpectQuery(`(?s)SELECT IdempotencyKey, Id.*FROM _System_ConnectorEvent.*IdempotencyKey IN`).
		WithArgs(providerInstallID, sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"IdempotencyKey", "Id"}))
	expectActiveCompanionPackLockMissing(mock, "companion-stale")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.EnqueueBatch(context.Background(), "hubspot-connector", providerInstallID, "delivery-companion", []BatchEvent{
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-1",
			Payload:                json.RawMessage(`{}`),
			CompanionPackInstallID: "companion-stale",
			CompanionFlowVersionID: "flow-v1",
		},
		{
			EventKind:              "contact.propertychange",
			IdempotencyKey:         "ikey-companion-2",
			Payload:                json.RawMessage(`{}`),
			CompanionPackInstallID: "companion-stale",
			CompanionFlowVersionID: "flow-v1",
		},
	})
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func expectWorkItemLaneReconcile(mock sqlmock.Sqlmock, installID, laneKey, headWorkID string) {
	mock.ExpectQuery(laneReadyHeadWorkItemQueryPattern()).
		WithArgs(installID, int64(0), laneKey,
			connector_workqueue.WorkStatePending,
			connector_workqueue.WorkStateRetry,
			connector_workqueue.WorkStateBlocked,
			connector_workqueue.WorkStateBlockedRetry,
			sqlmock.AnyArg(), installID, int64(0), laneKey,
			connector_workqueue.WorkStatePending,
			connector_workqueue.WorkStateRetry,
			connector_workqueue.WorkStateBlocked,
			connector_workqueue.WorkStateBlockedRetry,
			sqlmock.AnyArg(),
		).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow(headWorkID))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
			connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
			sqlmock.AnyArg(), installID, int64(0), laneKey,
			connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), headWorkID,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			connector_workqueue.WorkStatePending, connector_workqueue.WorkStateBlocked,
			connector_workqueue.WorkStateRetry, connector_workqueue.WorkStateBlockedRetry,
			sqlmock.AnyArg(), installID, int64(0), laneKey,
			connector_workqueue.WorkStatePending, connector_workqueue.WorkStateRetry, sqlmock.AnyArg(), headWorkID,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(
			connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStatePending,
			connector_workqueue.WorkStateBlockedRetry, connector_workqueue.WorkStateRetry,
			sqlmock.AnyArg(), headWorkID,
			connector_workqueue.WorkStateBlocked, connector_workqueue.WorkStateBlockedRetry,
			installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
			installID, int64(0), laneKey, connector_workqueue.WorkStateProcessing,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 3 — Enqueue: empty packInstallID → error (fail-loud validation).
// Since Enqueue no longer calls resolvePack, the caller is responsible for
// passing a valid packInstallID. Empty string is an explicit caller error.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_EmptyPackInstallID(t *testing.T) {
	svc, mock := newMock(t)

	// No DB expectations — validation must fire before any DB call.
	_, _, err := svc.Enqueue(context.Background(), "gone-connector", "webhook.event", "key", nil, "", 0, "", "", 0, "")
	if err == nil {
		t.Fatal("expected error for empty packInstallID, got nil")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 3b — ResolveActivePack: returns InstallID from active pack row.
// ─────────────────────────────────────────────────────────────────────────────

func TestService_ResolveActivePack_Found(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Id, PackInstallId, Name, Category, Status`).
		WithArgs("noop-connector", "connector-pack", "active").
		WillReturnRows(packRow("pack-test-001", "install-token-abc", "noop-connector"))

	packInstallID, err := svc.ResolveActivePack(context.Background(), "noop-connector")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if packInstallID != "install-token-abc" {
		t.Fatalf("want install-token-abc, got %q", packInstallID)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestService_ResolveActivePack_NotFound_ReturnsErrInactivePack(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Id, PackInstallId, Name, Category, Status`).
		WithArgs("gone-connector", "connector-pack", "active").
		WillReturnRows(noPackRow())

	_, err := svc.ResolveActivePack(context.Background(), "gone-connector")
	if !errors.Is(err, ErrInactivePack) {
		t.Fatalf("want ErrInactivePack, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 4c — MarkProcessed: lease-bound UPDATE returns 0 rows → ErrLeaseLost.
// Covers both the TOCTOU window (another worker raced us) and the stale-reclaim
// scenario (ClaimNext reassigned the lease to a new worker).
// ─────────────────────────────────────────────────────────────────────────────

func TestMarkProcessed_TOCTOUReturnsLeaseLost(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-toctou").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("noop-connector", statusProcessing))
	expectPessimisticBegin(mock)
	mock.ExpectQuery(`SELECT Status, ClaimToken`).
		WithArgs("evt-toctou").
		WillReturnRows(sqlmock.NewRows([]string{"Status", "ClaimToken"}).
			AddRow(statusProcessed, nil))
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	err := svc.MarkProcessed(context.Background(), "evt-toctou", "token-abc")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("want ErrLeaseLost, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestReleaseClaimUsesPostClaimOwnerFenceWithoutProjectionRead(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent e.*e\.ExecutionOwner = 'event'\s*$`).
		WithArgs(statusPending, sqlmock.AnyArg(), "evt-partial-schema", statusProcessing, "legacy-token").
		WillReturnResult(sqlmock.NewResult(0, 1))

	err := svc.ReleaseClaim(context.Background(), "evt-partial-schema", "legacy-token")
	if err != nil {
		t.Fatalf("ReleaseClaim partial schema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("partial schema expectations: %v", err)
	}
}

func TestReleaseClaimDoesNotDependOnWorkItemTableAfterLeaseGrant(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent e.*WHERE e\.Id = \?.*e\.ClaimToken = \?.*e\.ExecutionOwner = 'event'\s*$`).
		WithArgs(statusPending, sqlmock.AnyArg(), "evt-old-schema", statusProcessing, "legacy-token").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := svc.ReleaseClaim(context.Background(), "evt-old-schema", "legacy-token"); err != nil {
		t.Fatalf("ReleaseClaim old schema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("old schema expectations: %v", err)
	}
}

func TestReleaseClaimFallsBackWhenEventExecutionOwnerColumnIsUnavailable(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent e.*e\.ExecutionOwner = 'event'`).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'e.ExecutionOwner' in 'where clause'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorEvent e.*WHERE e\.Id = \?.*e\.ClaimToken = \?$`).
		WithArgs(statusPending, sqlmock.AnyArg(), "evt-mixed-schema", statusProcessing, "legacy-token").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := svc.ReleaseClaim(context.Background(), "evt-mixed-schema", "legacy-token"); err != nil {
		t.Fatalf("ReleaseClaim mixed schema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("mixed-schema expectations: %v", err)
	}
}

// TestReleaseClaimForRetry_BacksOffWithoutBudgetBurn verifies the OUT-1
// follow-up spin protection: the retriable re-drive sets the row to 'failed'
// with NextAttemptAt = now + connectorRetriableReclaimDelay (so ClaimNext's
// failed-arm gate holds it off until the backoff elapses — no hot-loop) and
// does NOT increment RetryCount (no genuine-failure budget burn). The exact
// arg list pins both: NextAttemptAt is the backed-off timestamp and RetryCount
// is absent from the SET clause (contrast MarkFailed, which passes RetryCount).
func TestReleaseClaimForRetry_BacksOffWithoutBudgetBurn(t *testing.T) {
	svc, mock := newMock(t)

	baseNow := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	wantNextAttempt := baseNow.Add(connectorRetriableReclaimDelay)
	if !wantNextAttempt.After(baseNow) {
		t.Fatal("connectorRetriableReclaimDelay must be a positive backoff to prevent a re-claim hot-loop")
	}

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(
			statusFailed,                       // Status -> failed (claim predicate gates failed rows on NextAttemptAt)
			wantNextAttempt,                    // NextAttemptAt -> now + backoff (spin protection)
			"retry after the winner finalizes", // LastError -> operator-visible reason
			baseNow,                            // LastErrorAt
			baseNow,                            // LastModifiedDate
			"evt-retriable",                    // Id
			statusProcessing,                   // Status guard
			"token-live",                       // ClaimToken guard
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := svc.ReleaseClaimForRetry(context.Background(), "evt-retriable", "token-live", "retry after the winner finalizes"); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestReleaseClaimForRetry_LeaseLost verifies the claim-token gate: a release
// whose token no longer matches the leased row affects zero rows and surfaces
// ErrLeaseLost (mirrors ReleaseClaim / MarkFailed lease-lost handling).
func TestReleaseClaimForRetry_LeaseLost(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(
			statusFailed,
			time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC).Add(connectorRetriableReclaimDelay),
			"reason",
			time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
			time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
			"evt-retriable",
			statusProcessing,
			"stale-token",
		).
		WillReturnResult(sqlmock.NewResult(0, 0))

	err := svc.ReleaseClaimForRetry(context.Background(), "evt-retriable", "stale-token", "reason")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("want ErrLeaseLost, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 5 — MarkFailed: under threshold → status='failed', RetryCount++.
// ─────────────────────────────────────────────────────────────────────────────

func TestMarkFailed_UnderThreshold(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-2").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("noop-connector", "processing"))

	// No pack check — MarkFailed does not gate on pack activation state.

	mock.ExpectQuery(`SELECT RetryCount`).
		WithArgs("evt-2").
		WillReturnRows(sqlmock.NewRows([]string{"RetryCount"}).AddRow(1))

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := svc.MarkFailed(context.Background(), "evt-2", "token-abc", "timeout"); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 8 — ListDeadEvents: missing permission → ErrPermissionDenied.
// ─────────────────────────────────────────────────────────────────────────────

func TestListDeadEvents_PermissionDenied(t *testing.T) {
	svc, _ := newMock(t)
	_, err := svc.ListDeadEvents(context.Background(), denyPerm(), "noop-connector", Page{Limit: 10})
	if !errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("want ErrPermissionDenied, got %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 10 — ReplayDeadEvent: missing permission → ErrPermissionDenied.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_PermissionDenied(t *testing.T) {
	svc, _ := newMock(t)
	_, err := svc.ReplayDeadEvent(context.Background(), denyPerm(), "some-id", "install-cur", 1)
	if !errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("want ErrPermissionDenied, got %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 10b — ReplayDeadEvent: empty packInstallID → error.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_EmptyPackInstallID(t *testing.T) {
	svc, _ := newMock(t)
	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "some-id", "", 1)
	if err == nil {
		t.Fatal("expected error for empty packInstallID, got nil")
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 10c — ReplayDeadEvent: routeVersion <= 0 → ErrInvalidReplayRequest.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_InvalidRouteVersion(t *testing.T) {
	svc, _ := newMock(t)
	perm := allowPerm(string(constants.PermReplayConnectorEvents))

	for _, rv := range []int64{0, -1, -99} {
		_, err := svc.ReplayDeadEvent(context.Background(), perm, "some-id", "install-cur", rv)
		if !errors.Is(err, ErrInvalidReplayRequest) {
			t.Fatalf("routeVersion=%d: want ErrInvalidReplayRequest, got %v", rv, err)
		}
	}
}

func TestReplayDeadEvent_RejectsAttestedLegacyGenerationBeforeLegacyReservation(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-evt").
		WillReturnRows(eventRow("dead-evt", "noop-connector", "pack-old-001", "webhook.event", "ikey-orig", "dead", 5, ""))
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-cur")
	mock.ExpectQuery(`(?s)SELECT Version, Description.*FROM _System_Migration.*WHERE Id = \?.*LIMIT 1`).
		WithArgs(WorkItemOnlyExecutionCutoverMigrationID).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}).AddRow(WorkItemOnlyExecutionCutoverMigrationVersion, WorkItemOnlyExecutionCutoverMigrationDescription))
	mock.ExpectQuery(`(?s)SELECT COUNT\(\*\).*FROM _System_ConnectorWorkItem.*EventId = \?.*WorkType IN \(\?, \?\).*Attempt > 0`).
		WithArgs("dead-evt", connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(0))
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-evt", "install-token-cur", 1)
	if !errors.Is(err, ErrLegacyEventReplayRetired) {
		t.Fatalf("ReplayDeadEvent error = %v, want %v", err, ErrLegacyEventReplayRetired)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestReplayDeadEvent_RejectsOwnerlessReservationAfterAttestedAuthorityPreflight(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-workitem-race").
		WillReturnRows(eventRow("dead-workitem-race", "noop-connector", "pack-old-001", "webhook.event", "ikey-race", statusDead, 5, ""))
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-cur")
	mock.ExpectQuery(`(?s)SELECT Version, Description.*FROM _System_Migration.*WHERE Id = \?.*LIMIT 1`).
		WithArgs(WorkItemOnlyExecutionCutoverMigrationID).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}).AddRow(WorkItemOnlyExecutionCutoverMigrationVersion, WorkItemOnlyExecutionCutoverMigrationDescription))
	mock.ExpectQuery(`(?s)SELECT COUNT\(\*\).*FROM _System_ConnectorWorkItem.*EventId = \?.*WorkType IN \(\?, \?\).*Attempt > 0`).
		WithArgs("dead-workitem-race", connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(1))
	expectReplayEventLock(mock, "dead-workitem-race", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-workitem-race")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-workitem-race", "install-token-cur", 1)
	if !errors.Is(err, ErrLegacyEventReplayRetired) {
		t.Fatalf("ReplayDeadEvent error = %v, want %v", err, ErrLegacyEventReplayRetired)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 11 — ReplayDeadEvent: non-dead event → ErrEventNotDead.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_NotDead(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("evt-processing").
		WillReturnRows(eventRow("evt-processing", "noop-connector", "pack-test-001", "webhook.event", "ikey-x", "processing", 0, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-cur", "evt-processing", statusProcessing, nil)
	expectLegacyReplayOwnershipProbe(mock, "evt-processing")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "evt-processing", "install-cur", 1)
	if !errors.Is(err, ErrEventNotDead) {
		t.Fatalf("want ErrEventNotDead, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// GetDeadEvent — compatibility wrapper over the WorkItem-owned replay detail
// gate. ReplayConnectorEvents authorizes the read; effective dead state comes
// from WorkItem when one exists.
// ─────────────────────────────────────────────────────────────────────────────

func TestGetDeadEvent_PermissionDenied(t *testing.T) {
	svc, mock := newMock(t)
	// Perms gate rejects before any DB read — no query expectation.
	_, err := svc.GetDeadEvent(context.Background(), denyPerm(), "evt-1")
	if !errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("want ErrPermissionDenied, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestGetDeadEvent_NotDead(t *testing.T) {
	svc, mock := newMock(t)
	mock.ExpectQuery(`(?s)SELECT CASE WHEN w.WorkId IS NOT NULL AND w.Attempt > 0.*LEFT JOIN _System_ConnectorWorkItem.*WHERE e.Id = \?.*ORDER BY w.WorkId.*LIMIT 2`).
		WithArgs(connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, "evt-processing").
		WillReturnRows(deadWorkItemViewRow("work-processing", "evt-processing", "noop-connector", "pack-001", "webhook.event", "ikey-x", connector_workqueue.WorkStateProcessing, 0, false))

	_, err := svc.GetDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "evt-processing")
	if !errors.Is(err, ErrEventNotDead) {
		t.Fatalf("want ErrEventNotDead, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestReplayDeadEvent_ZeroAttemptMirrorIsFencedBeforeChild(t *testing.T) {
	svc, mock := newMock(t, "new-zero-attempt-replay")

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-zero-attempt").
		WillReturnRows(eventRow("dead-zero-attempt", "noop-connector", "pack-old-001", "webhook.event", "ikey-zero", statusDead, 0, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", "dead-zero-attempt", statusDead, nil)
	expectZeroAttemptReplayReservation(mock, "dead-zero-attempt", "work-zero-attempt", connector_workqueue.WorkStatePending)
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectZeroAttemptReplayFence(mock, "dead-zero-attempt", "work-zero-attempt", 1)
	expectRouteBackedReplayEventInsert(mock, "new-zero-attempt-replay", "noop-connector", "install-token-cur", "webhook.event", "dead-zero-attempt", 1, 1)
	expectConnectorWorkItemReplayMaterialization(mock, "new-zero-attempt-replay", "install-token-cur", "work-zero-attempt", nil)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	newID, err := svc.ReplayDeadEvent(
		context.Background(),
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dead-zero-attempt",
		"install-token-cur",
		1,
	)
	if err != nil || newID != "new-zero-attempt-replay" {
		t.Fatalf("ReplayDeadEvent = (%q, %v), want fenced replay child", newID, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestReplayDeadEvent_WorkItemOwnedPersistsLineage(t *testing.T) {
	svc, mock := newMock(t, "new-workitem-replay")

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-workitem-event").
		WillReturnRows(eventRow("dead-workitem-event", "noop-connector", "pack-old-001", "webhook.event", "ikey-workitem", statusProcessing, 5, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", "dead-workitem-event", statusProcessing, nil)
	expectWorkItemReplayReservation(mock, "dead-workitem-event", "work-original", "")
	expectRouteBackedReplayEventInsert(mock, "new-workitem-replay", "noop-connector", "install-token-cur", "webhook.event", "dead-workitem-event", 1, 1)
	expectConnectorWorkItemReplayMaterialization(mock, "new-workitem-replay", "install-token-cur", "work-original", nil)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	newID, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-workitem-event", "install-token-cur", 1)
	if err != nil {
		t.Fatalf("ReplayDeadEvent: %v", err)
	}
	if newID != "new-workitem-replay" {
		t.Fatalf("newID = %q, want new-workitem-replay", newID)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestReplayDeadEvent_WorkItemOwnedIgnoresDivergentTerminalEventMirror(t *testing.T) {
	for _, status := range []string{statusProcessed, statusSkipped} {
		t.Run(status, func(t *testing.T) {
			eventID := "dead-workitem-" + status + "-event"
			newID := "replay-" + status
			svc, mock := newMock(t, newID)
			mock.ExpectQuery(`SELECT Id, Namespace`).
				WithArgs(eventID).
				WillReturnRows(eventRow(eventID, "noop-connector", "pack-old-001", "webhook.event", "ikey-"+status, status, 5, ""))
			expectPessimisticBegin(mock)
			expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", eventID, status, nil)
			expectWorkItemReplayReservation(mock, eventID, "work-original-"+status, "")
			expectRouteBackedReplayEventInsert(mock, newID, "noop-connector", "install-token-cur", "webhook.event", eventID, 1, 1)
			expectConnectorWorkItemReplayMaterialization(mock, newID, "install-token-cur", "work-original-"+status, nil)
			mock.ExpectCommit()
			expectPessimisticCleanup(mock)

			got, err := svc.ReplayDeadEvent(
				context.Background(),
				allowPerm(string(constants.PermReplayConnectorEvents)),
				eventID,
				"install-token-cur",
				1,
			)
			if err != nil || got != newID {
				t.Fatalf("ReplayDeadEvent = (%q, %v), want WorkItem-owned replay %q", got, err, newID)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sqlmock expectations: %v", err)
			}
		})
	}
}

func TestReplayDeadEvent_WorkItemOwnedMaterializationFailureRollsBackBothReservations(t *testing.T) {
	svc, mock := newMock(t, "new-workitem-replay-failed")
	materializeErr := errors.New("work item insert failed")

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-workitem-event-failed").
		WillReturnRows(eventRow("dead-workitem-event-failed", "noop-connector", "pack-old-001", "webhook.event", "ikey-workitem-failed", "dead", 5, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", "dead-workitem-event-failed", statusDead, nil)
	expectWorkItemReplayReservation(mock, "dead-workitem-event-failed", "work-original-failed", "")
	expectRouteBackedReplayEventInsert(mock, "new-workitem-replay-failed", "noop-connector", "install-token-cur", "webhook.event", "dead-workitem-event-failed", 1, 1)
	expectConnectorWorkItemReplayMaterialization(mock, "new-workitem-replay-failed", "install-token-cur", "work-original-failed", materializeErr)
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-workitem-event-failed", "install-token-cur", 1)
	if !errors.Is(err, materializeErr) {
		t.Fatalf("ReplayDeadEvent error = %v, want %v", err, materializeErr)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 12b — ReplayDeadEvent: a committed legacy replay marker fails closed.
// ─────────────────────────────────────────────────────────────────────────────

// ─────────────────────────────────────────────────────────────────────────────
// Case 12c — ReplayDeadEvent: INSERT fails → transaction rolls back CAS UPDATE.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_InsertFailsRollsBackReservation(t *testing.T) {
	svc, mock := newMock(t, "rollback-id")

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-evt-rollback").
		WillReturnRows(eventRow("dead-evt-rollback", "noop-connector", "pack-old-001", "webhook.event", "ikey-rb", "dead", 5, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", "dead-evt-rollback", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-evt-rollback")

	// CAS UPDATE succeeds — replay right reserved.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	// INSERT fails — e.g. transient network error.
	insertErr := errors.New("connection lost")
	expectRouteBackedReplayInsertError(mock, "rollback-id", "noop-connector", "install-token-cur", "webhook.event", "dead-evt-rollback", 1, insertErr)

	// Transaction must be rolled back so ReplayedAt is not permanently consumed.
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-evt-rollback", "install-token-cur", 1)
	if err == nil {
		t.Fatal("expected error, got nil")
	}
	if !errors.Is(err, insertErr) {
		t.Fatalf("want insertErr propagated, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case 13 — ClaimNext: pending → processing, returns row.
// ─────────────────────────────────────────────────────────────────────────────

func TestClaimNext_RemainsOnLegacyConnectorEventPath(t *testing.T) {
	svc, mock := newMock(t, "token-legacy-claim")

	mock.ExpectQuery(`FROM _System_ConnectorEvent`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).AddRow("legacy-evt", nil, 0))
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(statusProcessing, "worker-legacy", sqlmock.AnyArg(), "token-legacy-claim", sqlmock.AnyArg(), "legacy-evt", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`FROM _System_ConnectorEvent`).
		WithArgs("legacy-evt").
		WillReturnRows(eventRow("legacy-evt", "noop-connector", "pack-test-001", "webhook.event", "ikey-legacy", "processing", 0, ""))

	ev, claimed, err := svc.ClaimNext(context.Background(), "pack-test-001", "worker-legacy")
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if !claimed || ev == nil || ev.ID != "legacy-evt" {
		t.Fatalf("ClaimNext = (%+v, %v), want legacy-evt/true", ev, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestClaimNextFallsBackWhenEventExecutionOwnerColumnIsUnavailable(t *testing.T) {
	svc, mock := newMock(t, "token-mixed-schema-claim")

	// The WorkItem table may already exist and EnqueueBatch may already have
	// materialized a direct row. Until Event.ExecutionOwner lands, the supervisor
	// deliberately runs the legacy claimant, so the complete-schema guard must
	// detect this additive gap and retry on the bounded legacy query.
	mock.ExpectQuery(`(?s)SELECT e\.Id.*e\.ExecutionOwner = 'event'`).
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'e.ExecutionOwner' in 'where clause'"))
	mock.ExpectQuery(`SELECT e\.Id`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).
			AddRow("mixed-schema-event", nil, 0))
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(statusProcessing, "legacy-worker", sqlmock.AnyArg(), "token-mixed-schema-claim", sqlmock.AnyArg(), "mixed-schema-event", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("mixed-schema-event").
		WillReturnRows(eventRow("mixed-schema-event", "noop-connector", "pack-test-001", "webhook.event", "ikey-mixed-schema", "processing", 0, ""))

	ev, claimed, err := svc.ClaimNext(context.Background(), "pack-test-001", "legacy-worker")
	if err != nil {
		t.Fatalf("ClaimNext mixed schema: %v", err)
	}
	if !claimed || ev == nil || ev.ID != "mixed-schema-event" {
		t.Fatalf("ClaimNext mixed schema = (%+v, %v), want mixed-schema-event/true", ev, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("mixed-schema expectations: %v", err)
	}
}

func TestClaimNextForClassesFiltersQueueClass(t *testing.T) {
	svc, mock := newMock(t, "token-bulk-001")

	// Candidate SELECT binds (NCRM #3139 + #3140): lane-head (pack, now),
	// delivery-head (pack), outer pack, candidate status (pending now, failed now,
	// stale), queue class, then the residual source-lane guard's two `now` binds.
	mock.ExpectQuery(`COALESCE\(e\.QueueClass, 'default'\) IN \(\?\)`).
		WithArgs("pack-test-001", sqlmock.AnyArg(), "pack-test-001", "pack-test-001",
			sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "bulk", sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).AddRow("bulk-evt", nil, 0))

	// UPDATE binds: 5 SET, eventID, candidate status (pending now, failed now,
	// stale), queue class,
	// then the guard's two source-lane `now` binds.
	mock.ExpectExec(`COALESCE\(e\.QueueClass, 'default'\) IN \(\?\)`).
		WithArgs(statusProcessing, "worker-bulk", sqlmock.AnyArg(), "token-bulk-001", sqlmock.AnyArg(), "bulk-evt", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "bulk", sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("bulk-evt").
		WillReturnRows(eventRow("bulk-evt", "noop-connector", "pack-test-001", "webhook.event", "ikey-bulk", "processing", 0, ""))

	ev, claimed, err := svc.ClaimNextForClasses(context.Background(), "pack-test-001", "worker-bulk", []string{" bulk "})
	if err != nil {
		t.Fatalf("ClaimNextForClasses: %v", err)
	}
	if !claimed || ev == nil || ev.ID != "bulk-evt" {
		t.Fatalf("ClaimNextForClasses = (%v, %v), want bulk-evt/true", ev, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B1 — Pack-gate removal: finalize/drain must work regardless of pack state.
// ─────────────────────────────────────────────────────────────────────────────

// TestClaimNext_InactivePackReclaimsStaleProcessing verifies that ClaimNext
// succeeds without querying the pack status — the lease (UPDATE-WHERE predicate)
// is the sole authorization once a row is enqueued.
func TestClaimNext_InactivePackReclaimsStaleProcessing(t *testing.T) {
	// idgen produces the claim token for the reclaim UPDATE.
	svc, mock := newMock(t, "token-inactive-reclaim-001")

	// No packRow expectation — ClaimNext does not gate on pack activation state.
	// Workers supply the InstalledPackId directly; pack lookup is not needed.
	mock.ExpectQuery(`SELECT e\.Id`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).AddRow("evt-stale-inactive", nil, 0))

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(statusProcessing, "worker-reclaim", sqlmock.AnyArg(), "token-inactive-reclaim-001", sqlmock.AnyArg(), "evt-stale-inactive", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("evt-stale-inactive").
		WillReturnRows(eventRow("evt-stale-inactive", "noop-connector", "pack-test-001", "webhook.event", "ikey-si", "processing", 1, ""))

	ev, claimed, err := svc.ClaimNext(context.Background(), "pack-test-001", "worker-reclaim")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if ev == nil || ev.ID != "evt-stale-inactive" {
		t.Fatalf("expected evt-stale-inactive, got %v", ev)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestMarkProcessed_NoPackCheckDuringFinalize verifies that MarkProcessed does
// not depend on pack activation while finalizing an existing Event.
func TestMarkProcessed_NoPackCheckDuringFinalize(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-mp-nopk").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("inactive-connector", statusProcessing))
	expectLegacyTerminalEventNoMirror(mock, "evt-mp-nopk", statusProcessing, "token-abc")
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	if err := svc.MarkProcessed(context.Background(), "evt-mp-nopk", "token-abc"); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestMarkFailed_NoPackCheckDuringFinalize verifies that MarkFailed only
// issues loadNamespaceAndStatus + RetryCount SELECT + UPDATE — no pack lookup.
func TestMarkFailed_NoPackCheckDuringFinalize(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-mf-nopk").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("noop-connector", "processing"))

	// No packRow expectation.

	mock.ExpectQuery(`SELECT RetryCount`).
		WithArgs("evt-mf-nopk").
		WillReturnRows(sqlmock.NewRows([]string{"RetryCount"}).AddRow(1))

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := svc.MarkFailed(context.Background(), "evt-mf-nopk", "token-abc", "transient"); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestListDeadEvents_NoPackCheckDuringDrain verifies that ListDeadEvents issues
// only the dead-row SELECT — no pack lookup — so operators can drain during
// pack removing/inactive transitions.
func TestListDeadEvents_NoPackCheckDuringDrain(t *testing.T) {
	svc, mock := newMock(t)

	// No packRow expectation.
	mock.ExpectQuery(`(?s)SELECT CASE WHEN w.WorkId IS NOT NULL AND w.Attempt > 0.*LEFT JOIN _System_ConnectorWorkItem`).
		WithArgs(connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, "noop-connector", connector_workqueue.WorkStateDead, statusDead, 10, 0).
		WillReturnRows(deadWorkItemViewRow("work-dead-drain", "dead-drain", "noop-connector", "pack-test-001", "webhook.event", "ikey-drain", connector_workqueue.WorkStateDead, 5, false))

	events, err := svc.ListDeadEvents(context.Background(), allowPerm(string(constants.PermViewConnectorEvents)), "noop-connector", Page{Limit: 10})
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(events) != 1 || events[0].ID != "dead-drain" {
		t.Fatalf("expected 1 dead event with id dead-drain, got %v", events)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B2 — Bounded replay key: "replay:"+newID never overflows VARCHAR(255).
// ─────────────────────────────────────────────────────────────────────────────

// TestReplayDeadEvent_BoundedReplayKey seeds a dead event with a 240-byte
// IdempotencyKey and asserts the inserted replay key starts with "replay:"
// and is at most 255 chars, regardless of original key length.
func TestReplayDeadEvent_BoundedReplayKey(t *testing.T) {
	const deterministicNewID = "replay-new-001"
	svc, mock := newMock(t, deterministicNewID)

	longKey := strings.Repeat("a", 240)

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-long-key").
		WillReturnRows(eventRow("dead-long-key", "noop-connector", "pack-old-001", "webhook.event", longKey, "dead", 5, ""))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-token-cur", "dead-long-key", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-long-key")

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	// Capture the INSERT args to verify the replay key.
	wantReplayKey := "replay:" + deterministicNewID
	expectRouteBackedReplayInsert(mock, deterministicNewID, "noop-connector", "install-token-cur", "webhook.event", "dead-long-key", 1, 1)

	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	newID, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-long-key", "install-token-cur", 1)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if newID != deterministicNewID {
		t.Fatalf("want newID=%q, got %q", deterministicNewID, newID)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}

	// Verify the replay key is bounded: "replay:"+newID = 7+14 = 21 chars.
	// The key must start with "replay:" and be well under VARCHAR(255).
	if len(wantReplayKey) > 255 {
		t.Errorf("replay key length %d exceeds VARCHAR(255)", len(wantReplayKey))
	}
	if !strings.HasPrefix(wantReplayKey, "replay:") {
		t.Errorf("replay key %q does not start with replay:", wantReplayKey)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B1 — Cross-incarnation dedup isolation: same IdempotencyKey under two
// different InstalledPackIds must produce two distinct events (not a duplicate).
// ─────────────────────────────────────────────────────────────────────────────

// TestEnqueue_DedupCrossIncarnationIsolated verifies that two Enqueue calls
// with the same IdempotencyKey but belonging to different pack incarnations
// (different InstalledPackIds returned by ResolveConnectorPackByNamespace)
// each succeed and produce independent event rows.  The unique constraint is
// (InstalledPackId, IdempotencyKey) so old-incarnation keys do not collide
// with new-incarnation keys.
func TestEnqueue_DedupCrossIncarnationIsolated(t *testing.T) {
	svc, mock := newMock(t, "evt-incarnation-a", "evt-incarnation-b")

	// Enqueue for incarnation A — caller supplies packInstallID="install-token-a".
	// No packRow expectation — Enqueue does not call resolvePack.
	// Dedup SELECT runs inside tx after lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-a")
	expectNextEnqueueOrdinal(mock, "install-token-a", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-a", "shared-ikey").
		WillReturnError(sql.ErrNoRows) // not a duplicate for this incarnation
	expectRouteBackedEnqueueInsert(mock, "evt-incarnation-a", "noop-connector", "install-token-a", "webhook.event", "shared-ikey", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	idA, isDupA, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "shared-ikey", nil, "install-token-a", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue A: unexpected error: %v", err)
	}
	if isDupA {
		t.Fatal("Enqueue A: expected isDuplicate=false for incarnation A")
	}
	if idA != "evt-incarnation-a" {
		t.Fatalf("Enqueue A: want evt-incarnation-a, got %q", idA)
	}

	// Enqueue for incarnation B — same IdempotencyKey, different InstalledPackId.
	// Dedup SELECT runs inside tx after lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-token-b")
	expectNextEnqueueOrdinal(mock, "install-token-b", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-token-b", "shared-ikey").
		WillReturnError(sql.ErrNoRows) // not a duplicate for incarnation B
	expectRouteBackedEnqueueInsert(mock, "evt-incarnation-b", "noop-connector", "install-token-b", "webhook.event", "shared-ikey", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	idB, isDupB, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "shared-ikey", nil, "install-token-b", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("Enqueue B: unexpected error: %v", err)
	}
	if isDupB {
		t.Fatal("Enqueue B: expected isDuplicate=false for incarnation B (separate dedup scope)")
	}
	if idB != "evt-incarnation-b" {
		t.Fatalf("Enqueue B: want evt-incarnation-b, got %q", idB)
	}
	if idA == idB {
		t.Fatal("incarnation A and B must produce distinct event IDs")
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Schema-neutrality: no provider names in production source.
// ─────────────────────────────────────────────────────────────────────────────

func TestSchemaNeutrality_NoProviderNames(t *testing.T) {
	src, err := os.ReadFile("service.go")
	if err != nil {
		t.Fatalf("ReadFile service.go: %v", err)
	}
	content := strings.ToLower(string(src))
	for _, vendor := range []string{"hubspot", "linear", "slack", "stripe"} {
		if strings.Contains(content, vendor) {
			t.Errorf("service.go contains vendor name %q — L0 neutrality violation", vendor)
		}
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// C1 — ClaimNext mints a fresh ClaimToken per successful claim.
// ─────────────────────────────────────────────────────────────────────────────

// TestClaimNext_MintsFreshClaimToken verifies that ClaimNext mints a fresh
// per-claim token via s.idgen(), writes it into the UPDATE, and the returned
// ConnectorEvent carries that token in its ClaimToken field.
func TestClaimNext_MintsFreshClaimToken(t *testing.T) {
	// idgen produces "fresh-token-001" as the first call (used as the claim token).
	svc, mock := newMock(t, "fresh-token-001")

	// SELECT picks "evt-existing".
	mock.ExpectQuery(`SELECT e\.Id`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).AddRow("evt-existing", nil, 0))

	// UPDATE — ClaimToken "fresh-token-001" is among the bound args.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(statusProcessing, "worker-1", sqlmock.AnyArg(), "fresh-token-001", sqlmock.AnyArg(), "evt-existing", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))

	// loadByID returns the row with ClaimToken set.
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("evt-existing").
		WillReturnRows(eventRowWithToken(
			"evt-existing", "noop-connector", "pack-test-001", "webhook.event", "ikey-e",
			"processing", "fresh-token-001", 0,
		))

	ev, claimed, err := svc.ClaimNext(context.Background(), "pack-test-001", "worker-1")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if ev == nil {
		t.Fatal("expected non-nil event")
	}
	if ev.ClaimToken != "fresh-token-001" {
		t.Errorf("want ClaimToken=fresh-token-001, got %q", ev.ClaimToken)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// C2 — MarkProcessed: wrong ClaimToken (stale / different lease) → ErrLeaseLost.
// ─────────────────────────────────────────────────────────────────────────────

func TestMarkProcessed_LeaseTokenMismatch(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-mp-token").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("noop-connector", statusProcessing))
	expectPessimisticBegin(mock)
	mock.ExpectQuery(`SELECT Status, ClaimToken`).
		WithArgs("evt-mp-token").
		WillReturnRows(sqlmock.NewRows([]string{"Status", "ClaimToken"}).
			AddRow(statusProcessing, "current-token"))
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	err := svc.MarkProcessed(context.Background(), "evt-mp-token", "wrong-token")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("want ErrLeaseLost, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// C3 — MarkFailed: wrong ClaimToken → ErrLeaseLost.
// ─────────────────────────────────────────────────────────────────────────────

func TestMarkFailed_LeaseTokenMismatch(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`SELECT Namespace, Status`).
		WithArgs("evt-mf-token").
		WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Status"}).
			AddRow("noop-connector", "processing"))

	mock.ExpectQuery(`SELECT RetryCount`).
		WithArgs("evt-mf-token").
		WillReturnRows(sqlmock.NewRows([]string{"RetryCount"}).AddRow(1))

	// Stale or wrong token — UPDATE WHERE ClaimToken=? matches 0 rows.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 0))

	err := svc.MarkFailed(context.Background(), "evt-mf-token", "wrong-token", "timeout")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("want ErrLeaseLost, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Connector permission gate behavior (NCRM-INT-02a, #1449).
//
// These tests exercise the read-vs-replay permission split on the DLQ service
// for two example connector key sets. They do NOT define or pin a canonical
// "bundle": bundle names and exact key membership are application-shaped policy
// that belongs in future L1/L2 PermissionSet metadata, not in tests or docs
// (see docs/extension-model/connector-permission-bundles.md). The key sets here
// are test-local fixtures, not production constants:
//
//   - observer key set = ViewConnectorEvents + ViewConnectorHealth (read-only)
//   - operator key set = the four connector keys (read + replay + manage)
//
// What is actually asserted is narrow and owned by THIS service: an actor whose
// key set lacks ReplayConnectorEvents cannot replay; one that holds it can. The
// ViewConnectorHealth / ManageNamedCredentials membership is gated in other
// packages and is not asserted here.
// ─────────────────────────────────────────────────────────────────────────────

func observerKeySet() fakePerm {
	return allowPerm(
		string(constants.PermViewConnectorEvents),
		string(constants.PermViewConnectorHealth),
	)
}

func operatorKeySet() fakePerm {
	return allowPerm(
		string(constants.PermManageNamedCredentials),
		string(constants.PermViewConnectorEvents),
		string(constants.PermReplayConnectorEvents),
		string(constants.PermViewConnectorHealth),
	)
}

// connector.observer must NOT be able to replay (no ReplayConnectorEvents key).
func TestConnectorObserverBundle_CannotReplay(t *testing.T) {
	svc, _ := newMock(t)
	_, err := svc.ReplayDeadEvent(context.Background(), observerKeySet(), "any-id", "install-abc", 1)
	if !errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("connector.observer must not replay: want ErrPermissionDenied, got %v", err)
	}
}

// connector.observer CAN read the DLQ (has ViewConnectorEvents) — proves the
// bundle is "watch, don't act", not "no access".
func TestConnectorObserverBundle_CanView(t *testing.T) {
	svc, mock := newMock(t)
	mock.ExpectQuery(`(?s)SELECT CASE WHEN w.WorkId IS NOT NULL AND w.Attempt > 0.*LEFT JOIN _System_ConnectorWorkItem`).
		WithArgs(connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, "noop-connector", connector_workqueue.WorkStateDead, statusDead, 10, 0).
		WillReturnRows(deadWorkItemViewRow("work-dead-1", "dead-1", "noop-connector", "pack-test-001", "webhook.event", "ikey-d", connector_workqueue.WorkStateDead, 5, false))

	if _, err := svc.ListDeadEvents(context.Background(), observerKeySet(), "noop-connector", Page{Limit: 10}); err != nil {
		t.Fatalf("connector.observer must be able to read the DLQ: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// connector.operator passes the replay permission gate (has ReplayConnectorEvents).
// A non-dead target makes the call return ErrEventNotDead, which proves the gate
// was cleared (it is NOT ErrPermissionDenied).
func TestConnectorOperatorBundle_CanReplay(t *testing.T) {
	svc, mock := newMock(t)
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("evt-processing").
		WillReturnRows(eventRow("evt-processing", "noop-connector", "pack-test-001", "webhook.event", "ikey-x", "processing", 0, ""))
	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-abc", "evt-processing", statusProcessing, nil)
	expectLegacyReplayOwnershipProbe(mock, "evt-processing")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), operatorKeySet(), "evt-processing", "install-abc", 1)
	if errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("connector.operator must clear the replay permission gate, got ErrPermissionDenied")
	}
	if !errors.Is(err, ErrEventNotDead) {
		t.Fatalf("want ErrEventNotDead (gate cleared, event not dead), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case R9a — Enqueue: pack incarnation changed between Step 2 and INSERT
// → RowsAffected=0 → ErrPackIncarnationChanged.
//
// This test simulates the race window identified in CI Codex R8: the dispatcher
// resolves packInstallID='install-old' at Step 2, but between that resolve and
// the Enqueue INSERT, the admin reinstalls the pack so 'install-old' is no
// longer Status='active'. The INSERT…SELECT returns 0 rows.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_PackIncarnationChanged_ReturnsErrPackIncarnationChanged(t *testing.T) {
	svc, mock := newMock(t, "id-incarnation")

	// Dedup no longer runs before the lock (issue #2161).
	// The lock-missing path returns ErrPackIncarnationChanged immediately,
	// before any dedup SELECT. No pre-tx SELECT Id expectation.
	expectPessimisticBegin(mock)
	expectActivePackLockMissing(mock, "noop-connector", "install-old")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "ikey-race", json.RawMessage(`{}`), "install-old", 1, "", "", 0, "")
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case R9b — Enqueue: active pack unchanged → INSERT writes 1 row → success.
//
// Sanity check: when (namespace, packInstallID) is still the active incarnation,
// the INSERT…SELECT returns 1 row and Enqueue succeeds as before.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_ActivePackUnchanged_Succeeds(t *testing.T) {
	svc, mock := newMock(t, "id-stable")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	// Order: begin → lock → SELECT Id (no row) → INSERT → commit → cleanup.
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-stable")
	expectNextEnqueueOrdinal(mock, "install-stable", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-stable", "ikey-stable").
		WillReturnError(sql.ErrNoRows)
	// INSERT…SELECT: _System_InstalledPack has an active row for 'install-stable'
	// → RowsAffected=1 → success.
	expectRouteBackedEnqueueInsert(mock, "id-stable", "noop-connector", "install-stable", "webhook.event", "ikey-stable", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "ikey-stable", json.RawMessage(`{}`), "install-stable", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if isDup {
		t.Fatal("expected isDuplicate=false")
	}
	if id != "id-stable" {
		t.Fatalf("want id-stable, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueue_LocksInstalledPackRowBeforeInsert(t *testing.T) {
	cases := []struct {
		name         string
		routeVersion int64
		installID    string
		idempotency  string
	}{
		{
			name:         "route-backed enqueue",
			routeVersion: 1,
			installID:    "install-lock-route",
			idempotency:  "ikey-lock-route",
		},
		{
			name:         "routeVersion zero enqueue",
			routeVersion: 0,
			installID:    "install-lock-zero",
			idempotency:  "ikey-lock-zero",
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			svc, mock := newMock(t, "id-lock")
			// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
			expectPessimisticBegin(mock)
			expectActivePackLock(mock, "noop-connector", tc.installID)
			expectNextEnqueueOrdinal(mock, tc.installID, 1)
			mock.ExpectQuery(`SELECT Id`).
				WithArgs(tc.installID, tc.idempotency).
				WillReturnError(sql.ErrNoRows)
			if tc.routeVersion >= 1 {
				expectRouteBackedEnqueueInsert(mock, "id-lock", "noop-connector", tc.installID, "webhook.event", tc.idempotency, tc.routeVersion, 1)
			} else {
				expectRouteZeroEnqueueInsert(mock, "id-lock", "noop-connector", tc.installID, "webhook.event", tc.idempotency, 1)
			}
			mock.ExpectCommit()
			expectPessimisticCleanup(mock)

			id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", tc.idempotency, json.RawMessage(`{}`), tc.installID, tc.routeVersion, "", "", 0, "")
			if err != nil {
				t.Fatalf("Enqueue: %v", err)
			}
			if isDup {
				t.Fatal("isDuplicate = true, want false")
			}
			if id != "id-lock" {
				t.Fatalf("event id = %q, want id-lock", id)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Errorf("unmet expectations: %v", err)
			}
		})
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Case R9c — ReplayDeadEvent: requested pack incarnation is no longer active.
//
// Replay locks the active-pack row before Event and WorkItem reservation. A
// retired incarnation therefore returns ErrPackIncarnationChanged without
// consuming either replay marker.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_PackIncarnationChanged_ReturnsErrPackIncarnationChanged(t *testing.T) {
	svc, mock := newMock(t, "replay-incarnation-id")

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-evt-incarnation").
		WillReturnRows(eventRow("dead-evt-incarnation", "noop-connector", "pack-old-001", "webhook.event", "ikey-inc", "dead", 5, ""))

	expectPessimisticBegin(mock)
	expectActivePackLockMissing(mock, "noop-connector", "install-old")

	// Deferred rollback restores ReplayedAt=NULL on the original row.
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(context.Background(), allowPerm(string(constants.PermReplayConnectorEvents)), "dead-evt-incarnation", "install-old", 1)
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged, got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B20 — Enqueue: route version swept between ResolveFlow and INSERT.
//
// When routeVersion >= 1 the INSERT…SELECT JOINs _System_ConnectorInboundRoute
// on (InstalledPackId, EventKind, Version, IsDeleted=0). If the sweeper
// soft-deleted the route row between dispatcher's ResolveFlow call and Enqueue,
// the JOIN returns no rows → INSERT writes 0 rows → ErrPackIncarnationChanged.
//
// This test verifies that 0 RowsAffected from the with-JOIN INSERT path is
// mapped to ErrPackIncarnationChanged (retriable 503), not a silent success.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_RouteSweptBetweenResolveAndInsert_ReturnsTransient(t *testing.T) {
	svc, mock := newMock(t, "id-b20")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	// Order: begin → lock → SELECT Id (no row) → INSERT (route swept, 0 rows) → rollback → cleanup.
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-live")
	expectNextEnqueueOrdinal(mock, "install-live", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-live", "ikey-b20").
		WillReturnError(sql.ErrNoRows)
	// INSERT…SELECT with routeVersion=2 (>=1) JOINs ConnectorInboundRoute.
	// Sweeper soft-deleted version 2 between ResolveFlow and Enqueue → JOIN
	// finds no row → RowsAffected=0.
	expectRouteBackedEnqueueInsert(mock, "id-b20", "noop-connector", "install-live", "webhook.event", "ikey-b20", 2, 0)
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.Enqueue(context.Background(), "noop-connector", "webhook.event", "ikey-b20", json.RawMessage(`{}`), "install-live", 2, "", "", 0, "")
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (route swept), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B20 — Enqueue: routeVersion=0 (no-route sentinel) bypasses route JOIN.
//
// When routeVersion=0 no route row exists to validate. The INSERT uses the
// InstalledPack-only guard (original R9 path). This test verifies:
//   - routeVersion=0 still produces a successful INSERT when pack is active.
//   - routeVersion=0 still returns ErrPackIncarnationChanged when pack rotated.
//
// These two sub-cases confirm the routeVersion=0 path is unchanged by B20.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_RouteVersion0_BypassesRouteJoin_Succeeds(t *testing.T) {
	svc, mock := newMock(t, "id-b20-zero")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-live")
	expectNextEnqueueOrdinal(mock, "install-live", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-live", "ikey-zero").
		WillReturnError(sql.ErrNoRows)
	// routeVersion=0: INSERT uses InstalledPack-only guard (no JOIN).
	// Pack is still active → RowsAffected=1.
	expectRouteZeroEnqueueInsert(mock, "id-b20-zero", "noop-connector", "install-live", "no-route.event", "ikey-zero", 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "no-route.event", "ikey-zero", nil, "install-live", 0, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if isDup {
		t.Fatal("expected isDuplicate=false")
	}
	if id != "id-b20-zero" {
		t.Fatalf("want id-b20-zero, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueue_RouteVersion0_PackRotated_ReturnsErrPackIncarnationChanged(t *testing.T) {
	svc, mock := newMock(t, "id-b20-zero-rot")

	// Dedup no longer runs before the lock (issue #2161).
	// Lock-missing path returns ErrPackIncarnationChanged before any dedup SELECT.
	expectPessimisticBegin(mock)
	expectActivePackLockMissing(mock, "noop-connector", "install-old-zero")
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.Enqueue(context.Background(), "noop-connector", "no-route.event", "ikey-zero-rot", nil, "install-old-zero", 0, "", "", 0, "")
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (pack rotated, v0 path), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// R20 — Enqueue: preserved-for-backlog row (IsCurrent=0, IsDeleted=0) must NOT
// be reachable from live-traffic enqueue.
//
// Race: dispatcher calls ResolveFlow and gets routeVersion=2 just before the
// pack sweeper drops that EventKind and sets IsCurrent=0 (preserved for backlog
// drain).  The subsequent Enqueue with routeVersion=2 must fail the route JOIN
// (IsCurrent=0 row excluded) → RowsAffected=0 → ErrPackIncarnationChanged →
// dispatcher 503.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_RouteIsCurrent0_ReturnsTransient(t *testing.T) {
	svc, mock := newMock(t, "id-r20")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-live-r20")
	expectNextEnqueueOrdinal(mock, "install-live-r20", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-live-r20", "ikey-r20").
		WillReturnError(sql.ErrNoRows)
	// INSERT…SELECT with routeVersion=2 JOINs ConnectorInboundRoute.
	// The route row exists (IsDeleted=0) but is preserved-for-backlog
	// (IsCurrent=0) — sweeper dropped this EventKind after ResolveFlow.
	// The new AND COALESCE(r.IsCurrent,0)=1 predicate excludes the row
	// → RowsAffected=0.
	expectRouteBackedEnqueueInsert(mock, "id-r20", "noop-connector", "install-live-r20", "contact.created", "ikey-r20", 2, 0)
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.Enqueue(context.Background(), "noop-connector", "contact.created", "ikey-r20", json.RawMessage(`{}`), "install-live-r20", 2, "", "", 0, "")
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (route IsCurrent=0), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// B27 — Enqueue: mixed-case eventKind is canonicalized before route JOIN.
//
// Route rows are stored with canonical (lower-trimmed) EventKind. A parser or
// HTTP handler may supply "Contact.Created" (mixed case). Without normalization
// the route JOIN (r.EventKind = ?) would miss the canonical row → RowsAffected=0
// → ErrPackIncarnationChanged → endless 503.
//
// After the B27 fix Enqueue normalizes eventKind = strings.ToLower(strings.TrimSpace(...))
// before the INSERT…SELECT, so the JOIN succeeds.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_MixedCaseEventKind_CanonicalizedAndJoinSucceeds(t *testing.T) {
	svc, mock := newMock(t, "id-b27-mc")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-b27")
	expectNextEnqueueOrdinal(mock, "install-b27", 1)
	// No existing row for the canonical install + key.
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-b27", "ikey-b27-mc").
		WillReturnError(sql.ErrNoRows)
	// Enqueue with mixed-case "Contact.Created" (routeVersion=1).
	// After normalization the INSERT binds "contact.created" which matches
	// the canonical route row → RowsAffected=1.
	expectRouteBackedEnqueueInsert(mock, "id-b27-mc", "noop-connector", "install-b27", "Contact.Created", "ikey-b27-mc", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "Contact.Created", "ikey-b27-mc", json.RawMessage(`{}`), "install-b27", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if isDup {
		t.Fatal("expected isDuplicate=false")
	}
	if id != "id-b27-mc" {
		t.Fatalf("want id-b27-mc, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestEnqueue_StoredEventKindIsCanonical verifies that the eventKind stored in
// the _System_ConnectorEvent row is the canonical (lower-trimmed) form, not the
// raw caller-supplied value.
//
// sqlmock does not let us inspect INSERT column values directly, but we can
// verify indirectly: the dedup SELECT after a concurrent-race path re-reads the
// row, and if Enqueue stored the raw mixed-case value then a second call for
// the same installID+idempotencyKey would not find it via the canonical lookup.
// Instead we just verify that calling Enqueue with "  CONTACT.CREATED  " (with
// surrounding whitespace) does NOT return an error — canonicalization is the
// prerequisite for a correct route JOIN in the INSERT…SELECT.
func TestEnqueue_StoredEventKindIsCanonical(t *testing.T) {
	svc, mock := newMock(t, "id-b27-canon")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "noop-connector", "install-canon")
	expectNextEnqueueOrdinal(mock, "install-canon", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-canon", "ikey-canon").
		WillReturnError(sql.ErrNoRows)
	// Whitespace-padded + upper-case input canonicalizes to "contact.created"
	// which matches the canonical route row → INSERT succeeds.
	expectRouteBackedEnqueueInsert(mock, "id-b27-canon", "noop-connector", "install-canon", "  CONTACT.CREATED  ", "ikey-canon", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(context.Background(), "noop-connector", "  CONTACT.CREATED  ", "ikey-canon", json.RawMessage(`{}`), "install-canon", 1, "", "", 0, "")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if isDup {
		t.Fatal("expected isDuplicate=false")
	}
	if id != "id-b27-canon" {
		t.Fatalf("want id-b27-canon, got %q", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestReplayDeadEvent_MixedCaseEventKind_StoredCanonical verifies that
// ReplayDeadEvent stores the canonical (lower-trimmed) EventKind on the new
// replay row even when the original dead event was stored with a raw mixed-case
// or whitespace-padded EventKind (pre-canonicalization legacy row).
func TestReplayDeadEvent_MixedCaseEventKind_StoredCanonical(t *testing.T) {
	svc, mock := newMock(t, "id-b27-replay")

	// Dead event row — simulates a pre-canonicalization legacy row with raw
	// mixed-case EventKind ("Contact.Created").
	now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
	mock.ExpectQuery(`SELECT Id, Namespace, InstalledPackId, EventKind`).
		WithArgs("dead-evt-b27").
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
			"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
			"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
			"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey", "DeliveryId",
			"DeliverySequence", "CreatedDate", "LastModifiedDate",
		}).AddRow(
			"dead-evt-b27", "noop-connector", "install-b27", "Contact.Created", "orig-key", statusDead,
			"default", []byte(`{}`), 5, nil, nil,
			nil, nil, nil, nil, nil, nil,
			int64(1), nil, nil, nil, nil,
			0, now, now,
		))

	// CAS UPDATE reserves the replay right.
	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-b27", "dead-evt-b27", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-evt-b27")
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg(), "dead-evt-b27", statusDead).
		WillReturnResult(sqlmock.NewResult(1, 1))

	// INSERT new replay row — the EventKind must be canonical "contact.created"
	// (sqlmock matches the full INSERT; we verify no error which confirms the
	// canonical value threaded through correctly).
	expectRouteBackedReplayInsert(mock, "id-b27-replay", "noop-connector", "install-b27", "Contact.Created", "dead-evt-b27", 1, 1)
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	newID, err := svc.ReplayDeadEvent(
		context.Background(),
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dead-evt-b27",
		"install-b27",
		1,
	)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if newID != "id-b27-replay" {
		t.Fatalf("want id-b27-replay, got %q", newID)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// R22 — ReplayDeadEvent: route swept between route resolution and INSERT
// → INSERT…SELECT RowsAffected=0 → ErrPackIncarnationChanged; original dead
// row's ReplayedAt is restored to NULL (admin can retry after re-resolving).
//
// Symmetric to TestEnqueue_RouteSweptBetweenResolveAndInsert_ReturnsTransient
// (B20 for Enqueue).  The sweeper clears IsCurrent (preserved-for-backlog) or
// soft-deletes the route row; either way the JOIN in the INSERT returns no rows.
// ─────────────────────────────────────────────────────────────────────────────

func TestReplayDeadEvent_RouteSweptBetweenResolveAndInsert_ReturnsTransientAndPreservesOriginal(t *testing.T) {
	svc, mock := newMock(t, "id-r22-swept")

	// Dead event with RouteVersion=2.
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-evt-r22").
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
			"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
			"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
			"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey", "DeliveryId",
			"DeliverySequence", "CreatedDate", "LastModifiedDate",
		}).AddRow(
			"dead-evt-r22", "noop-connector", "install-r22", "contact.created", "ikey-r22", statusDead,
			"default", []byte(`{}`), 5, nil, nil,
			nil, nil, nil, nil, nil, nil,
			int64(2), nil, nil, nil, nil,
			0, time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
		))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-r22", "dead-evt-r22", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-evt-r22")

	// CAS UPDATE reserves the replay right.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	// INSERT…SELECT: route version 2 was swept (IsCurrent=0 or IsDeleted=1) between
	// the caller's route resolution and this INSERT → JOIN finds no row → RowsAffected=0.
	expectRouteBackedReplayInsert(mock, "id-r22-swept", "noop-connector", "install-r22", "contact.created", "dead-evt-r22", 2, 0)

	// Deferred rollback restores ReplayedAt=NULL on the original row so the admin
	// can retry after re-resolving the route.
	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(
		context.Background(),
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dead-evt-r22",
		"install-r22",
		2,
	)
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (route swept), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// HIGH1 regression (updated) — ListDeadEvents now delegates to the WorkItem-owned
// metadata projection. Payload, LastError, ClaimedBy/At/Token, RouteVersion,
// CompanionPackInstallId, and CompanionFlowVersionId are not selected; the
// compatibility Event shape must retain only safe metadata.
// ─────────────────────────────────────────────────────────────────────────────

func TestListDeadEvents_MetadataProjectionRoundTrip(t *testing.T) {
	svc, mock := newMock(t)

	mock.ExpectQuery(`(?s)SELECT CASE WHEN w.WorkId IS NOT NULL AND w.Attempt > 0.*LEFT JOIN _System_ConnectorWorkItem`).
		WithArgs(connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply, "ns-dlq", connector_workqueue.WorkStateDead, statusDead, 50, 0).
		WillReturnRows(deadWorkItemViewRow("work-dlq-001", "evt-dlq-001", "ns-dlq", "install-dlq", "webhook.dead", "ikey-dlq", connector_workqueue.WorkStateDead, 5, false))

	events, err := svc.ListDeadEvents(
		context.Background(),
		allowPerm(string(constants.PermViewConnectorEvents)),
		"ns-dlq",
		Page{Limit: 50, Offset: 0},
	)
	if err != nil {
		t.Fatalf("ListDeadEvents: unexpected scan error: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event, got %d", len(events))
	}
	ev := events[0]
	if ev.ID != "evt-dlq-001" {
		t.Errorf("ID: want %q, got %q", "evt-dlq-001", ev.ID)
	}
	if ev.Namespace != "ns-dlq" {
		t.Errorf("Namespace: want %q, got %q", "ns-dlq", ev.Namespace)
	}
	if ev.Status != statusDead {
		t.Errorf("Status: want %q, got %q", statusDead, ev.Status)
	}
	if ev.RetryCount != 5 {
		t.Errorf("RetryCount: want 5, got %d", ev.RetryCount)
	}
	// Payload/LastError are not selected — they stay zero-valued.
	if len(ev.Payload) != 0 {
		t.Errorf("Payload must be zero-valued for metadata projection, got %s", ev.Payload)
	}
	if ev.LastError != "" {
		t.Errorf("LastError must be zero-valued for metadata projection, got %q", ev.LastError)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// HIGH3 regression — Enqueue cross-pack: companion incarnation guard rejects
// stale companion pack (RowsAffected=0 → ErrPackIncarnationChanged).
// Same-pack enqueue (no companion) is unaffected.
// ─────────────────────────────────────────────────────────────────────────────

func TestEnqueue_CrossPack_CompanionInactiveReturnsRetriable(t *testing.T) {
	svc, mock := newMock(t, "id-cp-001")

	// Dedup SELECT now runs inside the tx, AFTER the lock (issue #2161).
	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "ns-provider", "install-provider")
	expectNextEnqueueOrdinal(mock, "install-provider", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-provider", "ikey-xp").
		WillReturnError(sql.ErrNoRows)

	// Cross-pack enqueue locks the companion installed-pack row before insert.
	// Missing/inactive companion maps to retriable ErrPackIncarnationChanged.
	expectActiveCompanionPackLockMissing(mock, "companion-install-stale")

	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, _, err := svc.Enqueue(
		context.Background(),
		"ns-provider", "webhook.xpack", "ikey-xp", json.RawMessage(`{}`),
		"install-provider", 1,
		"companion-install-stale", "companion-flow-v1",
		0, "",
	)
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (companion inactive), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

func TestEnqueue_CrossPack_LocksCompanionBeforeInsert(t *testing.T) {
	svc, mock := newMock(t, "id-cp-lock")

	expectPessimisticBegin(mock)
	expectActivePackLock(mock, "ns-provider", "install-provider")
	expectNextEnqueueOrdinal(mock, "install-provider", 1)
	mock.ExpectQuery(`SELECT Id`).
		WithArgs("install-provider", "ikey-xp-lock").
		WillReturnError(sql.ErrNoRows)
	expectActiveCompanionPackLock(mock, "companion-install-active")
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute.*EXISTS`).
		WithArgs(
			"id-cp-lock",
			"ns-provider",
			"install-provider",
			"webhook.xpack",
			"ikey-xp-lock",
			statusPending,
			sqlmock.AnyArg(), // payload
			connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
			"install-provider",
			"webhook.xpack",
			int64(1),
			"companion-install-active",
			"companion-flow-v1",
			sqlmock.AnyArg(), // source-settle SourceObjectKind
			sqlmock.AnyArg(), // source-settle SourceObjectID
			sqlmock.AnyArg(), // source-settle current time
			sqlmock.AnyArg(), // source-settle default deadline
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			int64(1),
			"companion-install-active",
			"companion-flow-v1",
			sqlmock.AnyArg(), // deliveryId
			sqlmock.AnyArg(), // deliverySequence
			sqlmock.AnyArg(), // enqueueOrdinal
			sqlmock.AnyArg(), // createdDate
			sqlmock.AnyArg(), // lastModifiedDate
			"install-provider",
			"webhook.xpack",
			int64(1),
			"companion-install-active",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectConnectorWorkItemMaterialization(mock, "id-cp-lock", "install-provider", "default", statusPending, "", "")
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	id, isDup, err := svc.Enqueue(
		context.Background(),
		"ns-provider", "webhook.xpack", "ikey-xp-lock", json.RawMessage(`{}`),
		"install-provider", 1,
		"companion-install-active", "companion-flow-v1",
		0, "",
	)
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}
	if isDup {
		t.Fatal("isDuplicate = true, want false")
	}
	if id != "id-cp-lock" {
		t.Fatalf("event id = %q, want id-cp-lock", id)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// HIGH3 regression — ReplayDeadEvent cross-pack: companion incarnation guard
// rejects replay when companion pack was reinstalled since original event.
// ─────────────────────────────────────────────────────────────────────────────

// Replay of a cross-pack dead event preserves the enqueue-time companion snapshot
// (CompanionPackInstallId + CompanionFlowVersionId copied verbatim) and re-enqueues
// without an enqueue-time companion-active guard. Correctness across companion
// evolution is the worker's drain-time incarnation re-check (§5.2): if the frozen
// companion incarnation is gone, the worker fails the replayed event closed. Replay
// does NOT hard-fail on an evolved companion (re-targeting a fresh companion
// incarnation is out of scope), so the replay INSERT carries no EXISTS guard and no
// companion bind arg.
func TestReplayDeadEvent_CrossPack_PreservesCompanionSnapshot(t *testing.T) {
	svc, mock := newMock(t, "id-replay-cp")

	// Original dead cross-pack event with companion snapshot.
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-xp-001").
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
			"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
			"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
			"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey", "DeliveryId",
			"DeliverySequence", "CreatedDate", "LastModifiedDate",
		}).AddRow(
			"dead-xp-001", "ns-provider", "install-provider", "webhook.xpack", "ikey-xp", statusDead,
			"default", []byte(`{}`), 3, nil, nil,
			nil, nil, nil, nil, nil, nil,
			int64(1), "companion-install-old", "companion-flow-old", nil, nil, // SourceLaneKey, DeliveryId (NULL legacy)
			0, time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
		))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "ns-provider", "install-provider", "dead-xp-001", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-xp-001")

	// CAS UPDATE reserves replay right.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	expectNextEnqueueOrdinal(mock, "install-provider", 1)

	// INSERT…SELECT carries the original companion snapshot verbatim and has NO
	// companion EXISTS guard / companion bind arg. Provider active-pack + route
	// liveness still gate the row; here they hold → RowsAffected=1 (re-enqueued).
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorEvent.*FROM _System_ConnectorInboundRoute`).
		WithArgs(
			"id-replay-cp",
			"ns-provider",
			"install-provider",
			"webhook.xpack",
			"replay:id-replay-cp",
			statusPending,
			sqlmock.AnyArg(), // payload
			sqlmock.AnyArg(), // SourceObjectKind
			sqlmock.AnyArg(), // SourceObjectID
			sqlmock.AnyArg(), // SourceLaneKey
			int64(1),
			"companion-install-old", // CompanionPackInstallId (preserved)
			"companion-flow-old",    // CompanionFlowVersionId (preserved)
			"dead-xp-001",           // ReplayOfId
			sqlmock.AnyArg(),        // deliveryId (copied from dead row; nil for legacy)
			sqlmock.AnyArg(),        // deliverySequence (copied from dead row)
			sqlmock.AnyArg(),        // enqueueOrdinal
			sqlmock.AnyArg(),        // createdDate
			sqlmock.AnyArg(),        // lastModifiedDate
			"install-provider",
			"webhook.xpack",
			int64(1),
		).
		WillReturnResult(sqlmock.NewResult(0, 1)) // re-enqueued
	expectConnectorWorkItemMaterialization(mock, "id-replay-cp", "install-provider", "default", statusPending, "", "")

	mock.ExpectCommit()
	expectPessimisticCleanup(mock)

	newID, err := svc.ReplayDeadEvent(
		context.Background(),
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dead-xp-001",
		"install-provider",
		1,
	)
	if err != nil {
		t.Fatalf("replay should re-enqueue cross-pack event preserving snapshot, got err %v", err)
	}
	if newID != "id-replay-cp" {
		t.Errorf("want replay id id-replay-cp, got %q", newID)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestReplayDeadEvent_RouteSoftDeletedBetweenResolveAndInsert is a bonus regression
// test for the fully-soft-deleted route case (IsDeleted=1), as distinct from the
// IsCurrent=0 preserved-for-backlog state above.  Both produce RowsAffected=0 from
// the JOIN in the INSERT…SELECT, which maps to ErrPackIncarnationChanged.
func TestReplayDeadEvent_RouteSoftDeletedBetweenResolveAndInsert_ReturnsTransient(t *testing.T) {
	svc, mock := newMock(t, "id-r22-deleted")

	// Dead event with RouteVersion=3.
	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("dead-evt-r22-del").
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "Namespace", "InstalledPackId", "EventKind", "IdempotencyKey", "Status",
			"QueueClass", "Payload", "RetryCount", "LastError", "LastErrorAt",
			"ClaimedBy", "ClaimedAt", "ClaimToken", "ProcessedAt", "ReplayedAt", "ReplayOfId",
			"RouteVersion", "CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey", "DeliveryId",
			"DeliverySequence", "CreatedDate", "LastModifiedDate",
		}).AddRow(
			"dead-evt-r22-del", "noop-connector", "install-r22-del", "order.placed", "ikey-r22-del", statusDead,
			"default", []byte(`{}`), 3, nil, nil,
			nil, nil, nil, nil, nil, nil,
			int64(3), nil, nil, nil, nil,
			0, time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
		))

	expectPessimisticBegin(mock)
	expectReplayPackAndEventLocks(mock, "noop-connector", "install-r22-del", "dead-evt-r22-del", statusDead, nil)
	expectLegacyReplayOwnershipProbe(mock, "dead-evt-r22-del")

	// CAS UPDATE reserves the replay right.
	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WillReturnResult(sqlmock.NewResult(0, 1))

	// INSERT…SELECT: route version 3 is fully soft-deleted (IsDeleted=1) →
	// COALESCE(r.IsDeleted,0)=0 condition fails → JOIN returns no row → RowsAffected=0.
	expectRouteBackedReplayInsert(mock, "id-r22-deleted", "noop-connector", "install-r22-del", "order.placed", "dead-evt-r22-del", 3, 0)

	mock.ExpectRollback()
	expectPessimisticCleanup(mock)

	_, err := svc.ReplayDeadEvent(
		context.Background(),
		allowPerm(string(constants.PermReplayConnectorEvents)),
		"dead-evt-r22-del",
		"install-r22-del",
		3,
	)
	if !errors.Is(err, ErrPackIncarnationChanged) {
		t.Fatalf("want ErrPackIncarnationChanged (route soft-deleted), got %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("unmet expectations: %v", err)
	}
}

// TestClaimNext_IntraBatchOrdersByDeliverySequence asserts that ClaimNext's
// inner SELECT uses ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC,
// while still keeping the NOT EXISTS guard that now also enforces same-source
// serialization across DeliveryIds.
func TestClaimNext_IntraBatchOrdersByDeliverySequence(t *testing.T) {
	svc, mock := newMock(t, "token-order-001")

	// The inner SELECT must include the NOT EXISTS guard and the ORDER BY.
	// DeliveryId="dlv-batch-001", DeliverySequence=0: first event in a delivery batch.
	mock.ExpectQuery(`NOT EXISTS.*ORDER BY e\.CreatedDate ASC, e\.DeliverySequence ASC, e\.Id ASC`).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "DeliveryId", "DeliverySequence"}).AddRow("evt-seq-0", "dlv-batch-001", 0))

	mock.ExpectExec(`UPDATE _System_ConnectorEvent`).
		WithArgs(statusProcessing, "worker-order", sqlmock.AnyArg(), "token-order-001", sqlmock.AnyArg(), "evt-seq-0", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(0, 1))

	mock.ExpectQuery(`SELECT Id, Namespace`).
		WithArgs("evt-seq-0").
		WillReturnRows(eventRow("evt-seq-0", "noop-connector", "pack-ord-001", "webhook.event", "ikey-ord", "processing", 0, ""))

	ev, claimed, err := svc.ClaimNext(context.Background(), "pack-ord-001", "worker-order")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if ev == nil || ev.ID != "evt-seq-0" {
		t.Fatalf("unexpected event: %v", ev)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Errorf("sqlmock expectations unmet (ORDER BY clause missing or wrong): %v", err)
	}
}