package connector_events

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events/claimpolicy"
	"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-connectors/internal/id"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

// MaxRetryAttempts is the retry threshold matching the outbox precedent.
const MaxRetryAttempts = 5

// StaleProcessingSeconds bounds how long a row may remain in 'processing'
// before ClaimNext is allowed to reclaim it. A worker that holds a row
// past this window is assumed dead; without reclamation the event would
// be stranded forever. Mirrors OutboxStaleProcessingSeconds.
const StaleProcessingSeconds = claimpolicy.StaleProcessingSeconds

// claimContentionRetries is how many times ClaimNext retries the UPDATE-WHERE
// on a lost-race (RowsAffected == 0) before returning (nil, false, nil).
// 32 gives comfortable headroom under heavy worker stampedes without risking
// an unbounded livelock — external poll-loop is the caller's back-off strategy.
const claimContentionRetries = 32

// Status constants.
const (
	statusPending    = "pending"
	statusProcessing = "processing"
	statusProcessed  = "processed"
	statusFailed     = "failed"
	statusSkipped    = "skipped"
	statusDead       = "dead"
)

// Sentinel errors.
var (
	ErrInactivePack     = errors.New("connector-pack inactive or missing")
	ErrPermissionDenied = errors.New("connector events: permission denied")
	ErrEventNotFound    = errors.New("connector event not found")
	ErrEventNotDead     = errors.New("connector event is not in dead state")
	ErrReplayBlocked    = errors.New("connector event replay is blocked by an active source projection")
	// ErrLegacyEventReplayRetired is returned after the attested WorkItem-only
	// cutover. Event replay is permanently unavailable in that release mode.
	ErrLegacyEventReplayRetired = errors.New("connector event replay is retired after the WorkItem-only cutover")
	ErrEventAlreadyTerminal     = errors.New("connector event already in terminal state")
	// ErrLeaseLost is returned by MarkProcessed/MarkFailed when the event's
	// lease is no longer held by the calling worker — either because ClaimNext
	// reclaimed the row from a stale worker or because a concurrent ack/fail
	// already transitioned it to a terminal state.
	ErrLeaseLost = errors.New("connector event lease no longer held by worker")
	// ErrPackIncarnationChanged is returned by Enqueue and ReplayDeadEvent when
	// the (namespace, packInstallID) pair is no longer the active incarnation at
	// the moment of INSERT.  The caller resolved packInstallID earlier in the
	// request lifecycle and the pack was reinstalled (PackInstallId rotated to a
	// new value) between that resolve and this write.  The caller should map this
	// to a retriable 503 so the provider retries; on retry the caller re-resolves
	// the new active pack and the cycle completes against the correct incarnation.
	ErrPackIncarnationChanged = errors.New("connector_events: pack incarnation changed before enqueue")
	// ErrInvalidReplayRequest is returned by ReplayDeadEvent when the caller
	// passes an invalid routeVersion argument (e.g. a negative value).
	ErrInvalidReplayRequest = errors.New("connector_events: invalid replay request")
	// ErrCompanionConsentMissing is returned by Enqueue (via the dispatcher)
	// when a cross-pack route's companion app-pack has not declared a
	// PackDependency on the routing connector, or when the companion is not
	// currently active, or when the dependency row's DependentPackInstallId
	// no longer matches the current active companion incarnation (incarnation
	// drift), or when the companion's FlowDefinition is missing. Maps to
	// retriable 503 at the dispatcher — same shape as ErrRouteNotDeclared.
	ErrCompanionConsentMissing = errors.New("connector_events: companion consent missing or incarnation drift")
	// ErrCompanionVersionUnsatisfied is returned when a cross-pack route's
	// companion manifest VersionConstraint is not satisfied by the provider
	// connector-pack's current PackVersion. Maps to retriable 503.
	ErrCompanionVersionUnsatisfied = errors.New("connector_events: companion version constraint not satisfied")
)

// PermissionChecker is satisfied by any session that can report whether the
// current user holds a named standard permission.
type PermissionChecker interface {
	Has(key string) bool
}

// ConnectorEvent is the read-model for a single _System_ConnectorEvent row.
type ConnectorEvent struct {
	ID              string
	Namespace       string
	InstalledPackID string
	EventKind       string
	IdempotencyKey  string
	Status          string
	QueueClass      string
	Payload         json.RawMessage
	RetryCount      int
	LastError       string
	LastErrorAt     *time.Time
	NextAttemptAt   *time.Time
	ClaimedBy       string
	ClaimedAt       *time.Time
	ClaimToken      string
	ProcessedAt     *time.Time
	ReplayedAt      *time.Time
	ReplayOfID      string
	// RouteVersion is the three-state snapshot of the _System_ConnectorInboundRoute
	// version resolved at enqueue time (option 3 / versioned route declarations):
	//   - Valid=false (NULL): legacy row written before RouteVersion was added.
	//     Worker should fall back to live RouteResolver.ResolveFlow lookup.
	//   - Valid=true, Int64=0: sentinel — no route was declared at enqueue time
	//     for this (InstalledPackId, EventKind). Worker must failTerminal.
	//     Version numbers start at 1 so 0 is an unambiguous sentinel.
	//   - Valid=true, Int64>=1: specific immutable route version. Worker calls
	//     RouteResolver.ResolveFlowByVersion(InstalledPackId, EventKind, Version)
	//     to get the exact FlowDeveloperName in effect when the event arrived.
	RouteVersion sql.NullInt64
	// CompanionPackInstallID is the companion app-pack PackInstallId frozen at
	// enqueue for cross-pack connector-inbound events. Empty = same-pack/legacy
	// (worker uses provider-side name-based flow resolution unchanged).
	CompanionPackInstallID string
	// CompanionFlowVersionID is the exact Flow.Id frozen at enqueue for
	// cross-pack connector-inbound events. Empty = same-pack/legacy.
	// When non-empty the worker must execute this exact version via
	// getFlowVersionById, never getActiveFlowVersion-by-name.
	CompanionFlowVersionID string
	// DeliveryId is the app-assigned per-delivery grouping id minted once by the
	// dispatcher per webhook delivery; all events in one delivery share the same value.
	// Empty = legacy singleton (pre-#2154): the NOT EXISTS guard never fires when
	// DeliveryId is NULL/empty so legacy rows drain as independent singletons.
	//
	// Warm-upgrade compatibility: events enqueued BEFORE #2154 carry no DeliveryId —
	// the grouping key never existed, so there is NO data to backfill them into delivery
	// groups (a backfill is structurally impossible, not omitted). They therefore drain
	// unordered, byte-identical to pre-#2154 behaviour — this is a no-op for the legacy
	// backlog, NOT an ordering regression. Per-delivery serialization applies only to
	// events enqueued with a DeliveryId (post-#2154); out-of-order execution among
	// ungrouped legacy events is backstopped by #2134's IsRetriable retry, exactly the
	// connector-flow retry contract that governed every event before this change.
	SourceLaneKey    string
	DeliveryId       string
	DeliverySequence int
	CreatedDate      time.Time
	LastModifiedDate time.Time
}

// Page is a simple limit/offset pagination descriptor.
type Page struct {
	Limit  int
	Offset int
}

// Service is the connector event queue.  Use NewService to construct.
type Service struct {
	db             *sql.DB
	idgen          func() string
	now            func() time.Time
	retryThreshold int
	workqueue      *connector_workqueue.Repository
	admissionGate  connector_admission.TransactionGate

	// afterEnqueueTransactionBegin is a test seam immediately after the
	// pessimistic enqueue transaction begins, which fixes the TiDB transaction
	// snapshot, and before its first authority read. Production leaves it nil;
	// integration tests use it to order a concurrent materializer commit.
	afterEnqueueTransactionBegin func()

	// afterIngressFenceLock is a test seam immediately after the namespace
	// fence is locked and before any Event or WorkItem write. Production leaves
	// it nil; integration tests use it to prove an arm waits for an already
	// admitted materialization transaction to finish.
	afterIngressFenceLock func()

	// afterClaimSelect, when non-nil, is invoked once inside ClaimNext between the
	// candidate SELECT and the claim UPDATE. Production leaves it nil; tests set it
	// to deterministically exercise the SELECT->UPDATE race (e.g. a concurrent
	// ReplayDeadEvent re-introducing an earlier-DeliverySequence sibling) that the
	// UPDATE's ordering re-check must catch.
	afterClaimSelect func()

	// afterLegacyTerminalEventLock is a test seam for the Event->WorkItem
	// terminal handoff. Production leaves it nil; integration tests use it to
	// commit an Event-first source-projection enqueue while completion holds the
	// current Event lock but has not yet locked its WorkItem.
	afterLegacyTerminalEventLock func()

	// beforeReplayEventLock and afterReplayEventLock are test seams around the
	// replay transaction's Event lock, after its active-pack lock. Production
	// leaves both nil.
	beforeReplayEventLock func()
	afterReplayEventLock  func()
}

// NewService constructs a Service backed by db.  idgen and now are injectable
// for deterministic tests; pass nil to get the production defaults
// (id.Generate and time.Now).
func NewService(db *sql.DB, idgen func() string, now func() time.Time) *Service {
	if idgen == nil {
		idgen = id.Generate
	}
	if now == nil {
		now = time.Now
	}
	return &Service{
		db:             db,
		idgen:          idgen,
		now:            now,
		retryThreshold: MaxRetryAttempts,
		workqueue:      connector_workqueue.NewRepository(db, idgen, now),
		admissionGate:  connector_admission.NewSQLGate(db),
	}
}

// requireIngressAdmission is the only Event/WorkItem materialization admission
// callsite. Both webhook enqueue and admin replay invoke it after starting the
// same pessimistic transaction that writes durable rows. There is intentionally
// no public gate-replacement hook: production wiring cannot bypass L0 admission.
func (s *Service) requireIngressAdmission(ctx context.Context, tx *sql.Tx, namespace string) error {
	if s == nil || s.admissionGate == nil {
		return &connector_admission.SchemaUnavailableError{Namespace: namespace}
	}
	_, err := s.admissionGate.RequireActiveTx(ctx, tx, namespace)
	return err
}

// WorkItemRepository returns the repository owned by this Service's atomic
// Event-to-WorkItem bridge. Callers that opt into that bridge must use this
// exact repository for WorkItem claims, so a custom claim implementation never
// gets bypassed during terminal completion.
func (s *Service) WorkItemRepository() *connector_workqueue.Repository {
	return s.workqueue
}

// resolvePack returns the active connector-pack row for namespace.
// Returns ErrInactivePack (not a DB error) when the pack is missing/inactive.
func (s *Service) resolvePack(ctx context.Context, namespace string) (*packstate.ConnectorPackRow, error) {
	row, ok, err := packstate.ResolveConnectorPackByNamespace(ctx, s.db, namespace)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, ErrInactivePack
	}
	return row, nil
}

// ResolveActivePack returns the InstallID (PackInstallId) for the active
// connector-pack row for namespace.  This is the per-incarnation identity used
// to scope event queue rows and route lookups to a specific pack install.
//
// The Dispatcher calls this ONCE per inbound request and passes the returned
// packInstallID into both RouteResolver.ResolveFlow and Service.Enqueue, so
// both calls are bound to the same pack incarnation.  This eliminates the
// install-transition race where a route resolved for pack-A could be enqueued
// against pack-B.
//
// Returns ErrInactivePack when no active pack exists for namespace.
func (s *Service) ResolveActivePack(ctx context.Context, namespace string) (packInstallID string, err error) {
	row, err := s.resolvePack(ctx, namespace)
	if err != nil {
		return "", err
	}
	return row.InstallID, nil
}

// BatchEvent holds the per-event inputs for EnqueueBatch.
type BatchEvent struct {
	EventKind      string
	IdempotencyKey string
	Payload        json.RawMessage
	RouteVersion   int64
	// DeliveryID optionally overrides the EnqueueBatch deliveryID argument for
	// this event. Existing callers keep passing the batch-level deliveryID;
	// receipt-router batches use this to enqueue multiple deliveries in one tx.
	DeliveryID string
	// QueueClass is honored only for legacy/unrouted rows (RouteVersion < 1).
	// Route-backed rows snapshot QueueClass from _System_ConnectorInboundRoute.
	QueueClass             string
	CompanionPackInstallID string
	CompanionFlowVersionID string
	// SourceLaneObjectKind/SourceLaneID optionally override the materialized
	// SourceLaneKey used by ClaimNext ordering. Payload still carries the source
	// object envelope used by the worker for BatchRead.
	SourceLaneObjectKind string
	SourceLaneID         string
	// TargetMutationLaneObjectKind/TargetMutationLaneID optionally populate the
	// WorkItem target lane when a caller already has first-class target metadata.
	// Current legacy compatibility still uses SourceLaneKey for event claiming.
	TargetMutationLaneObjectKind string
	TargetMutationLaneID         string
	DeliverySequence             int
}

// EnqueueBatch atomically enqueues all events in a single transaction.
// If the whole batch succeeds, ids[i] is the event ID (new or existing-dedup)
// and dups[i] is true when the event already existed.
//
// The lock on the active pack incarnation is acquired ONCE, before the loop,
// and held across all inserts (FOR UPDATE row-lock).  This eliminates the P1
// race: a retired-pack caller cannot get stale duplicates via the pre-tx
// fast-path (dedup runs inside the tx, after the lock).  It also makes the
// whole delivery all-or-nothing: if any event's INSERT returns 0 RowsAffected
// (pack rotated or route swept mid-batch), the transaction rolls back and
// ErrPackIncarnationChanged is returned — the dispatcher returns 503 and the
// provider retries the whole delivery, maintaining strict DeliverySequence
// uniqueness within a DeliveryId.
//
// Dedup (IdempotencyKey) is handled inside the transaction so a tx sees its
// own uncommitted inserts — intra-batch duplicate keys are caught without
// needing a unique-constraint slow path.
func (s *Service) EnqueueBatch(ctx context.Context, namespace, packInstallID, deliveryID string, events []BatchEvent) (ids []string, dups []bool, err error) {
	if strings.TrimSpace(packInstallID) == "" {
		return nil, nil, errors.New("connector-events: Enqueue requires a non-empty packInstallID")
	}
	if len(events) == 0 {
		return nil, nil, nil
	}
	enqueueStarted := time.Now()
	enqueuePath := "per_event"
	recordConnectorEventEnqueueBatchRows("event_fact", len(events))
	defer func() {
		outcome := "ok"
		if err != nil {
			outcome = "error"
		}
		recordConnectorEventEnqueueStageDuration("total_"+enqueuePath, outcome, time.Since(enqueueStarted).Seconds())
	}()

	defaultDeliveryID := strings.TrimSpace(deliveryID)

	// Admission still serializes on namespace/pack rows. A waiter must also
	// see immutable Events selected by the previous materializer after this
	// transaction began; WorkItem coverage cannot be decided from an old snapshot.
	tx, cleanup, txErr := sqltx.BeginPessimisticReadCommittedTx(ctx, s.db)
	if txErr != nil {
		return nil, nil, txErr
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()
	if s.afterEnqueueTransactionBegin != nil {
		s.afterEnqueueTransactionBegin()
	}

	// This is the admission authority. The dispatcher may have accepted a fast
	// SQL snapshot before this transaction began, but only this lock survives a
	// concurrent arm and covers the Event + WorkItem writes below.
	if gateErr := s.requireIngressAdmission(ctx, tx, namespace); gateErr != nil {
		return nil, nil, gateErr
	}
	if s.afterIngressFenceLock != nil {
		s.afterIngressFenceLock()
	}

	// Acquire the lock BEFORE any dedup or insert — this is the P1 fix.
	// A retired-pack caller gets ErrPackIncarnationChanged here, not a stale duplicate.
	if lockErr := lockActivePackIncarnation(ctx, tx, namespace, packInstallID); lockErr != nil {
		return nil, nil, lockErr
	}

	ids = make([]string, len(events))
	dups = make([]bool, len(events))
	now := s.now()
	touchedLanes := map[string]connector_workqueue.CreateWorkItemInput{}
	deferBatchWorkItemReconcile := len(events) > 1
	routeEnqueueSpecs := map[routeEnqueueSpecKey]routeEnqueueSpec{}
	settleDeadlines := map[sourceSettleKey]sql.NullTime{}
	nextEnqueueOrdinal, err := s.nextEnqueueOrdinal(ctx, tx, packInstallID)
	if err != nil {
		return nil, nil, err
	}
	if fastIDs, fastDups, fastPath, fastErr := s.enqueueFreshRouteBatchFastPath(ctx, tx, namespace, packInstallID, defaultDeliveryID, events, nextEnqueueOrdinal, now); fastPath || fastErr != nil {
		if fastPath {
			enqueuePath = "fresh_batch"
		}
		if fastErr != nil {
			return nil, nil, fastErr
		}
		if commitErr := tx.Commit(); commitErr != nil {
			return nil, nil, commitErr
		}
		committed = true
		return fastIDs, fastDups, nil
	}

	for i, e := range events {
		eventKind := strings.ToLower(strings.TrimSpace(e.EventKind))
		targetMutationLaneKey, laneErr := targetMutationLaneKeyFromBatchEvent(e)
		if laneErr != nil {
			return nil, nil, laneErr
		}

		// Dedup INSIDE the tx so we see our own uncommitted inserts and the lock.
		if existingID, found, lookupErr := s.lookupExisting(ctx, tx, packInstallID, e.IdempotencyKey); lookupErr != nil {
			return nil, nil, lookupErr
		} else if found {
			touched, err := s.ensureWorkItemForEvent(ctx, tx, existingID, targetMutationLaneKey, now, true, false)
			if err != nil {
				return nil, nil, err
			}
			rememberTouchedWorkItemLane(touchedLanes, touched)
			ids[i] = existingID
			dups[i] = true
			continue
		}

		companionPack := strings.TrimSpace(e.CompanionPackInstallID)
		companionFlow := strings.TrimSpace(e.CompanionFlowVersionID)
		if companionPack != "" {
			if lockErr := lockActiveCompanionPackIncarnation(ctx, tx, companionPack); lockErr != nil {
				return nil, nil, lockErr
			}
		}

		newID := s.idgen()
		enqueueOrdinal := nextEnqueueOrdinal
		nextEnqueueOrdinal++

		var companionPackArg interface{}
		var companionFlowArg interface{}
		if companionPack != "" {
			companionPackArg = companionPack
		}
		if companionFlow != "" {
			companionFlowArg = companionFlow
		}
		eventDeliveryID := strings.TrimSpace(e.DeliveryID)
		if eventDeliveryID == "" {
			eventDeliveryID = defaultDeliveryID
		}
		var deliveryIDArg interface{}
		if eventDeliveryID != "" {
			deliveryIDArg = eventDeliveryID
		}
		lane := sourceLaneFromPayload(e.Payload)
		laneKey := lane.Key
		overrideLaneObjectKind := strings.TrimSpace(e.SourceLaneObjectKind)
		overrideLaneID := e.SourceLaneID
		hasOverrideLaneID := strings.TrimSpace(overrideLaneID) != ""
		if overrideLaneObjectKind != "" || hasOverrideLaneID {
			if overrideLaneObjectKind == "" || !hasOverrideLaneID {
				return nil, nil, errors.New("connector-events: SourceLaneObjectKind and SourceLaneID must be configured together")
			}
			laneKey = sourceLaneKeyFor(overrideLaneObjectKind, overrideLaneID)
		}
		var sourceObjectKindArg interface{}
		var sourceObjectIDArg interface{}
		var sourceLaneKeyArg interface{}
		if lane.ObjectKind != "" {
			sourceObjectKindArg = lane.ObjectKind
		}
		if lane.SourceID != "" {
			sourceObjectIDArg = lane.SourceID
		}
		if laneKey != "" {
			sourceLaneKeyArg = laneKey
		}

		var result sql.Result
		var insertErr error
		var queueClass string
		var nextAttemptAt sql.NullTime
		var sourceProjection bool
		if e.RouteVersion >= 1 {
			if deferBatchWorkItemReconcile {
				enqueueSpec, specErr := lookupRouteEnqueueSpec(ctx, tx, packInstallID, eventKind, e.RouteVersion, routeEnqueueSpecs)
				if specErr != nil {
					return nil, nil, specErr
				}
				queueClass = enqueueSpec.queueClass
				sourceProjection = enqueueSpec.sourceProjection
				if enqueueSpec.sourceSettleDelay > 0 {
					nextAttemptAt, specErr = s.sourceSettleDeadline(ctx, tx, sourceSettleKey{
						installedPackID: packInstallID,
						eventKind:       eventKind,
						routeVersion:    e.RouteVersion,
						companionPack:   companionPack,
						companionFlow:   companionFlow,
						sourceKind:      lane.ObjectKind,
						sourceID:        lane.SourceID,
					}, now, enqueueSpec.sourceSettleDelay, settleDeadlines)
					if specErr != nil {
						return nil, nil, specErr
					}
				}
			}
			routeNotBeforeSQL := `CASE WHEN LOWER(TRIM(COALESCE(r.InvocationMode, ''))) = ? THEN ? ELSE NULL END`
			routeNotBeforeArgs := []interface{}{
				connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
				nullableTime(nextAttemptAt),
			}
			if !deferBatchWorkItemReconcile {
				routeNotBeforeSQL = `CASE WHEN LOWER(TRIM(COALESCE(r.InvocationMode, ''))) = ? THEN COALESCE((
SELECT MIN(existing.NextAttemptAt)
  FROM _System_ConnectorEvent existing
 WHERE existing.InstalledPackId = ?
   AND existing.EventKind = ?
   AND COALESCE(existing.RouteVersion, 0) = ?
   AND COALESCE(existing.CompanionPackInstallId, '') = ?
   AND COALESCE(existing.CompanionFlowVersionId, '') = ?
   AND existing.SourceObjectKind = ?
   AND existing.SourceObjectID = ?
   AND existing.Status = 'pending'
   AND existing.RetryCount = 0
   AND existing.NextAttemptAt > ?
), ?) ELSE NULL END`
				routeNotBeforeArgs = []interface{}{
					connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
					packInstallID, eventKind, e.RouteVersion, companionPack, companionFlow,
					lane.ObjectKind, lane.SourceID, now, now.Add(connector_inbound_routing.DefaultSourceSettleDelay).UTC(),
				}
			}
			routeQuery := `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, Payload, RetryCount,
   NextAttemptAt, SourceObjectKind, SourceObjectID, SourceLaneKey, RouteVersion, CompanionPackInstallId, CompanionFlowVersionId, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
SELECT ?, ?, ?, ?, ?, ?, COALESCE(NULLIF(r.QueueClass, ''), 'default'), ?, 0,
       ` + routeNotBeforeSQL + `,
       ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
  FROM _System_ConnectorInboundRoute r
 WHERE r.InstalledPackId = ?
   AND r.EventKind       = ?
   AND r.Version         = ?
   AND COALESCE(r.IsDeleted, 0) = 0
   AND COALESCE(r.IsCurrent, 0) = 1`
			routeArgs := []interface{}{
				newID, namespace, packInstallID, eventKind, e.IdempotencyKey, statusPending,
				nullableJSON(e.Payload),
			}
			routeArgs = append(routeArgs, routeNotBeforeArgs...)
			routeArgs = append(routeArgs,
				sourceObjectKindArg, sourceObjectIDArg, sourceLaneKeyArg,
				e.RouteVersion, companionPackArg, companionFlowArg, deliveryIDArg, e.DeliverySequence, enqueueOrdinal, now, now,
				packInstallID, eventKind, e.RouteVersion,
			)
			if companionPack != "" {
				routeQuery += `
   AND EXISTS (SELECT 1 FROM _System_InstalledPack ip
                WHERE ip.PackInstallId = ?
                  AND ip.Status = 'active'
                  AND COALESCE(ip.IsDeleted, 0) = 0)`
				routeArgs = append(routeArgs, companionPack)
			}
			routeQuery += "\nLIMIT 1"
			result, insertErr = tx.ExecContext(ctx, routeQuery, routeArgs...)
		} else {
			var qErr error
			queueClass, qErr = connector_inbound_routing.NormalizeQueueClass(e.QueueClass)
			if qErr != nil {
				return nil, nil, qErr
			}
			result, insertErr = tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, Payload, RetryCount,
   NextAttemptAt, SourceObjectKind, SourceObjectID, SourceLaneKey, RouteVersion, CompanionPackInstallId, CompanionFlowVersionId, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
				newID, namespace, packInstallID, eventKind, e.IdempotencyKey, statusPending,
				queueClass, nullableJSON(e.Payload), nullableTime(nextAttemptAt), sourceObjectKindArg, sourceObjectIDArg, sourceLaneKeyArg,
				e.RouteVersion, companionPackArg, companionFlowArg, deliveryIDArg, e.DeliverySequence, enqueueOrdinal, now, now)
		}
		if insertErr != nil {
			return nil, nil, insertErr
		}
		n, raErr := result.RowsAffected()
		if raErr != nil {
			return nil, nil, raErr
		}
		if n == 0 {
			// Pack rotated or route swept mid-batch: roll back the whole delivery.
			return nil, nil, ErrPackIncarnationChanged
		}
		ids[i] = newID
		dups[i] = false
		var touched *connector_workqueue.CreateWorkItemInput
		if deferBatchWorkItemReconcile {
			touched, err = s.ensureWorkItemForFreshEvent(ctx, tx, freshWorkItemEvent{
				eventID:               newID,
				installedPackID:       packInstallID,
				eventKind:             eventKind,
				routeVersion:          e.RouteVersion,
				companionPack:         companionPack,
				companionFlow:         companionFlow,
				queueClass:            queueClass,
				sourceLaneKey:         laneKey,
				targetMutationLaneKey: targetMutationLaneKey,
				laneOrdinal:           enqueueOrdinal,
				nextAttemptAt:         nextAttemptAt,
				sourceProjection:      sourceProjection,
			}, now, true)
		} else {
			touched, err = s.ensureWorkItemForEvent(ctx, tx, newID, targetMutationLaneKey, now, false, false)
		}
		if err != nil {
			return nil, nil, err
		}
		rememberTouchedWorkItemLane(touchedLanes, touched)
	}

	if len(events) > 1 && s.workqueue != nil {
		workqueue := s.workqueue.WithTx(tx)
		for _, touched := range touchedLanes {
			if err := workqueue.ReconcileLaneReadyState(ctx, touched); err != nil {
				return nil, nil, err
			}
		}
	}

	if commitErr := tx.Commit(); commitErr != nil {
		return nil, nil, commitErr
	}
	committed = true
	return ids, dups, nil
}

type freshRouteBatchEvent struct {
	eventID               string
	eventKind             string
	idempotencyKey        string
	queueClass            string
	payload               json.RawMessage
	sourceObjectKindArg   interface{}
	sourceObjectIDArg     interface{}
	sourceLaneKeyArg      interface{}
	sourceObjectKind      string
	sourceObjectID        string
	sourceLaneKey         string
	targetMutationLaneKey string
	routeVersion          int64
	companionPack         string
	companionFlow         string
	companionPackArg      interface{}
	companionFlowArg      interface{}
	deliveryIDArg         interface{}
	deliverySequence      int
	enqueueOrdinal        int64
	inputStartOrdinal     int64
	workID                string
	workType              string
	workState             string
	runAfter              time.Time
	nextAttemptAt         sql.NullTime
	sourceProjection      bool
}

func (s *Service) enqueueFreshRouteBatchFastPath(
	ctx context.Context,
	tx *sql.Tx,
	namespace, packInstallID, defaultDeliveryID string,
	events []BatchEvent,
	nextEnqueueOrdinal int64,
	now time.Time,
) ([]string, []bool, bool, error) {
	if s.workqueue == nil || len(events) <= 1 {
		return nil, nil, false, nil
	}
	stageStarted := time.Now()
	seenKeys := make(map[string]struct{}, len(events))
	companionPackInstallIDs := make(map[string]struct{})
	prepared := make([]freshRouteBatchEvent, 0, len(events))
	ids := make([]string, len(events))
	dups := make([]bool, len(events))

	for i, e := range events {
		if _, ok := seenKeys[e.IdempotencyKey]; ok {
			return nil, nil, false, nil
		}
		seenKeys[e.IdempotencyKey] = struct{}{}
		targetMutationLaneKey, laneErr := targetMutationLaneKeyFromBatchEvent(e)
		if laneErr != nil {
			return nil, nil, true, laneErr
		}
		eventKind := strings.ToLower(strings.TrimSpace(e.EventKind))
		lane := sourceLaneFromPayload(e.Payload)
		laneKey := lane.Key
		overrideLaneObjectKind := strings.TrimSpace(e.SourceLaneObjectKind)
		overrideLaneID := e.SourceLaneID
		hasOverrideLaneID := strings.TrimSpace(overrideLaneID) != ""
		if overrideLaneObjectKind != "" || hasOverrideLaneID {
			if overrideLaneObjectKind == "" || !hasOverrideLaneID {
				return nil, nil, true, errors.New("connector-events: SourceLaneObjectKind and SourceLaneID must be configured together")
			}
			laneKey = sourceLaneKeyFor(overrideLaneObjectKind, overrideLaneID)
		}
		var sourceObjectKindArg interface{}
		var sourceObjectIDArg interface{}
		var sourceLaneKeyArg interface{}
		if lane.ObjectKind != "" {
			sourceObjectKindArg = lane.ObjectKind
		}
		if lane.SourceID != "" {
			sourceObjectIDArg = lane.SourceID
		}
		if laneKey != "" {
			sourceLaneKeyArg = laneKey
		}
		eventDeliveryID := strings.TrimSpace(e.DeliveryID)
		if eventDeliveryID == "" {
			eventDeliveryID = defaultDeliveryID
		}
		var deliveryIDArg interface{}
		if eventDeliveryID != "" {
			deliveryIDArg = eventDeliveryID
		}
		companionPack := strings.TrimSpace(e.CompanionPackInstallID)
		companionFlow := strings.TrimSpace(e.CompanionFlowVersionID)
		var companionPackArg interface{}
		var companionFlowArg interface{}
		if companionPack != "" {
			companionPackArg = companionPack
			companionPackInstallIDs[companionPack] = struct{}{}
		}
		if companionFlow != "" {
			companionFlowArg = companionFlow
		}
		state, runAfter, ok := workItemInitialState(statusPending, sql.NullTime{}, now)
		if !ok {
			return nil, nil, false, nil
		}
		workType := workTypeForEventKind(eventKind)
		eventID := s.idgen()
		enqueueOrdinal := nextEnqueueOrdinal + int64(i)
		workID := connectorEventWorkIDForWorkType(workType, eventID)
		ids[i] = eventID
		prepared = append(prepared, freshRouteBatchEvent{
			eventID:               eventID,
			eventKind:             eventKind,
			idempotencyKey:        e.IdempotencyKey,
			payload:               e.Payload,
			sourceObjectKindArg:   sourceObjectKindArg,
			sourceObjectIDArg:     sourceObjectIDArg,
			sourceLaneKeyArg:      sourceLaneKeyArg,
			sourceObjectKind:      lane.ObjectKind,
			sourceObjectID:        lane.SourceID,
			sourceLaneKey:         laneKey,
			targetMutationLaneKey: targetMutationLaneKey,
			routeVersion:          e.RouteVersion,
			companionPack:         companionPack,
			companionFlow:         companionFlow,
			companionPackArg:      companionPackArg,
			companionFlowArg:      companionFlowArg,
			deliveryIDArg:         deliveryIDArg,
			deliverySequence:      e.DeliverySequence,
			enqueueOrdinal:        enqueueOrdinal,
			workID:                workID,
			workType:              workType,
			workState:             state,
			runAfter:              runAfter,
		})
	}
	// Existing events remain acknowledgeable after their route or companion
	// incarnation is retired; only fresh rows require current-incarnation locks.
	existing, err := lookupExistingBatch(ctx, tx, packInstallID, seenKeys)
	if err != nil {
		return nil, nil, true, err
	}
	if len(existing) > 0 {
		return nil, nil, false, nil
	}
	if err := lockActiveCompanionPackIncarnations(ctx, tx, companionPackInstallIDs); err != nil {
		return nil, nil, true, err
	}
	routeEnqueueSpecs := map[routeEnqueueSpecKey]routeEnqueueSpec{}
	settleDeadlines := map[sourceSettleKey]sql.NullTime{}
	for i, e := range events {
		var queueClass string
		var err error
		if e.RouteVersion >= 1 {
			enqueueSpec, specErr := lookupRouteEnqueueSpec(ctx, tx, packInstallID, prepared[i].eventKind, e.RouteVersion, routeEnqueueSpecs)
			err = specErr
			queueClass = enqueueSpec.queueClass
			if enqueueSpec.sourceProjection && prepared[i].sourceLaneKey != "" {
				prepared[i].sourceProjection = true
				if err := validateSourceProjectionTargetMutationLane(freshWorkItemEvent{
					sourceLaneKey:         prepared[i].sourceLaneKey,
					targetMutationLaneKey: prepared[i].targetMutationLaneKey,
					sourceProjection:      true,
				}); err != nil {
					return nil, nil, true, err
				}
				prepared[i].workID = sourceProjectionWorkID(
					packInstallID,
					prepared[i].eventKind,
					prepared[i].routeVersion,
					prepared[i].companionPack,
					prepared[i].companionFlow,
					prepared[i].workType,
					prepared[i].sourceLaneKey,
				)
			}
			if enqueueSpec.sourceSettleDelay > 0 {
				key := normalizeSourceSettleKey(sourceSettleKey{
					installedPackID: packInstallID,
					eventKind:       prepared[i].eventKind,
					routeVersion:    e.RouteVersion,
					companionPack:   strings.TrimSpace(e.CompanionPackInstallID),
					companionFlow:   strings.TrimSpace(e.CompanionFlowVersionID),
					sourceKind:      prepared[i].sourceObjectKind,
					sourceID:        prepared[i].sourceObjectID,
				})
				prepared[i].nextAttemptAt = sql.NullTime{Time: now.Add(enqueueSpec.sourceSettleDelay).UTC(), Valid: true}
				if key.sourceKind != "" && key.sourceID != "" {
					settleDeadlines[key] = prepared[i].nextAttemptAt
				}
			}
		} else {
			queueClass, err = connector_inbound_routing.NormalizeQueueClass(e.QueueClass)
		}
		if err != nil {
			return nil, nil, true, err
		}
		prepared[i].queueClass = queueClass
	}
	if err := loadSourceSettleDeadlines(ctx, tx, now, settleDeadlines); err != nil {
		return nil, nil, true, err
	}
	for i := range prepared {
		row := &prepared[i]
		if !row.nextAttemptAt.Valid {
			continue
		}
		key := normalizeSourceSettleKey(sourceSettleKey{packInstallID, row.eventKind, row.routeVersion, row.companionPack, row.companionFlow, row.sourceObjectKind, row.sourceObjectID})
		if deadline, ok := settleDeadlines[key]; ok {
			row.nextAttemptAt = deadline
		}
		_, row.runAfter, _ = workItemInitialState(statusPending, row.nextAttemptAt, now)
	}
	recordConnectorEventEnqueueStageDuration("prepare_and_settle", "ok", time.Since(stageStarted).Seconds())
	stageStarted = time.Now()
	assignFreshRouteBatchWorkStates(prepared)
	if err := bulkInsertFreshRouteEvents(ctx, tx, namespace, packInstallID, prepared, now.UTC()); err != nil {
		return nil, nil, true, err
	}
	ordinaryWorkItems, projectionHeads := partitionFreshRouteBatchWorkItems(prepared)
	recordConnectorEventEnqueueBatchRows("ordinary_work_item", len(ordinaryWorkItems))
	recordConnectorEventEnqueueBatchRows("source_projection_work_item", len(projectionHeads))
	if err := bulkInsertFreshRouteWorkItems(ctx, tx, packInstallID, ordinaryWorkItems, now.UTC()); err != nil {
		return nil, nil, true, err
	}
	touchedLanes := freshRouteBatchTouchedWorkItemLanes(packInstallID, ordinaryWorkItems, now.UTC())
	recordConnectorEventEnqueueStageDuration("bulk_facts", "ok", time.Since(stageStarted).Seconds())
	stageStarted = time.Now()
	projectionInputs := make([]connector_workqueue.SourceProjectionWorkItemInput, 0, len(projectionHeads))
	for _, row := range projectionHeads {
		state, runAfter, _ := workItemInitialState(statusPending, row.nextAttemptAt, now)
		in := connector_workqueue.CreateWorkItemInput{
			WorkID: row.workID, EventID: row.eventID, InstalledPackID: packInstallID,
			WorkType: row.workType, QueueClass: row.queueClass, State: state, RunAfter: runAfter,
			SourceLaneKey: row.sourceLaneKey, TargetMutationLaneKey: row.targetMutationLaneKey,
			LaneOrdinal: row.enqueueOrdinal, Generation: 0, Now: now.UTC(),
		}
		projectionInputs = append(projectionInputs, connector_workqueue.SourceProjectionWorkItemInput{WorkItem: in, InputStartOrdinal: row.inputStartOrdinal})
		rememberTouchedWorkItemLane(touchedLanes, &in)
	}
	if err := s.workqueue.WithTx(tx).AdmitSourceProjectionWorkItems(ctx, projectionInputs, true); err != nil {
		return nil, nil, true, err
	}
	recordConnectorEventEnqueueStageDuration("projections", "ok", time.Since(stageStarted).Seconds())
	stageStarted = time.Now()
	workqueue := s.workqueue.WithTx(tx)
	lanes := make([]connector_workqueue.CreateWorkItemInput, 0, len(touchedLanes))
	for _, touched := range touchedLanes {
		lanes = append(lanes, touched)
	}
	if err := workqueue.ReconcileLaneReadyStates(ctx, lanes); err != nil {
		return nil, nil, true, err
	}
	recordConnectorEventEnqueueStageDuration("lanes", "ok", time.Since(stageStarted).Seconds())
	return ids, dups, true, nil
}

// partitionFreshRouteBatchWorkItems keeps ordinary command WorkItems one per
// Event while reducing a state projection to the newest Event per durable
// projection identity. All Event facts are inserted before this partition is
// applied; only executable queue writes are collapsed.
func partitionFreshRouteBatchWorkItems(rows []freshRouteBatchEvent) (ordinary, projectionHeads []freshRouteBatchEvent) {
	projectionHeadIndex := make(map[string]int)
	for _, row := range rows {
		if !row.sourceProjection {
			ordinary = append(ordinary, row)
			continue
		}
		if idx, ok := projectionHeadIndex[row.workID]; ok {
			row.inputStartOrdinal = projectionHeads[idx].inputStartOrdinal
			projectionHeads[idx] = row
			continue
		}
		row.inputStartOrdinal = row.enqueueOrdinal
		projectionHeadIndex[row.workID] = len(projectionHeads)
		projectionHeads = append(projectionHeads, row)
	}
	// Multiple receipt batches may overlap the same sources. Stable lock order
	// prevents two materializers from acquiring projection WorkItems in opposite
	// order while preserving each projection's newest EnqueueOrdinal.
	sort.Slice(projectionHeads, func(i, j int) bool {
		return projectionHeads[i].workID < projectionHeads[j].workID
	})
	return ordinary, projectionHeads
}

func assignFreshRouteBatchWorkStates(rows []freshRouteBatchEvent) {
	headAssigned := make(map[string]bool)
	for i := range rows {
		lane := freshRouteWorkItemLaneKey(rows[i].sourceLaneKey, rows[i].targetMutationLaneKey)
		if lane == "" {
			rows[i].workState = connector_workqueue.WorkStatePending
		} else if headAssigned[lane] {
			rows[i].workState = connector_workqueue.WorkStateBlocked
		} else {
			rows[i].workState = connector_workqueue.WorkStatePending
			headAssigned[lane] = true
		}
	}
}

func freshRouteWorkItemLaneKey(sourceLaneKey, targetMutationLaneKey string) string {
	if lane := strings.TrimSpace(targetMutationLaneKey); lane != "" {
		return lane
	}
	return strings.TrimSpace(sourceLaneKey)
}

func freshRouteBatchTouchedWorkItemLanes(packInstallID string, rows []freshRouteBatchEvent, now time.Time) map[string]connector_workqueue.CreateWorkItemInput {
	touched := map[string]connector_workqueue.CreateWorkItemInput{}
	for _, row := range rows {
		in := connector_workqueue.CreateWorkItemInput{
			WorkID:                row.workID,
			EventID:               row.eventID,
			InstalledPackID:       packInstallID,
			WorkType:              row.workType,
			QueueClass:            row.queueClass,
			State:                 row.workState,
			RunAfter:              row.runAfter,
			SourceLaneKey:         row.sourceLaneKey,
			TargetMutationLaneKey: row.targetMutationLaneKey,
			LaneOrdinal:           row.enqueueOrdinal,
			Generation:            0,
			Now:                   now,
		}
		rememberTouchedWorkItemLane(touched, &in)
	}
	return touched
}

func lookupExistingBatch(ctx context.Context, tx *sql.Tx, packInstallID string, keys map[string]struct{}) (map[string]string, error) {
	if len(keys) == 0 {
		return nil, nil
	}
	args := make([]interface{}, 0, len(keys)+1)
	args = append(args, packInstallID)
	placeholders := make([]string, 0, len(keys))
	for key := range keys {
		placeholders = append(placeholders, "?")
		args = append(args, key)
	}
	rows, err := tx.QueryContext(ctx, `
SELECT IdempotencyKey, Id
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND IdempotencyKey IN (`+strings.Join(placeholders, ",")+`)`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]string{}
	for rows.Next() {
		var key, id string
		if err := rows.Scan(&key, &id); err != nil {
			return nil, err
		}
		out[key] = id
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func bulkInsertFreshRouteEvents(ctx context.Context, tx *sql.Tx, namespace, packInstallID string, rows []freshRouteBatchEvent, now time.Time) error {
	return forEachFreshRouteBatchChunk(rows, 1000, func(chunk []freshRouteBatchEvent) error {
		args := make([]interface{}, 0, len(chunk)*21)
		placeholders := make([]string, 0, len(chunk))
		for _, row := range chunk {
			placeholders = append(placeholders, "(?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
			args = append(args,
				row.eventID, namespace, packInstallID, row.eventKind, row.idempotencyKey, statusPending,
				row.queueClass, nullableJSON(row.payload), nullableTime(row.nextAttemptAt), row.sourceObjectKindArg, row.sourceObjectIDArg, row.sourceLaneKeyArg,
				row.routeVersion, row.companionPackArg, row.companionFlowArg, row.deliveryIDArg, row.deliverySequence, row.enqueueOrdinal, now, now,
			)
		}
		res, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, Payload, RetryCount,
   NextAttemptAt, SourceObjectKind, SourceObjectID, SourceLaneKey, RouteVersion, CompanionPackInstallId, CompanionFlowVersionId, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
VALUES `+strings.Join(placeholders, ","), args...)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n != int64(len(chunk)) {
			return ErrPackIncarnationChanged
		}
		return nil
	})
}

func bulkInsertFreshRouteWorkItems(ctx context.Context, tx *sql.Tx, packInstallID string, rows []freshRouteBatchEvent, now time.Time) error {
	return forEachFreshRouteBatchChunk(rows, 1000, func(chunk []freshRouteBatchEvent) error {
		args := make([]interface{}, 0, len(chunk)*14)
		placeholders := make([]string, 0, len(chunk))
		for _, row := range chunk {
			placeholders = append(placeholders, "(?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, 0, ?, ?)")
			args = append(args,
				row.workID, row.eventID, packInstallID, row.workType, row.queueClass, row.workState, row.runAfter,
				row.sourceLaneKey, row.targetMutationLaneKey, row.enqueueOrdinal, now, now,
			)
		}
		res, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
        Attempt, SourceLaneKey, TargetMutationLaneKey, LaneOrdinal, Generation, CreatedDate, LastModifiedDate)
VALUES `+strings.Join(placeholders, ","), args...)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n != int64(len(chunk)) {
			return ErrPackIncarnationChanged
		}
		return nil
	})
}

func forEachFreshRouteBatchChunk(rows []freshRouteBatchEvent, size int, fn func([]freshRouteBatchEvent) error) error {
	if size <= 0 {
		size = len(rows)
	}
	for len(rows) > 0 {
		n := size
		if n > len(rows) {
			n = len(rows)
		}
		if err := fn(rows[:n]); err != nil {
			return err
		}
		rows = rows[n:]
	}
	return nil
}

// Enqueue inserts a new connector event row for namespace.  If a row with the
// same (InstalledPackId, IdempotencyKey) already exists the call returns the
// existing eventID with isDuplicate=true and no error — idempotent delivery
// from an upstream dispatcher is safe.
//
// packInstallID is the per-incarnation identity of the active connector pack,
// obtained by the caller via ResolveActivePack.  The caller MUST resolve the
// active pack once and pass the same packInstallID into both
// RouteResolver.ResolveFlow and this Enqueue call so both operations are bound
// to the same pack incarnation.  Passing an empty packInstallID is an error.
//
// Enqueue is a thin wrapper around EnqueueBatch (one-element batch).  The
// pre-tx dedup fast-path that formerly ran on s.db is now replaced by a
// within-tx dedup that runs AFTER lockActivePackIncarnation — this is the P1
// fix: a retired-pack caller gets ErrPackIncarnationChanged rather than a stale
// cross-incarnation duplicate.
//
// The INSERT uses an INSERT…SELECT pattern with an active-pack guard and
// (when routeVersion>=1) a route-liveness JOIN — see EnqueueBatch for full
// documentation of those guards.
func (s *Service) Enqueue(
	ctx context.Context,
	namespace, eventKind, idempotencyKey string,
	payload json.RawMessage,
	packInstallID string,
	routeVersion int64,
	companionPackInstallID, companionFlowVersionID string,
	deliverySequence int,
	deliveryID string,
) (eventID string, isDuplicate bool, err error) {
	ids, dupFlags, err := s.EnqueueBatch(ctx, namespace, packInstallID, deliveryID, []BatchEvent{
		{
			EventKind:              eventKind,
			IdempotencyKey:         idempotencyKey,
			Payload:                payload,
			RouteVersion:           routeVersion,
			CompanionPackInstallID: companionPackInstallID,
			CompanionFlowVersionID: companionFlowVersionID,
			DeliverySequence:       deliverySequence,
		},
	})
	if err != nil {
		return "", false, err
	}
	return ids[0], dupFlags[0], nil
}

func lockActivePackIncarnation(ctx context.Context, tx *sql.Tx, namespace, packInstallID string) error {
	var one int
	err := tx.QueryRowContext(ctx, `
SELECT 1
  FROM _System_InstalledPack
 WHERE Name = ?
   AND PackInstallId = ?
   AND Status = 'active'
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1
 FOR UPDATE`, namespace, packInstallID).Scan(&one)
	if err == sql.ErrNoRows {
		return ErrPackIncarnationChanged
	}
	if err != nil {
		return err
	}
	return nil
}

func lockActiveCompanionPackIncarnation(ctx context.Context, tx *sql.Tx, packInstallID string) error {
	var one int
	err := tx.QueryRowContext(ctx, `
SELECT 1
  FROM _System_InstalledPack
 WHERE PackInstallId = ?
   AND Category = ?
   AND Status = 'active'
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1
 FOR UPDATE`, packInstallID, string(packstate.AppPack)).Scan(&one)
	if err == sql.ErrNoRows {
		return ErrPackIncarnationChanged
	}
	if err != nil {
		return err
	}
	return nil
}

func lockActiveCompanionPackIncarnations(ctx context.Context, tx *sql.Tx, packInstallIDs map[string]struct{}) error {
	ordered := make([]string, 0, len(packInstallIDs))
	for packInstallID := range packInstallIDs {
		ordered = append(ordered, packInstallID)
	}
	sort.Strings(ordered)
	for _, packInstallID := range ordered {
		if err := lockActiveCompanionPackIncarnation(ctx, tx, packInstallID); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) nextEnqueueOrdinal(ctx context.Context, tx *sql.Tx, packInstallID string) (int64, error) {
	// EnqueueBatch already holds the active _System_InstalledPack row lock for
	// this pack incarnation, so concurrent deliveries for one pack cannot race
	// this MAX()+1 allocator even across multiple workers / pods.
	cutoff, err := s.workqueue.WithTx(tx).ProjectionLegacyMaxOrdinal(ctx, packInstallID)
	if err != nil {
		return 0, err
	}
	var next int64
	err = tx.QueryRowContext(ctx, `
SELECT COALESCE(MAX(EnqueueOrdinal), 0) + 1
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?`, packInstallID).Scan(&next)
	if err != nil {
		return 0, err
	}
	if next < 1 || next <= cutoff {
		return 0, fmt.Errorf("%w: pack ordinal does not advance beyond its cutover", connector_workqueue.ErrProjectionProtocolRequired)
	}
	return next, nil
}

// lookupExisting returns the eventID of an existing row matching
// (installedPackID, idempotencyKey), or ("", false, nil) when none exists.
// Dedup is scoped to the pack incarnation (InstalledPackId) so same-name
// pack rebirths do not inherit events from a previous incarnation.
//
// q accepts either *sql.DB (for the concurrent-race slow-path re-read after
// a 1062 error, where the tx is already rolled back) or *sql.Tx (for the
// in-transaction dedup check after lockActivePackIncarnation, so a retired
// incarnation fails the lock first and never reaches the dedup SELECT).
func (s *Service) lookupExisting(ctx context.Context, q rowQuerier, installedPackID, idempotencyKey string) (string, bool, error) {
	var id string
	err := q.QueryRowContext(ctx, `
SELECT Id
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND IdempotencyKey = ?
 LIMIT 1`, installedPackID, idempotencyKey).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	return id, true, nil
}

type workItemEventSnapshot struct {
	eventID         string
	installedPackID string
	eventKind       string
	queueClass      string
	status          string
	sourceLaneKey   string
	laneOrdinal     int64
	nextAttemptAt   sql.NullTime
}

type freshWorkItemEvent struct {
	eventID               string
	installedPackID       string
	eventKind             string
	routeVersion          int64
	companionPack         string
	companionFlow         string
	queueClass            string
	sourceLaneKey         string
	targetMutationLaneKey string
	laneOrdinal           int64
	inputStartOrdinal     int64
	nextAttemptAt         sql.NullTime
	sourceProjection      bool
}

func validateSourceProjectionTargetMutationLane(event freshWorkItemEvent) error {
	if event.sourceProjection && strings.TrimSpace(event.sourceLaneKey) != "" && strings.TrimSpace(event.targetMutationLaneKey) != "" {
		return errors.New("connector-events: TargetMutationLane is not supported by settled source projection routes")
	}
	return nil
}

type routeEnqueueSpecKey struct {
	eventKind string
	version   int64
}

type routeEnqueueSpec struct {
	queueClass        string
	sourceSettleDelay time.Duration
	sourceProjection  bool
}

type sourceSettleKey struct {
	installedPackID string
	eventKind       string
	routeVersion    int64
	companionPack   string
	companionFlow   string
	sourceKind      string
	sourceID        string
}

func (s *Service) sourceSettleDeadline(ctx context.Context, tx *sql.Tx, key sourceSettleKey, now time.Time, delay time.Duration, cache map[sourceSettleKey]sql.NullTime) (sql.NullTime, error) {
	deadline := sql.NullTime{Time: now.Add(delay).UTC(), Valid: true}
	key = normalizeSourceSettleKey(key)
	if key.sourceKind == "" || key.sourceID == "" {
		return deadline, nil
	}
	if cached, ok := cache[key]; ok {
		return cached, nil
	}

	deadlines := map[sourceSettleKey]sql.NullTime{key: deadline}
	if err := loadSourceSettleDeadlines(ctx, tx, now, deadlines); err != nil {
		return sql.NullTime{}, err
	}
	deadline = deadlines[key]
	cache[key] = deadline
	return deadline, nil
}

func sqlPlaceholders(n int) string {
	if n <= 0 {
		return ""
	}
	return strings.TrimRight(strings.Repeat("?,", n), ",")
}

func lookupRouteEnqueueSpec(ctx context.Context, tx *sql.Tx, packInstallID, eventKind string, routeVersion int64, cache map[routeEnqueueSpecKey]routeEnqueueSpec) (routeEnqueueSpec, error) {
	key := routeEnqueueSpecKey{eventKind: strings.ToLower(strings.TrimSpace(eventKind)), version: routeVersion}
	if cached, ok := cache[key]; ok {
		return cached, nil
	}
	var queueClass, invocationMode string
	// Hold the current route through commit so the bulk VALUES insert cannot
	// snapshot a route/flow incarnation that is being republished concurrently.
	err := tx.QueryRowContext(ctx, `
SELECT COALESCE(NULLIF(QueueClass, ''), 'default'), COALESCE(InvocationMode, '')
  FROM _System_ConnectorInboundRoute
 WHERE InstalledPackId = ?
   AND EventKind       = ?
   AND Version         = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND COALESCE(IsCurrent, 0) = 1
 LIMIT 1
 FOR UPDATE`, packInstallID, key.eventKind, routeVersion).Scan(&queueClass, &invocationMode)
	if errors.Is(err, sql.ErrNoRows) {
		return routeEnqueueSpec{}, ErrPackIncarnationChanged
	}
	if err != nil {
		return routeEnqueueSpec{}, err
	}
	normalized, err := connector_inbound_routing.NormalizeQueueClass(queueClass)
	if err != nil {
		return routeEnqueueSpec{}, err
	}
	spec := routeEnqueueSpec{
		queueClass:        normalized,
		sourceSettleDelay: connector_inbound_routing.SourceSettleDelay(invocationMode),
		sourceProjection: strings.EqualFold(strings.TrimSpace(invocationMode),
			connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled),
	}
	cache[key] = spec
	return spec, nil
}

func (s *Service) ensureWorkItemForFreshEvent(ctx context.Context, tx *sql.Tx, event freshWorkItemEvent, now time.Time, deferLaneReconcile bool) (*connector_workqueue.CreateWorkItemInput, error) {
	if event.inputStartOrdinal == 0 {
		event.inputStartOrdinal = event.laneOrdinal
	}
	if s.workqueue == nil {
		return nil, errors.New("connector-events: WorkItem repository is required for enqueue")
	}
	if err := validateSourceProjectionTargetMutationLane(event); err != nil {
		return nil, err
	}
	state, runAfter, ok := workItemInitialState(statusPending, event.nextAttemptAt, now)
	if !ok {
		return nil, nil
	}
	workType := workTypeForEventKind(event.eventKind)
	workID := connectorEventWorkIDForWorkType(workType, event.eventID)
	if event.sourceProjection && strings.TrimSpace(event.sourceLaneKey) != "" {
		workID = sourceProjectionWorkID(event.installedPackID, event.eventKind, event.routeVersion,
			event.companionPack, event.companionFlow, workType, event.sourceLaneKey)
	}
	in := connector_workqueue.CreateWorkItemInput{
		WorkID:                workID,
		EventID:               event.eventID,
		InstalledPackID:       event.installedPackID,
		WorkType:              workType,
		QueueClass:            event.queueClass,
		State:                 state,
		RunAfter:              runAfter,
		LaneOrdinal:           event.laneOrdinal,
		SourceLaneKey:         event.sourceLaneKey,
		TargetMutationLaneKey: event.targetMutationLaneKey,
		Generation:            0,
		Now:                   now.UTC(),
	}
	workqueue := s.workqueue.WithTx(tx)
	var err error
	if connector_workqueue.IsSourceProjectionWorkID(workID) {
		var selected bool
		_, selected, err = workqueue.AdmitSourceProjectionWorkItem(ctx, connector_workqueue.SourceProjectionWorkItemInput{WorkItem: in, InputStartOrdinal: event.inputStartOrdinal}, deferLaneReconcile)
		if err != nil || !selected {
			return nil, err
		}
	} else if deferLaneReconcile {
		_, err = workqueue.EnsureWorkItemDeferredLaneReconcile(ctx, in)
	} else {
		_, err = workqueue.EnsureWorkItem(ctx, in)
	}
	if err != nil {
		return nil, err
	}
	return &in, nil
}

func (s *Service) ensureWorkItemForEvent(ctx context.Context, tx *sql.Tx, eventID, targetMutationLaneKey string, now time.Time, reuseExistingMirror, resolveProjectionSnapshot bool) (*connector_workqueue.CreateWorkItemInput, error) {
	return s.ensureWorkItemForEventWithReplay(
		ctx, tx, eventID, targetMutationLaneKey, now, reuseExistingMirror, resolveProjectionSnapshot,
		"", false, "", false,
	)
}

func (s *Service) ensureWorkItemForEventWithReplay(
	ctx context.Context,
	tx *sql.Tx,
	eventID, targetMutationLaneKey string,
	now time.Time,
	reuseExistingMirror, resolveProjectionSnapshot bool,
	replayOfWorkID string,
	replay bool,
	replayNamespace string,
	replaySourceProjection bool,
) (*connector_workqueue.CreateWorkItemInput, error) {
	if s.workqueue == nil {
		return nil, errors.New("connector-events: WorkItem repository is required for enqueue")
	}
	snapshot, found, err := s.lookupWorkItemEventSnapshot(ctx, tx, eventID)
	if err != nil {
		return nil, err
	}
	if !found {
		return nil, ErrEventNotFound
	}
	state, runAfter, ok := workItemInitialState(snapshot.status, snapshot.nextAttemptAt, now)
	if !ok && (!reuseExistingMirror || strings.TrimSpace(snapshot.sourceLaneKey) == "") {
		return nil, nil
	}
	workType := workTypeForEventKind(snapshot.eventKind)
	if reuseExistingMirror {
		existingWorkType, found, err := lookupExistingWorkItemTypeForEvent(ctx, tx, snapshot.eventID)
		if err != nil {
			return nil, err
		}
		if found {
			workType = existingWorkType
		}
	}
	projectionEvent := freshWorkItemEvent{
		eventID: snapshot.eventID, installedPackID: snapshot.installedPackID,
		eventKind: snapshot.eventKind, queueClass: snapshot.queueClass,
		sourceLaneKey: snapshot.sourceLaneKey, targetMutationLaneKey: targetMutationLaneKey,
		laneOrdinal: snapshot.laneOrdinal, nextAttemptAt: snapshot.nextAttemptAt,
	}
	if (snapshot.nextAttemptAt.Valid || resolveProjectionSnapshot || reuseExistingMirror) && strings.TrimSpace(snapshot.sourceLaneKey) != "" {
		projectionEvent, err = s.lookupSourceProjectionEvent(ctx, tx, projectionEvent)
		if err != nil {
			return nil, err
		}
	}
	// A selected projection Event becomes skipped at claim, even while its
	// WorkItem is still running. Receipt status cannot bypass durable coverage
	// verification or prove that a missing generation is safe to recreate.
	if !ok && !(reuseExistingMirror && projectionEvent.sourceProjection && !replay) {
		return nil, nil
	}
	if err := validateSourceProjectionTargetMutationLane(projectionEvent); err != nil {
		return nil, err
	}
	if replay && projectionEvent.sourceProjection != replaySourceProjection {
		return nil, connector_workqueue.ErrReplayBlocked
	}
	if replay && projectionEvent.sourceProjection {
		if err := s.rejectSourceProjectionReplayConflict(ctx, tx, replayNamespace, projectionEvent); err != nil {
			return nil, err
		}
	}
	workID := connectorEventWorkIDForWorkType(workType, snapshot.eventID)
	if projectionEvent.sourceProjection {
		workID = sourceProjectionWorkID(snapshot.installedPackID, snapshot.eventKind, projectionEvent.routeVersion,
			projectionEvent.companionPack, projectionEvent.companionFlow, workType, snapshot.sourceLaneKey)
	}
	in := connector_workqueue.CreateWorkItemInput{
		WorkID:                workID,
		EventID:               snapshot.eventID,
		InstalledPackID:       snapshot.installedPackID,
		WorkType:              workType,
		QueueClass:            snapshot.queueClass,
		State:                 state,
		RunAfter:              runAfter,
		LaneOrdinal:           snapshot.laneOrdinal,
		SourceLaneKey:         snapshot.sourceLaneKey,
		TargetMutationLaneKey: targetMutationLaneKey,
		Generation:            0,
		Now:                   now.UTC(),
	}
	workqueue := s.workqueue.WithTx(tx)
	if projectionEvent.sourceProjection {
		if !replay {
			var selected bool
			if reuseExistingMirror {
				_, selected, err = workqueue.EnsureSourceProjectionWorkItem(ctx, in, false)
			} else {
				_, selected, err = workqueue.AdmitSourceProjectionWorkItem(ctx, connector_workqueue.SourceProjectionWorkItemInput{WorkItem: in, InputStartOrdinal: in.LaneOrdinal}, false)
			}
			if err != nil || !selected {
				return nil, err
			}
		} else {
			_, _, err = workqueue.EnsureSourceProjectionReplayWorkItem(ctx, connector_workqueue.ReplayWorkItemInput{
				WorkItem:       in,
				ReplayOfWorkID: replayOfWorkID,
			}, false)
		}
	} else {
		if !replay {
			_, err = workqueue.EnsureWorkItem(ctx, in)
		} else {
			_, err = workqueue.EnsureReplayWorkItem(ctx, connector_workqueue.ReplayWorkItemInput{
				WorkItem:       in,
				ReplayOfWorkID: replayOfWorkID,
			})
		}
	}
	if err != nil {
		return nil, err
	}
	return &in, nil
}

// rejectSourceProjectionReplayConflict fences every nonterminal generation in
// the immutable source scope, including generations whose pack, route, or
// companion identity differs from the replay target. ReplayDeadEvent holds the
// current active-pack row while this check runs, serializing it with current
// Enqueue transactions. Event-only rows remain covered during the bounded
// compatibility window.
func (s *Service) rejectSourceProjectionReplayConflict(ctx context.Context, tx *sql.Tx, namespace string, replay freshWorkItemEvent) error {
	namespace = strings.TrimSpace(namespace)
	sourceLaneKey := strings.TrimSpace(replay.sourceLaneKey)
	if !replay.sourceProjection || namespace == "" || sourceLaneKey == "" {
		return nil
	}
	args := []any{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
		namespace,
		strings.ToLower(strings.TrimSpace(replay.eventKind)),
		sourceLaneKey,
		replay.eventID,
		connector_workqueue.WorkStateProcessed,
		connector_workqueue.WorkStateSkipped,
		connector_workqueue.WorkStateDead,
		statusPending,
		statusProcessing,
		statusFailed,
	}
	var conflictEventID string
	err := tx.QueryRowContext(ctx, `
SELECT candidate.Id
  FROM _System_ConnectorEvent candidate
 LEFT JOIN _System_ConnectorWorkItem owner
    ON owner.EventId = candidate.Id
   AND owner.WorkType IN (?, ?)
 WHERE candidate.Namespace = ?
   AND LOWER(TRIM(candidate.EventKind)) = ?
   AND COALESCE(candidate.SourceLaneKey, '') = ?
   AND candidate.Id <> ?
	   AND (
	       (owner.WorkId IS NOT NULL
	        AND owner.Attempt > 0
	        AND owner.State NOT IN (?, ?, ?))
	       OR ((owner.WorkId IS NULL OR owner.Attempt = 0)
	           AND candidate.Status IN (?, ?, ?)
	           AND NOT EXISTS (
	               SELECT 1
	                 FROM _System_ConnectorWorkItem superseding
	                 JOIN _System_ConnectorEvent selected
	                   ON selected.Id = superseding.EventId
	                  AND selected.InstalledPackId = superseding.InstalledPackId
	                WHERE superseding.InstalledPackId = candidate.InstalledPackId
	                  AND (
	                      NULLIF(COALESCE(superseding.ProjectionKey, ''), '') IS NOT NULL
	                      OR superseding.WorkId LIKE 'csp:%'
	                  )
	                  AND superseding.WorkType = CASE
	                      WHEN LOWER(TRIM(candidate.EventKind)) = 'workflow.action.execute'
	                          THEN 'workflow_action_apply'
	                      ELSE 'connector_event'
	                  END
	                  AND COALESCE(superseding.SourceLaneKey, '') = COALESCE(candidate.SourceLaneKey, '')
	                  AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(candidate.EventKind))
	                  AND COALESCE(selected.RouteVersion, 0) = COALESCE(candidate.RouteVersion, 0)
	                  AND COALESCE(selected.CompanionPackInstallId, '') = COALESCE(candidate.CompanionPackInstallId, '')
	                  AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE(candidate.CompanionFlowVersionId, '')
	                  AND (
	                      (
	                          selected.EnqueueOrdinal <> 0
	                          AND candidate.EnqueueOrdinal <> 0
	                          AND (
	                              selected.EnqueueOrdinal > candidate.EnqueueOrdinal
	                              OR (selected.EnqueueOrdinal = candidate.EnqueueOrdinal AND selected.Id > candidate.Id)
	                          )
	                      )
	                      OR (
	                          (selected.EnqueueOrdinal = 0 OR candidate.EnqueueOrdinal = 0)
	                          AND (
	                              selected.CreatedDate > candidate.CreatedDate
	                              OR (
	                                  selected.CreatedDate = candidate.CreatedDate
	                                  AND (
	                                      selected.DeliverySequence > candidate.DeliverySequence
	                                      OR (selected.DeliverySequence = candidate.DeliverySequence AND selected.Id > candidate.Id)
	                                  )
	                              )
	                          )
	                      )
	                  )
	           ))
	   )
 ORDER BY candidate.EnqueueOrdinal DESC, candidate.CreatedDate DESC, candidate.Id DESC
 LIMIT 1
 FOR UPDATE`, args...).Scan(&conflictEventID)
	if isMissingConnectorWorkItemSchema(err) {
		err = tx.QueryRowContext(ctx, `
SELECT candidate.Id
  FROM _System_ConnectorEvent candidate
 WHERE candidate.Namespace = ?
   AND LOWER(TRIM(candidate.EventKind)) = ?
   AND COALESCE(candidate.SourceLaneKey, '') = ?
   AND candidate.Id <> ?
   AND candidate.Status IN (?, ?, ?)
 ORDER BY candidate.EnqueueOrdinal DESC, candidate.CreatedDate DESC, candidate.Id DESC
 LIMIT 1
 FOR UPDATE`,
			namespace,
			strings.ToLower(strings.TrimSpace(replay.eventKind)),
			sourceLaneKey,
			replay.eventID,
			statusPending,
			statusProcessing,
			statusFailed,
		).Scan(&conflictEventID)
	}
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return connector_workqueue.ErrReplayBlocked
}

func (s *Service) lookupSourceProjectionEvent(ctx context.Context, tx *sql.Tx, event freshWorkItemEvent) (freshWorkItemEvent, error) {
	var invocationMode string
	err := tx.QueryRowContext(ctx, `
SELECT COALESCE(e.RouteVersion, 0),
       COALESCE(e.CompanionPackInstallId, ''),
       COALESCE(e.CompanionFlowVersionId, ''),
       COALESCE(r.InvocationMode, '')
  FROM _System_ConnectorEvent e
  LEFT JOIN _System_ConnectorInboundRoute r
    ON r.InstalledPackId = e.InstalledPackId
   AND r.EventKind = LOWER(TRIM(e.EventKind))
   AND r.Version = e.RouteVersion
 WHERE e.Id = ?
 LIMIT 1`, event.eventID).Scan(&event.routeVersion, &event.companionPack, &event.companionFlow, &invocationMode)
	if err != nil {
		return freshWorkItemEvent{}, err
	}
	event.sourceProjection = strings.EqualFold(strings.TrimSpace(invocationMode),
		connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled)
	return event, nil
}

func rememberTouchedWorkItemLane(touched map[string]connector_workqueue.CreateWorkItemInput, in *connector_workqueue.CreateWorkItemInput) {
	if touched == nil || in == nil {
		return
	}
	laneKey := strings.TrimSpace(in.TargetMutationLaneKey)
	if laneKey == "" {
		laneKey = strings.TrimSpace(in.SourceLaneKey)
	}
	if laneKey == "" {
		return
	}
	touched[fmt.Sprintf("%s\x00%d\x00%s", strings.TrimSpace(in.InstalledPackID), in.Generation, laneKey)] = *in
}

func (s *Service) lookupWorkItemEventSnapshot(ctx context.Context, q rowQuerier, eventID string) (workItemEventSnapshot, bool, error) {
	var snapshot workItemEventSnapshot
	err := q.QueryRowContext(ctx, `
SELECT Id, InstalledPackId, EventKind, QueueClass, Status, COALESCE(SourceLaneKey, ''), EnqueueOrdinal, NextAttemptAt
  FROM _System_ConnectorEvent
 WHERE Id = ?
 LIMIT 1`, eventID).Scan(
		&snapshot.eventID,
		&snapshot.installedPackID,
		&snapshot.eventKind,
		&snapshot.queueClass,
		&snapshot.status,
		&snapshot.sourceLaneKey,
		&snapshot.laneOrdinal,
		&snapshot.nextAttemptAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return workItemEventSnapshot{}, false, nil
	}
	if err != nil {
		return workItemEventSnapshot{}, false, err
	}
	return snapshot, true, nil
}

func lookupExistingWorkItemTypeForEvent(ctx context.Context, q rowQuerier, eventID string) (string, bool, error) {
	var workType string
	err := q.QueryRowContext(ctx, `
SELECT WorkType
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?
 ORDER BY CASE WHEN WorkType = ? THEN 0 ELSE 1 END, CreatedDate ASC, WorkId ASC
 LIMIT 1`, eventID, connector_workqueue.WorkTypeConnectorEvent).Scan(&workType)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	workType = strings.TrimSpace(workType)
	if !isEventBackedWorkItemType(workType) {
		return "", false, nil
	}
	return workType, true, nil
}

func isEventBackedWorkItemType(workType string) bool {
	switch strings.TrimSpace(workType) {
	case connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply:
		return true
	default:
		return false
	}
}

func workItemInitialState(status string, nextAttemptAt sql.NullTime, now time.Time) (string, time.Time, bool) {
	switch strings.ToLower(strings.TrimSpace(status)) {
	case statusPending:
		runAfter := now.UTC()
		if nextAttemptAt.Valid && !nextAttemptAt.Time.IsZero() {
			runAfter = nextAttemptAt.Time.UTC()
		}
		return connector_workqueue.WorkStatePending, runAfter, true
	case statusFailed:
		runAfter := now.UTC()
		if nextAttemptAt.Valid && !nextAttemptAt.Time.IsZero() {
			runAfter = nextAttemptAt.Time.UTC()
		}
		return connector_workqueue.WorkStateRetry, runAfter, true
	default:
		return "", time.Time{}, false
	}
}

func connectorEventWorkIDForWorkType(workType, eventID string) string {
	seed := strings.TrimSpace(workType) + "\x00" + strings.TrimSpace(eventID)
	sum := sha256.Sum256([]byte(seed))
	return "cwi:" + hex.EncodeToString(sum[:])
}

func sourceProjectionWorkID(installedPackID, eventKind string, routeVersion int64, companionPack, companionFlow, workType, sourceLaneKey string) string {
	return connector_workqueue.SourceProjectionKey(connector_workqueue.SourceProjectionIdentity{
		InstalledPackID: installedPackID, EventKind: eventKind, RouteVersion: routeVersion, CompanionPackInstallID: companionPack,
		CompanionFlowVersionID: companionFlow, WorkType: workType, SourceLaneKey: sourceLaneKey,
	})
}

func workTypeForEventKind(eventKind string) string {
	if strings.EqualFold(strings.TrimSpace(eventKind), "workflow.action.execute") {
		return connector_workqueue.WorkTypeWorkflowActionApply
	}
	return connector_workqueue.WorkTypeConnectorEvent
}

func targetMutationLaneKeyFromBatchEvent(e BatchEvent) (string, error) {
	targetLaneObjectKind := strings.TrimSpace(e.TargetMutationLaneObjectKind)
	targetLaneID := e.TargetMutationLaneID
	hasTargetLaneID := strings.TrimSpace(targetLaneID) != ""
	if targetLaneObjectKind == "" && !hasTargetLaneID {
		return "", nil
	}
	if targetLaneObjectKind == "" || !hasTargetLaneID {
		return "", errors.New("connector-events: TargetMutationLaneObjectKind and TargetMutationLaneID must be configured together")
	}
	return sourceLaneKeyFor(targetLaneObjectKind, targetLaneID), nil
}

func isMissingConnectorWorkItemSchema(err error) bool {
	if err == nil {
		return false
	}
	missingColumns := []string{"laneordinal", "projectionkey", "projectionactivekey", "terminalat", "replayedat", "replayofworkid"}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		msg := strings.ToLower(mysqlErr.Message)
		if mysqlErr.Number == 1146 && strings.Contains(msg, "_system_connectorworkitem") {
			return true
		}
		if mysqlErr.Number == 1054 {
			for _, column := range missingColumns {
				if strings.Contains(msg, column) {
					return true
				}
			}
		}
	}
	msg := strings.ToLower(err.Error())
	if strings.Contains(msg, "_system_connectorworkitem") &&
		(strings.Contains(msg, "1146") || strings.Contains(msg, "doesn't exist") || strings.Contains(msg, "no such table")) {
		return true
	}
	for _, column := range missingColumns {
		if strings.Contains(msg, column) &&
			(strings.Contains(msg, "1054") || strings.Contains(msg, "unknown column") || strings.Contains(msg, "no such column")) {
			return true
		}
	}
	return false
}

type legacyEventWorkItemGuardMode uint8

const (
	legacyEventWorkItemGuardFull legacyEventWorkItemGuardMode = iota
	legacyEventWorkItemGuardDirect
	legacyEventWorkItemGuardNone
)

func nextLegacyEventWorkItemGuardMode(mode legacyEventWorkItemGuardMode, err error) (legacyEventWorkItemGuardMode, bool) {
	if err == nil {
		return mode, false
	}
	if isMissingConnectorEventExecutionOwner(err) {
		if mode != legacyEventWorkItemGuardNone {
			return legacyEventWorkItemGuardNone, true
		}
		return mode, false
	}
	if isMissingConnectorWorkItemTable(err) {
		if mode != legacyEventWorkItemGuardNone {
			return legacyEventWorkItemGuardNone, true
		}
		return mode, false
	}
	if mode == legacyEventWorkItemGuardFull && isMissingConnectorProjectionGuardSchema(err) {
		// Projection columns are part of the WorkItem claimant's complete
		// schema. When they are absent the supervisor runs the legacy Event
		// claimant, so retain only the per-Event ExecutionOwner fence. Existing
		// csp:* compatibility rows must not fence their whole source lane.
		return legacyEventWorkItemGuardDirect, true
	}
	return mode, false
}

func isMissingConnectorEventExecutionOwner(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		return mysqlErr.Number == 1054 &&
			strings.Contains(strings.ToLower(mysqlErr.Message), "executionowner")
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "executionowner") &&
		(strings.Contains(msg, "1054") || strings.Contains(msg, "unknown column") ||
			strings.Contains(msg, "no such column"))
}

func isMissingConnectorWorkItemTable(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		return mysqlErr.Number == 1146 &&
			strings.Contains(strings.ToLower(mysqlErr.Message), "_system_connectorworkitem")
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "_system_connectorworkitem") &&
		(strings.Contains(msg, "1146") || strings.Contains(msg, "doesn't exist") ||
			strings.Contains(msg, "does not exist") || strings.Contains(msg, "no such table"))
}

func isMissingConnectorProjectionGuardSchema(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	if !strings.Contains(msg, "projectionkey") && !strings.Contains(msg, "projectionactivekey") {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		return mysqlErr.Number == 1054
	}
	return strings.Contains(msg, "1054") || strings.Contains(msg, "unknown column") ||
		strings.Contains(msg, "no such column")
}

// ClaimNext atomically transitions one pending-or-failed event for installedPackID
// to processing, returning the claimed row.  Returns (nil, false, nil) when no
// claimable event is available.
//
// installedPackID is the _System_InstalledPack.Id for the specific pack incarnation
// the calling worker belongs to.  Scoping claims to InstalledPackId (rather than
// Namespace) ensures workers for a reborn pack cannot accidentally drain events
// enqueued for an earlier incarnation that shares the same name.
//
// Uses two-step SELECT + UPDATE-WHERE to ensure only one worker wins per row
// even under TiDB optimistic txn mode (SELECT … FOR UPDATE SKIP LOCKED is
// forbidden — see outbox_repository.go:ClaimEvent for context).  Retries up
// to claimContentionRetries times on lost-race before giving up.
//
// Single-winner guarantee: the UPDATE-WHERE predicate (AND Status IN ('pending',
// 'failed') OR stale-processing) is the exclusive arbiter — only the one worker
// whose UPDATE touches RowsAffected==1 wins the row.  The retry loop is a
// livelock guard, not an ordering mechanism.  Callers MUST implement their own
// external poll interval if (nil, false, nil) is returned under heavy contention;
// busy-spinning against ClaimNext without a delay will starve lower-priority work.
//
// Per-delivery execution ordering (issue #2154): the claim SELECT excludes any row
// that still has an earlier-DeliverySequence sibling in the same DeliveryId in a
// non-terminal state, so a later same-delivery event is never claimed (hence never
// executed) before its prerequisite reaches a terminal state. The claim UPDATE
// re-applies the SAME exclusion atomically (keyed on the candidate's DeliveryId +
// DeliverySequence), so a concurrent ReplayDeadEvent that re-introduces an earlier
// pending sibling between the SELECT and the UPDATE cannot let a later event win the
// claim — RowsAffected==0 and the loop re-selects. Different DeliveryIds can
// still drain in parallel unless the rows also share the same queue-envelope
// lane (`objectKind` + `sourceID`), so one stuck event blocks only its own
// delivery and same-source siblings, not the whole pack; NULL-DeliveryId
// (legacy) rows are independent singletons unless the same-source branch
// applies.
//
// Stale-processing reclamation: if a worker dies between claim and ack, its row
// stays in 'processing' indefinitely unless another caller can reclaim it.
// ClaimNext reclaims any row whose LastModifiedDate is older than
// StaleProcessingSeconds, matching the outbox precedent in outbox_repository.go.
// Both the SELECT and UPDATE use the same staleCutoff snapshot so the two steps
// agree on what counts as stale within one ClaimNext call.
//
// Does not gate on pack activation state — once an event is enqueued, drain/finalize
// must work during pack removing/inactive transitions.
func (s *Service) ClaimNext(ctx context.Context, installedPackID, workerID string) (*ConnectorEvent, bool, error) {
	return s.claimNextImpl(ctx, installedPackID, workerID, nil, false)
}

func (s *Service) ClaimNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*ConnectorEvent, bool, error) {
	return s.claimNextImpl(ctx, installedPackID, workerID, queueClasses, false)
}

// ReclaimStaleNext is the stale-reclaim-only claim used by the limiter's limp
// path. It claims ONLY a stale-processing row (never a fresh pending/failed row),
// so a degraded tick recovers a stranded lease without admitting new work. Because
// it re-leases a row that is already 'processing', ProcessingCount stays flat and
// it cannot over-admit past the limiter's maxProcessingRows ceiling. Ordering
// guard, ORDER BY and the UPDATE-side CAS are identical to ClaimNext — only the
// candidate SELECT's status predicate is narrowed to the stale-processing arm.
func (s *Service) ReclaimStaleNext(ctx context.Context, installedPackID, workerID string) (*ConnectorEvent, bool, error) {
	return s.claimNextImpl(ctx, installedPackID, workerID, nil, true)
}

func (s *Service) ReclaimStaleNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*ConnectorEvent, bool, error) {
	return s.claimNextImpl(ctx, installedPackID, workerID, queueClasses, true)
}

// claimCandidateStatusSQL is the candidate status predicate shared by the claim
// SELECT and PeekBatch: a row is claimable when its pending not-before is due,
// its failed retry is due, or its processing lease is stale. Binds: (pending-due
// cutoff = now, failed-due cutoff = now, stale cutoff = staleCutoff).
const claimCandidateStatusSQL = `(
     (e.Status = 'pending' AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
     OR (e.Status = 'failed' AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
     OR (e.Status = 'processing' AND e.LastModifiedDate < ?)
   )`

// claimStaleOnlyStatusSQL is the stale-reclaim-only predicate for the limiter's
// limp path: only a stale-processing row (past the reclaim cutoff) is a candidate.
// Reclaiming such a row re-leases an existing 'processing' row, so ProcessingCount
// stays flat and the limp claim cannot over-admit past maxProcessingRows. Bind:
// (stale cutoff = staleCutoff).
const claimStaleOnlyStatusSQL = `(e.Status = 'processing' AND e.LastModifiedDate < ?)`

func (s *Service) claimNextImpl(ctx context.Context, installedPackID, workerID string, queueClasses []string, staleOnly bool) (*ConnectorEvent, bool, error) {
	queueClassSQL, queueClassArgs, err := queueClassFilterSQL("e", queueClasses)
	if err != nil {
		return nil, false, err
	}
	now := s.now()
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)
	// staleOnly narrows the candidate SELECT to the stale-processing arm (the limp
	// path) so a degraded tick reclaims a stranded lease without admitting a fresh
	// pending/failed row. The UPDATE below keeps the full predicate — a stale
	// candidate still CASes cleanly through its stale arm.
	candidateStatusSQL := claimCandidateStatusSQL
	candidateStatusArgs := []interface{}{now, now, staleCutoff}
	if staleOnly {
		candidateStatusSQL = claimStaleOnlyStatusSQL
		candidateStatusArgs = []interface{}{staleCutoff}
	}
	// A legacy Event worker can coexist briefly with WorkItem workers during a
	// rolling transition. The durable guard excludes every Event whose execution
	// is represented by a WorkItem, plus unselected successors owned by an active
	// source projection. If the WorkItem schema is unavailable, the guarded query
	// falls back to the explicit legacy-only predicate below.
	workItemGuardMode := legacyEventWorkItemGuardFull
	for attempt := 0; attempt < claimContentionRetries; attempt++ {
		// Step 1: pick the oldest claimable pending/failed/stale-processing
		// candidate.
		//
		// ORDER BY (CreatedDate, DeliverySequence, Id): events from one webhook
		// delivery share a CreatedDate, so DeliverySequence — the dispatcher's
		// per-delivery array index — is the FIFO tiebreaker that makes a link event
		// claimed after the create events it depends on (issue #2125).
		// Delivery ordering plus same-source ordering: the NOT EXISTS guard below
		// makes claim SELECTION enforce execution COMPLETION order within a delivery,
		// and also within one queue-envelope lane (objectKind + sourceID) when
		// related rows arrive as separate deliveries. Under multiple concurrent
		// workers on one InstalledPackId (concurrent_claim_integration_test), a
		// later-sequence event of the same DeliveryId is not claimable until every
		// earlier sibling is terminal, and a later same-source row is likewise
		// blocked behind the earlier same-source row. #2134's IsRetriable re-queue
		// remains the backstop for genuine failures.
		//
		// NOT EXISTS guard: a row is claimable only when every earlier sibling in the
		// same DeliveryId — by strict DeliverySequence — is terminal, and when every
		// earlier sibling in the same (objectKind, sourceID) lane is terminal. This
		// keeps one logical source record single-flight even when the dispatcher
		// emits one row per delivery.
		//
		// Strict `<` is safe because EnqueueBatch writes a delivery's events
		// all-or-nothing in one transaction, so live rows for one DeliveryId always
		// have distinct DeliverySequences (a complete 0..N-1 range). ReplayDeadEvent
		// copies the original dead row's DeliverySequence onto the replay row; the
		// replay row is pending while the original is dead, so the guard's
		// `Status NOT IN ('processed','dead')` excludes the dead original — no
		// equal-sequence live collision is possible with the strict `<` predicate.
		//
		// NULL DeliveryId (legacy rows): the per-delivery branch never matches when
		// either side is NULL (SQL NULL semantics). They can still participate in the
		// same-source branch when the queue envelope carries objectKind + sourceID.
		// Stale-safe: a stale-processing earlier sibling is still non-terminal → blocks
		// its delivery tail, but is itself the lowest non-terminal → reclaimable by the
		// CAS (processing AND LastModifiedDate < staleCutoff) → retried → eventually
		// terminal → unblocks. No deadlock; does not fight stale-reclaim.
		var candidateID string
		var candidateDeliveryID sql.NullString
		var candidateSeq int
		// Direction-2 (NCRM #3140): the source-lane + delivery anti-join branches
		// of the guard are expressed as equi-joins to per-lane / per-delivery HEAD
		// derived tables (buildLaneHeadJoin / buildDeliveryHeadJoin +
		// claimHeadWhereSQL), computed once instead of per-candidate, so a deep
		// backlog no longer makes this SELECT an O(N^2) full-scan anti-join. The
		// claimable set is identical to the single-row guard (the UPDATE below
		// still uses that form for the atomic re-check).
		//
		// NCRM #3139 (Problem A): the lane-head source set and the residual
		// source-lane arms carry the NextAttemptAt-gated blocker (bound to `now`),
		// so a backing-off failed head is skipped and its newer sibling becomes
		// claimable; the delivery-head stays strict FIFO.
		laneHeadJoinSQL, laneHeadBinds := buildLaneHeadJoin(installedPackID, now)
		deliveryHeadJoinSQL, deliveryHeadBinds := buildDeliveryHeadJoin(installedPackID)
		residualSQL, residualBinds := buildClaimOrderingResidualSQL(now)
		candidateArgs := append([]interface{}{}, laneHeadBinds...)
		candidateArgs = append(candidateArgs, deliveryHeadBinds...)
		candidateArgs = append(candidateArgs, installedPackID)
		candidateArgs = append(candidateArgs, candidateStatusArgs...)
		candidateArgs = append(candidateArgs, queueClassArgs...)
		candidateArgs = append(candidateArgs, residualBinds...)
		workItemGuardSQL, workItemGuardBinds := s.legacyEventWorkItemGuardSQL("e", workItemGuardMode)
		candidateArgs = append(candidateArgs, workItemGuardBinds...)
		err := s.db.QueryRowContext(ctx, `
SELECT e.Id, e.DeliveryId, e.DeliverySequence
  FROM _System_ConnectorEvent e`+laneHeadJoinSQL+deliveryHeadJoinSQL+`
 WHERE e.InstalledPackId = ?
   AND `+candidateStatusSQL+`
`+queueClassSQL+claimHeadWhereSQL+residualSQL+workItemGuardSQL+`
 ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC
	 LIMIT 1`, candidateArgs...).Scan(&candidateID, &candidateDeliveryID, &candidateSeq)
		if errors.Is(err, sql.ErrNoRows) {
			recordConnectorEventClaim("next", "empty")
			return nil, false, nil
		}
		if nextMode, retry := nextLegacyEventWorkItemGuardMode(workItemGuardMode, err); retry {
			workItemGuardMode = nextMode
			continue
		}
		if err != nil {
			recordConnectorEventClaim("next", "error")
			return nil, false, err
		}

		// Test seam (nil in production): deterministically exercise the window
		// between the candidate SELECT and the claim UPDATE.
		if s.afterClaimSelect != nil {
			s.afterClaimSelect()
		}

		// Step 2: claim — exclusive single-winner update.
		// Mint a fresh per-claim token before the UPDATE so every successful
		// claim gets a nonce that is unique to this particular lease grant.
		// Workers that restart with the same workerID, or replicas sharing a
		// configured workerID, cannot match a different worker's ClaimToken.
		freshToken := s.idgen()
		// The CAS keeps the SAME status predicate as the candidate SELECT. For the
		// stale-only (limp) path this means a candidate that a concurrent reclaim
		// advanced (processing→failed, or refreshed to a non-stale lease) between the
		// SELECT and here no longer matches → RowsAffected 0 → retry/empty, so a limp
		// tick re-leases ONLY a row that is still stale-processing, never a drifted
		// pending/failed row.
		guardSQL, guardBinds := buildClaimOrderingGuardSQL(now)
		updateArgs := append([]interface{}{statusProcessing, workerID, now, freshToken, now, candidateID}, candidateStatusArgs...)
		updateArgs = append(updateArgs, queueClassArgs...)
		updateArgs = append(updateArgs, guardBinds...)
		updateArgs = append(updateArgs, workItemGuardBinds...)
		res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       ClaimedBy = ?,
       ClaimedAt = ?,
       ClaimToken = ?,
       LastModifiedDate = ?
 WHERE e.Id = ?
   AND `+candidateStatusSQL+`
`+queueClassSQL+guardSQL+workItemGuardSQL,
			updateArgs...)
		if nextMode, retry := nextLegacyEventWorkItemGuardMode(workItemGuardMode, err); retry {
			workItemGuardMode = nextMode
			continue
		}
		if err != nil {
			recordConnectorEventClaim("next", "error")
			return nil, false, err
		}
		n, err := res.RowsAffected()
		if err != nil {
			recordConnectorEventClaim("next", "error")
			return nil, false, err
		}
		if n == 0 {
			// Lost race — another worker claimed this row, OR a concurrent
			// ReplayDeadEvent re-introduced an earlier-DeliverySequence sibling in
			// the same DeliveryId between our SELECT and UPDATE (the UPDATE's
			// NOT EXISTS now fails). Either way, retry: the next SELECT re-evaluates
			// ordering and picks the correct (lowest non-terminal) candidate.
			continue
		}

		// Won the claim — load the full row (ClaimToken will be freshToken).
		ev, err := s.loadByID(ctx, candidateID)
		if err != nil {
			recordConnectorEventClaim("next", "error")
			return nil, false, err
		}
		recordConnectorEventClaim("next", "claimed")
		return ev, true, nil
	}
	// Exhausted retries — caller should back off and re-poll.
	recordConnectorEventClaim("next", "contention")
	return nil, false, nil
}

// legacyEventWorkItemClaimGuardSQL prevents every legacy Event claim surface
// from racing durable WorkItem authority. ExecutionOwner is the per-Event
// arbiter: a materialized attempt-zero WorkItem remains legacy-claimable until
// its atomic WorkItem claim changes the Event owner. Source projections
// additionally fence their whole active lane because a newly enqueued successor
// is not yet the projection WorkItem's EventId while the current generation is
// processing.
//
// ExecutionOwner is part of the guard for two reasons: once present, it fences a
// transferred Event even if a corrupt/manual repair removes its direct WorkItem;
// while absent, its unknown-column error proves the complete WorkItem claim
// schema is unavailable and makes the caller retry without the entire guard.
// That keeps the bounded legacy claimant live while the supervisor is also
// falling back to it during a mixed-schema rollout.
func (s *Service) legacyEventWorkItemClaimGuardSQL(eventAlias string) (string, []interface{}) {
	return s.legacyEventWorkItemGuardSQL(eventAlias, legacyEventWorkItemGuardFull)
}

func (s *Service) legacyEventWorkItemGuardSQL(eventAlias string, mode legacyEventWorkItemGuardMode) (string, []interface{}) {
	eventAlias = strings.TrimSpace(eventAlias)
	if eventAlias == "" {
		eventAlias = "e"
	}
	if mode == legacyEventWorkItemGuardNone {
		return "", nil
	}
	ownerGuard := `
   AND ` + eventAlias + `.ExecutionOwner = '` + connector_workqueue.EventExecutionOwnerEvent + `'`
	if mode == legacyEventWorkItemGuardDirect {
		return ownerGuard, nil
	}
	projectionAuthority := `
              NULLIF(COALESCE(projection.ProjectionActiveKey, ''), '') IS NOT NULL
              OR (
                  NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NULL
                  AND projection.WorkId LIKE 'csp:%'
                  AND projection.State NOT IN ('processed', 'skipped', 'dead')
              )`
	return ownerGuard + `
	   AND NOT EXISTS (
	       SELECT 1
	         FROM _System_ConnectorWorkItem projection
	         JOIN _System_ConnectorEvent selected ON selected.Id = projection.EventId
	        WHERE projection.InstalledPackId = ` + eventAlias + `.InstalledPackId
	          AND (
` + projectionAuthority + `
	          )
          AND COALESCE(projection.SourceLaneKey, '') = COALESCE(` + eventAlias + `.SourceLaneKey, '')
          AND selected.InstalledPackId = ` + eventAlias + `.InstalledPackId
          AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(` + eventAlias + `.EventKind))
          AND COALESCE(selected.RouteVersion, 0) = COALESCE(` + eventAlias + `.RouteVersion, 0)
          AND COALESCE(selected.CompanionPackInstallId, '') = COALESCE(` + eventAlias + `.CompanionPackInstallId, '')
          AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE(` + eventAlias + `.CompanionFlowVersionId, '')
   )`, nil
}

func (s *Service) execLegacyEventMutation(ctx context.Context, q workItemEventQueryExecer, query string, args ...any) (sql.Result, error) {
	// Once a legacy Event lease has been granted, its ClaimToken and
	// ExecutionOwner are the durable arbitration facts. Re-running the
	// lane-wide pre-claim projection guard here could strand the winning legacy
	// lease if a compatibility WorkItem appears after selection.
	mode := legacyEventWorkItemGuardDirect
	for {
		guardSQL, guardArgs := s.legacyEventWorkItemGuardSQL("e", mode)
		allArgs := append(append([]any(nil), args...), guardArgs...)
		result, err := q.ExecContext(ctx, query+guardSQL, allArgs...)
		nextMode, retry := nextLegacyEventWorkItemGuardMode(mode, err)
		if retry {
			mode = nextMode
			continue
		}
		return result, err
	}
}

func queueClassFilterSQL(alias string, queueClasses []string) (string, []interface{}, error) {
	if len(queueClasses) == 0 {
		return "", nil, nil
	}
	seen := make(map[string]struct{}, len(queueClasses))
	normalized := make([]string, 0, len(queueClasses))
	for _, raw := range queueClasses {
		queueClass, err := connector_inbound_routing.NormalizeQueueClass(raw)
		if err != nil {
			return "", nil, err
		}
		if _, ok := seen[queueClass]; ok {
			continue
		}
		seen[queueClass] = struct{}{}
		normalized = append(normalized, queueClass)
	}
	if len(normalized) == 0 {
		return "", nil, nil
	}
	column := "QueueClass"
	if strings.TrimSpace(alias) != "" {
		column = strings.TrimSpace(alias) + ".QueueClass"
	}
	placeholders := make([]string, len(normalized))
	args := make([]interface{}, len(normalized))
	for i, queueClass := range normalized {
		placeholders[i] = "?"
		args[i] = queueClass
	}
	return "\n   AND COALESCE(" + column + ", 'default') IN (" + strings.Join(placeholders, ", ") + ")\n", args, nil
}

// MarkProcessed transitions the event from processing to processed (terminal success).
//
// claimToken must match the ClaimToken column written by ClaimNext for this
// specific lease grant.  ClaimToken is a fresh nonce minted per successful
// ClaimNext call and is returned in ConnectorEvent.ClaimToken.  Using a
// per-claim token instead of workerID prevents a restarted worker (same ID)
// or a replica sharing a configured workerID from accidentally acking another
// worker's in-flight row.
//
// Returns ErrEventAlreadyTerminal when the event is already processed or dead.
// Returns ErrLeaseLost when RowsAffected == 0 (covers both lease-lost and
// any residual TOCTOU window between the pre-check and the UPDATE).
//
// Does not gate on pack activation state — once an event is enqueued, drain/finalize
// must work during pack removing/inactive transitions.
func (s *Service) MarkProcessed(ctx context.Context, eventID, claimToken string) error {
	return s.markTerminalSuccess(ctx, eventID, claimToken, statusProcessed)
}

// MarkSkipped transitions the event from processing to skipped. Skipped is a
// terminal, non-failure outcome used when route metadata declares that this
// event should remain auditable but should not execute the routed Flow.
func (s *Service) MarkSkipped(ctx context.Context, eventID, claimToken string) error {
	return s.markTerminalSuccess(ctx, eventID, claimToken, statusSkipped)
}

// ReleaseClaim returns a currently leased row to the pending queue without
// incrementing RetryCount. It is used when a worker intentionally claimed a
// later same-delivery row for batch assembly but must defer finalization because
// an earlier same-delivery member in that same batch did not reach terminal
// success. The claim token gate keeps release scoped to the caller's lease.
func (s *Service) ReleaseClaim(ctx context.Context, eventID, claimToken string) error {
	now := s.now()
	res, err := s.execLegacyEventMutation(ctx, s.db, `
	UPDATE _System_ConnectorEvent e
	   SET Status = ?,
	       ClaimedBy = NULL,
	       ClaimedAt = NULL,
	       ClaimToken = NULL,
	       LastModifiedDate = ?
	 WHERE e.Id = ?
	   AND e.Status = ?
	   AND e.ClaimToken = ?`,
		statusPending, now, eventID, statusProcessing, claimToken)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrLeaseLost
	}
	return nil
}

// RenewLease is the lease-heartbeat primitive: it bumps LastModifiedDate to now()
// for a row this worker still holds in 'processing' under claimToken, so a
// slow-but-alive flow keeps its lease fresh and never crosses the stale-reclaim
// cutoff (ClaimNext/ReclaimStaleNext reclaim a row only when
// LastModifiedDate < now-StaleProcessingSeconds). Reusing LastModifiedDate — the
// exact column the stale cutoff already reads — keeps one lease SSoT: there is no
// second heartbeat clock that could disagree with the reclaim decision.
//
// The worker calls this periodically (every StaleProcessingSeconds/3) for each
// in-flight claimed event until that event finalizes. A CRASHED worker stops
// renewing, so its row crosses the cutoff after StaleProcessingSeconds and is
// reclaimed exactly as today. So "stale" now means dead, not slow.
//
// Race-safety (the renewal MUST be a no-op once the lease is gone). The CAS gate
// (Status='processing' AND ClaimToken=?) is byte-identical to the gate every
// finalizer uses (MarkProcessed/markTerminalSuccess, MarkFailed, ReleaseClaim,
// ReleaseClaimForRetry). Each of those, on success, either flips Status off
// 'processing' (terminal) or sets ClaimToken=NULL (release), or a concurrent
// reclaim re-leases the row under a FRESH token. In every one of those cases this
// UPDATE's WHERE no longer matches → RowsAffected==0 → ErrLeaseLost and NO row is
// touched. A renewal can therefore never un-finalize a terminal row, never revive
// a released row, and never bump LastModifiedDate on a row another worker
// reclaimed under a new token. It only ever refreshes a lease the caller still
// genuinely holds. Returns ErrLeaseLost (not an error) when the lease is gone so
// the heartbeat caller can stop quietly.
//
// Does not gate on pack activation state — like the finalizers, an in-flight
// drain must keep its lease alive during pack removing/inactive transitions.
func (s *Service) RenewLease(ctx context.Context, eventID, claimToken string) error {
	now := s.now()
	res, err := s.execLegacyEventMutation(ctx, s.db, `
	UPDATE _System_ConnectorEvent e
	   SET LastModifiedDate = ?
	 WHERE e.Id = ?
	   AND e.Status = ?
	   AND e.ClaimToken = ?`,
		now, eventID, statusProcessing, claimToken)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrLeaseLost
	}
	return nil
}

// ReleaseClaimForRetry returns a currently leased row to the queue for a
// backed-off re-drive WITHOUT incrementing RetryCount, so a retriable condition
// does not consume the genuine-failure dead-letter budget. It is used when a
// flow signals FlowRetriableFault (IsRetriable on a failed result): the worker
// must re-attempt later rather than dead-letter the event.
//
// The row is set to 'failed' (not 'pending') with NextAttemptAt = now + backoff.
// Routing through the 'failed' arm distinguishes a retriable execution from an
// initial pending route settle window. RetryCount is left untouched, so the row
// never dead-letters on the genuine-failure budget while it is only deferring
// for a concurrent winner; termination is bounded upstream (see
// connectorRetriableReclaimDelay).
//
// reason is written to LastError so operators can distinguish a transient
// retriable defer from a permanent failure. The claim token gate keeps release
// scoped to the caller's lease.
func (s *Service) ReleaseClaimForRetry(ctx context.Context, eventID, claimToken, reason string) error {
	now := s.now()
	nextAttemptAt := now.Add(connectorRetriableReclaimDelay)
	res, err := s.execLegacyEventMutation(ctx, s.db, `
	UPDATE _System_ConnectorEvent e
	   SET Status = ?,
	       NextAttemptAt = ?,
	       LastError = ?,
	       LastErrorAt = ?,
	       ClaimedBy = NULL,
	       ClaimedAt = NULL,
	       ClaimToken = NULL,
	       LastModifiedDate = ?
	 WHERE e.Id = ?
	   AND e.Status = ?
	   AND e.ClaimToken = ?`,
		statusFailed, nextAttemptAt, reason, now, now, eventID, statusProcessing, claimToken)
	if err != nil {
		recordConnectorEventFinalize("retriable_defer", "error")
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		recordConnectorEventFinalize("retriable_defer", "error")
		return err
	}
	if n == 0 {
		recordConnectorEventFinalize("retriable_defer", "lease_lost")
		return ErrLeaseLost
	}
	recordConnectorEventFinalize("retriable_defer", "ok")
	return nil
}

func (s *Service) markTerminalSuccess(ctx context.Context, eventID, claimToken, terminalStatus string) error {
	// Preserve the historical error boundary: an Event that was already
	// terminal when acknowledgement began reports ErrEventAlreadyTerminal,
	// while a terminalization that wins the race after this read is a lost lease.
	_, currentStatus, err := s.loadNamespaceAndStatus(ctx, eventID)
	if err != nil {
		recordConnectorEventFinalize("processed", "error")
		return err
	}
	if currentStatus == statusProcessed || currentStatus == statusSkipped || currentStatus == statusDead {
		recordConnectorEventFinalize("processed", "already_terminal")
		return ErrEventAlreadyTerminal
	}

	now := s.now()
	err = s.terminalizeLegacyEvent(ctx, eventID, claimToken, func(tx *sql.Tx) (bool, error) {
		res, err := s.execLegacyEventMutation(ctx, tx, `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       ProcessedAt = ?,
       LastError = CASE WHEN ? = ? THEN NULL ELSE LastError END,
       LastErrorAt = CASE WHEN ? = ? THEN NULL ELSE LastErrorAt END,
       NextAttemptAt = NULL,
       LastModifiedDate = ?
WHERE e.Id = ?
   AND e.Status = ?
   AND e.ClaimToken = ?`,
			terminalStatus, now,
			terminalStatus, statusSkipped,
			terminalStatus, statusSkipped,
			now, eventID, statusProcessing, claimToken)
		if err != nil {
			return false, err
		}
		return workItemBridgeRowsAffected(res)
	})
	if err == nil {
		recordConnectorEventFinalize("processed", "ok")
		return nil
	}
	if errors.Is(err, ErrEventAlreadyTerminal) {
		recordConnectorEventFinalize("processed", "already_terminal")
		return err
	}
	if errors.Is(err, ErrLeaseLost) {
		recordConnectorEventFinalize("processed", "lease_lost")
		return err
	}
	recordConnectorEventFinalize("processed", "error")
	return err
}

// terminalizeLegacyEvent preserves the Event-only compatibility finalizer while
// the durable mutation guard excludes every WorkItem-owned generation. The
// Event lock and guarded UPDATE remain in one transaction, so a concurrent
// ownership cutover makes the legacy acknowledgement lose its lease.
func (s *Service) terminalizeLegacyEvent(ctx context.Context, eventID, claimToken string, updateEvent func(*sql.Tx) (bool, error)) error {
	tx, cleanup, err := sqltx.BeginPessimisticReadCommittedTx(ctx, s.db)
	if err != nil {
		return err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	var currentStatus string
	var currentClaimToken sql.NullString
	err = tx.QueryRowContext(ctx, `
SELECT Status, ClaimToken
  FROM _System_ConnectorEvent
 WHERE Id = ?
 FOR UPDATE`, eventID).Scan(&currentStatus, &currentClaimToken)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrEventNotFound
	}
	if err != nil {
		return err
	}
	if currentStatus == statusProcessed || currentStatus == statusSkipped || currentStatus == statusDead {
		// The caller's preflight distinguished an already-terminal Event from
		// this concurrent terminalization, which historically reported a lost
		// lease through the conditional UPDATE.
		return ErrLeaseLost
	}
	if currentStatus != statusProcessing || !currentClaimToken.Valid || currentClaimToken.String != claimToken {
		return ErrLeaseLost
	}
	if s.afterLegacyTerminalEventLock != nil {
		s.afterLegacyTerminalEventLock()
	}

	updated, err := updateEvent(tx)
	if err != nil {
		return err
	}
	if !updated {
		return ErrLeaseLost
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	committed = true
	return nil
}

// MarkFailed increments RetryCount.  If RetryCount reaches the threshold the
// row transitions to dead (DLQ); otherwise it becomes failed (re-claimable).
//
// claimToken must match the ClaimToken column written by ClaimNext for this
// specific lease grant.  ClaimToken is a fresh nonce minted per successful
// ClaimNext call and is returned in ConnectorEvent.ClaimToken.  Using a
// per-claim token instead of workerID prevents a restarted worker (same ID)
// or a replica sharing a configured workerID from accidentally failing another
// worker's in-flight row.
//
// Calling MarkFailed on a row already in processed, skipped, or dead returns
// ErrEventAlreadyTerminal.
//
// Does not gate on pack activation state — once an event is enqueued, drain/finalize
// must work during pack removing/inactive transitions.
func (s *Service) MarkFailed(ctx context.Context, eventID, claimToken, errorMsg string) error {
	_, currentStatus, err := s.loadNamespaceAndStatus(ctx, eventID)
	if err != nil {
		recordConnectorEventFinalize("failed", "error")
		return err
	}
	if currentStatus == statusProcessed || currentStatus == statusSkipped || currentStatus == statusDead {
		recordConnectorEventFinalize("failed", "already_terminal")
		return ErrEventAlreadyTerminal
	}

	// Load current retry count.
	var retryCount int
	if err := s.db.QueryRowContext(ctx, `
SELECT RetryCount FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&retryCount); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			recordConnectorEventFinalize("failed", "not_found")
			return ErrEventNotFound
		}
		recordConnectorEventFinalize("failed", "error")
		return err
	}

	newRetryCount := retryCount + 1
	newStatus := statusFailed
	if newRetryCount >= s.retryThreshold {
		newStatus = statusDead
	}

	now := s.now()
	if newStatus == statusDead {
		err := s.terminalizeLegacyEvent(ctx, eventID, claimToken, func(tx *sql.Tx) (bool, error) {
			res, err := s.execLegacyEventMutation(ctx, tx, `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       RetryCount = ?,
       LastError = ?,
       LastErrorAt = ?,
       NextAttemptAt = ?,
       LastModifiedDate = ?
 WHERE e.Id = ?
   AND e.Status = ?
   AND e.ClaimToken = ?`,
				statusDead, newRetryCount, errorMsg, now, nil, now, eventID, statusProcessing, claimToken)
			if err != nil {
				return false, err
			}
			return workItemBridgeRowsAffected(res)
		})
		if err != nil {
			if errors.Is(err, ErrEventAlreadyTerminal) {
				recordConnectorEventFinalize("failed", "already_terminal")
			} else if errors.Is(err, ErrLeaseLost) {
				recordConnectorEventFinalize("failed", "lease_lost")
			} else {
				recordConnectorEventFinalize("failed", "error")
			}
			return err
		}
		recordConnectorEventFinalize("dead", "ok")
		return nil
	}

	var nextAttemptAt interface{}
	nextAttemptAt = now.Add(claimpolicy.FailureRetryDelay(newRetryCount))
	res, err := s.execLegacyEventMutation(ctx, s.db, `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       RetryCount = ?,
       LastError = ?,
       LastErrorAt = ?,
       NextAttemptAt = ?,
       LastModifiedDate = ?
 WHERE e.Id = ?
   AND e.Status = ?
   AND e.ClaimToken = ?`,
		newStatus, newRetryCount, errorMsg, now, nextAttemptAt, now, eventID, statusProcessing, claimToken)
	if err != nil {
		recordConnectorEventFinalize("failed", "error")
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		recordConnectorEventFinalize("failed", "error")
		return err
	}
	if n == 0 {
		recordConnectorEventFinalize("failed", "lease_lost")
		return ErrLeaseLost
	}
	recordConnectorEventFinalize("retry", "ok")
	return nil
}

// connectorRetriableReclaimDelay is the fixed backoff applied when a flow signals
// a retriable condition (FlowRetriableFault -> FlowExecutionResult.IsRetriable)
// and the worker re-drives the event via ReleaseClaimForRetry WITHOUT consuming
// the genuine-failure retry budget.
//
// It is deliberately separate from claimpolicy.FailureRetryDelay. A retriable
// concurrent-loser is waiting on an in-flight winner that typically finalizes
// within seconds, so a short fixed cadence re-checks promptly while still
// preventing the bare-release hot-loop (claim -> retriable -> release -> claim).
//
// The total re-drive window needs no per-row attempt cap because it is bounded
// upstream by the producer. The first (and currently only) producer of a
// failed FlowRetriableFault is the connector-push create path introduced by
// OUT-1 (pingcap-inc/NexusCRM#3109, automation/connector_push_create_attempts.go):
// it emits FlowRetriableFault ONLY while the winner's 15-minute
// connectorPushProcessingLease is live; once that lease expires it instead
// returns a terminal (non-retriable) fault, so the next re-drive routes through
// failWithSyncState -> MarkFailed and the event drains/dead-letters on the
// normal RetryCount budget. This consumer-side change pairs with #3109; any
// future producer of FlowRetriableFault must uphold the same lease-bounded
// emission for the no-cap reasoning to hold.
const connectorRetriableReclaimDelay = 30 * time.Second

// loadByID fetches a single ConnectorEvent row by primary key.
func (s *Service) loadByID(ctx context.Context, eventID string) (*ConnectorEvent, error) {
	row := s.db.QueryRowContext(ctx, `
SELECT 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
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID)

	events, err := scanConnectorEvents(singleRowScanner{row})
	if err != nil {
		return nil, err
	}
	if len(events) == 0 {
		return nil, ErrEventNotFound
	}
	return &events[0], nil
}

// LoadEventForWorkItem loads the immutable connector event fact for an already
// claimed WorkItem. It intentionally does not claim or mutate the legacy event
// queue; WorkItem claim generations finalize through _System_ConnectorWorkItem.
func (s *Service) LoadEventForWorkItem(ctx context.Context, eventID string) (*ConnectorEvent, error) {
	ev, err := s.loadByID(ctx, eventID)
	if err != nil {
		return nil, err
	}
	var nextAttemptAt sql.NullTime
	err = s.db.QueryRowContext(ctx, `SELECT NextAttemptAt FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&nextAttemptAt)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrEventNotFound
	}
	if err != nil {
		return nil, err
	}
	if nextAttemptAt.Valid {
		t := nextAttemptAt.Time
		ev.NextAttemptAt = &t
	}
	return ev, nil
}

// LoadEventsForWorkItems hydrates a scheduler candidate window without the
// per-candidate two-query loop of LoadEventForWorkItem.
func (s *Service) LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*ConnectorEvent, error) {
	unique := make([]string, 0, len(eventIDs))
	seen := make(map[string]struct{}, len(eventIDs))
	for _, eventID := range eventIDs {
		eventID = strings.TrimSpace(eventID)
		if eventID == "" {
			continue
		}
		if _, ok := seen[eventID]; ok {
			continue
		}
		seen[eventID] = struct{}{}
		unique = append(unique, eventID)
	}
	result := make(map[string]*ConnectorEvent, len(unique))
	if len(unique) == 0 {
		return result, nil
	}
	args := make([]any, len(unique))
	for i := range unique {
		args[i] = unique[i]
	}
	rows, err := s.db.QueryContext(ctx, `
SELECT 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
  FROM _System_ConnectorEvent
 WHERE Id IN (`+sqlPlaceholders(len(unique))+`)`, args...)
	if err != nil {
		return nil, err
	}
	events, err := scanConnectorEvents(rows)
	rows.Close()
	if err != nil {
		return nil, err
	}
	for i := range events {
		ev := events[i]
		result[ev.ID] = &ev
	}
	nextRows, err := s.db.QueryContext(ctx, `
SELECT Id, NextAttemptAt
  FROM _System_ConnectorEvent
 WHERE Id IN (`+sqlPlaceholders(len(unique))+`)
   AND NextAttemptAt IS NOT NULL`, args...)
	if err != nil {
		return nil, err
	}
	defer nextRows.Close()
	for nextRows.Next() {
		var eventID string
		var nextAttemptAt time.Time
		if err := nextRows.Scan(&eventID, &nextAttemptAt); err != nil {
			return nil, err
		}
		if ev := result[eventID]; ev != nil {
			ev.NextAttemptAt = &nextAttemptAt
		}
	}
	if err := nextRows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

// loadNamespaceAndStatus fetches only the Namespace and Status columns for an
// event; used by MarkProcessed/MarkFailed before the heavier pack check.
func (s *Service) loadNamespaceAndStatus(ctx context.Context, eventID string) (namespace, status string, err error) {
	err = s.db.QueryRowContext(ctx, `
SELECT Namespace, Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&namespace, &status)
	if errors.Is(err, sql.ErrNoRows) {
		return "", "", ErrEventNotFound
	}
	return namespace, status, err
}

// nullableJSON marshals payload to a byte slice suitable for a JSON column,
// returning nil if payload is nil or empty.
func nullableJSON(payload json.RawMessage) interface{} {
	if len(payload) == 0 {
		return nil
	}
	return []byte(payload)
}

func nullableTime(value sql.NullTime) interface{} {
	if !value.Valid || value.Time.IsZero() {
		return nil
	}
	return value.Time.UTC()
}

type sourceLane struct {
	ObjectKind string
	SourceID   string
	Key        string
}

func sourceLaneFromPayload(payload json.RawMessage) sourceLane {
	if len(payload) == 0 {
		return sourceLane{}
	}
	var envelope struct {
		ObjectKind string `json:"objectKind"`
		SourceID   string `json:"sourceID"`
	}
	if err := json.Unmarshal(payload, &envelope); err != nil {
		return sourceLane{}
	}
	return sourceLane{
		ObjectKind: envelope.ObjectKind,
		SourceID:   envelope.SourceID,
		Key:        sourceLaneKeyFor(envelope.ObjectKind, envelope.SourceID),
	}
}

func sourceLaneKeyFor(objectKind, sourceID string) string {
	if objectKind == "" || sourceID == "" {
		return ""
	}
	seed := fmt.Sprintf("%d:%s:%d:%s", len(objectKind), objectKind, len(sourceID), sourceID)
	sum := sha256.Sum256([]byte(seed))
	return hex.EncodeToString(sum[:])
}

// rowQuerier is satisfied by both *sql.DB and *sql.Tx, allowing lookupExisting
// to run on either a non-transactional connection or an in-flight transaction.
type rowQuerier interface {
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

// rowScanner is satisfied by both *sql.Rows and the singleRowScanner adapter.
type rowScanner interface {
	Scan(dest ...interface{}) error
}

// singleRowScanner wraps *sql.Row to implement an interface compatible with
// *sql.Rows so scanConnectorEvents can work with both.
type singleRowScanner struct{ row *sql.Row }

func (s singleRowScanner) Scan(dest ...interface{}) error { return s.row.Scan(dest...) }

// scanConnectorEvents scans a result set (or single-row adapter) into a slice.
func scanConnectorEvents(scanner interface{}) ([]ConnectorEvent, error) {
	switch v := scanner.(type) {
	case *sql.Rows:
		var events []ConnectorEvent
		for v.Next() {
			ev, err := scanOne(v)
			if err != nil {
				return nil, err
			}
			events = append(events, ev)
		}
		return events, v.Err()
	case singleRowScanner:
		ev, err := scanOne(v)
		if errors.Is(err, sql.ErrNoRows) {
			return nil, nil
		}
		if err != nil {
			return nil, err
		}
		return []ConnectorEvent{ev}, nil
	default:
		panic("unsupported scanner type")
	}
}

func scanOne(s rowScanner) (ConnectorEvent, error) {
	return scanOneWithLeading(s)
}

func scanOneWithLeading(s rowScanner, leading ...interface{}) (ConnectorEvent, error) {
	var ev ConnectorEvent
	var queueClass sql.NullString
	var payload []byte
	var lastError sql.NullString
	var lastErrorAt sql.NullTime
	var claimedBy sql.NullString
	var claimedAt sql.NullTime
	var claimToken sql.NullString
	var processedAt sql.NullTime
	var replayedAt sql.NullTime
	var replayOfID sql.NullString
	var routeVersion sql.NullInt64
	var companionPackInstallID sql.NullString
	var companionFlowVersionID sql.NullString
	var sourceLaneKey sql.NullString
	var deliveryID sql.NullString

	destinations := append(leading,
		&ev.ID, &ev.Namespace, &ev.InstalledPackID, &ev.EventKind, &ev.IdempotencyKey, &ev.Status,
		&queueClass, &payload, &ev.RetryCount, &lastError, &lastErrorAt,
		&claimedBy, &claimedAt, &claimToken, &processedAt, &replayedAt, &replayOfID,
		&routeVersion, &companionPackInstallID, &companionFlowVersionID,
		&sourceLaneKey, &deliveryID, &ev.DeliverySequence, &ev.CreatedDate, &ev.LastModifiedDate,
	)
	err := s.Scan(destinations...)
	if err != nil {
		return ConnectorEvent{}, err
	}
	ev.Payload = json.RawMessage(payload)
	if queueClass.Valid {
		ev.QueueClass = strings.TrimSpace(queueClass.String)
	}
	if ev.QueueClass == "" {
		ev.QueueClass = connector_inbound_routing.DefaultQueueClass
	}
	ev.RouteVersion = routeVersion
	if companionPackInstallID.Valid {
		ev.CompanionPackInstallID = strings.TrimSpace(companionPackInstallID.String)
	}
	if companionFlowVersionID.Valid {
		ev.CompanionFlowVersionID = strings.TrimSpace(companionFlowVersionID.String)
	}
	if sourceLaneKey.Valid {
		ev.SourceLaneKey = strings.TrimSpace(sourceLaneKey.String)
	}
	if lastError.Valid {
		ev.LastError = lastError.String
	}
	if lastErrorAt.Valid {
		t := lastErrorAt.Time
		ev.LastErrorAt = &t
	}
	if claimedBy.Valid {
		ev.ClaimedBy = claimedBy.String
	}
	if claimedAt.Valid {
		t := claimedAt.Time
		ev.ClaimedAt = &t
	}
	if claimToken.Valid {
		ev.ClaimToken = claimToken.String
	}
	if processedAt.Valid {
		t := processedAt.Time
		ev.ProcessedAt = &t
	}
	if replayedAt.Valid {
		t := replayedAt.Time
		ev.ReplayedAt = &t
	}
	if replayOfID.Valid {
		ev.ReplayOfID = replayOfID.String
	}
	if deliveryID.Valid {
		ev.DeliveryId = deliveryID.String
	}
	return ev, nil
}