package connector_workqueue

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"hash/fnv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events/claimpolicy"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/id"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

type Repository struct {
	db    queryExecer
	idgen func() string
	now   func() time.Time
}

type queryExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

type sourceProjectionWorkItemRow struct {
	workID                         string
	eventID                        string
	installedPackID                string
	workType                       string
	sourceLaneKey                  string
	state                          string
	projectionKey                  string
	activeKey                      string
	inputStartOrdinal, laneOrdinal int64
	attempt                        int
}

const claimWorkItemCandidateWindow = 1024
const lanePromotionRetryAttempts = 4

func NewRepository(db *sql.DB, idgen func() string, now func() time.Time) *Repository {
	if idgen == nil {
		idgen = id.Generate
	}
	if now == nil {
		now = time.Now
	}
	return &Repository{db: db, idgen: idgen, now: now}
}

func (r *Repository) WithTx(tx *sql.Tx) *Repository {
	if r == nil {
		return nil
	}
	return &Repository{db: tx, idgen: r.idgen, now: r.now}
}

// UsesWorkItemRowCASClaimAuthority reports that this repository's WorkItem row
// transition is the sole distributed claim authority. The inbound worker uses
// this capability to skip legacy SchedulerLease hooks.
func (r *Repository) UsesWorkItemRowCASClaimAuthority() bool {
	return r != nil
}

// UsesWorkItemExecutorStateAuthority reports that this repository owns the
// mutable executor lifecycle for claimed WorkItem generations. Event rows are
// immutable execution input on this path.
func (r *Repository) UsesWorkItemExecutorStateAuthority() bool {
	return r != nil
}

// HonorsClaimWorkTypes declares that every shared-queue selection surface on
// Repository applies the request-local executor WorkType scope before limits.
func (r *Repository) HonorsClaimWorkTypes() bool {
	return r != nil
}

func (r *Repository) CreateWorkItem(ctx context.Context, in CreateWorkItemInput) (*WorkItem, bool, error) {
	workID, created, err := r.materializeOrdinaryWorkItem(ctx, in, ordinaryWorkItemDirect)
	if err != nil {
		return nil, false, err
	}
	if created {
		work, loadErr := r.GetWorkItem(ctx, workID)
		return work, true, loadErr
	}
	work, err := r.getWorkItemCurrentForUpdate(ctx, workID)
	return work, created, err
}

func (r *Repository) EnsureWorkItem(ctx context.Context, in CreateWorkItemInput) (bool, error) {
	_, created, err := r.materializeOrdinaryWorkItem(ctx, in, ordinaryWorkItemReconcileLane)
	return created, err
}

// EnsureReplayWorkItem materializes one ordinary WorkItem generation with
// immutable lineage to the terminal generation that was replayed.
func (r *Repository) EnsureReplayWorkItem(ctx context.Context, in ReplayWorkItemInput) (bool, error) {
	return r.ensureWorkItem(ctx, in.WorkItem, strings.TrimSpace(in.ReplayOfWorkID))
}

func (r *Repository) ensureWorkItem(ctx context.Context, in CreateWorkItemInput, replayOfWorkID string) (bool, error) {
	in = r.defaultCreateWorkItemInput(in)
	if err := validateCreateWorkItemInput(in); err != nil {
		return false, err
	}
	if replayOfWorkID != "" && replayOfWorkID == in.WorkID {
		return false, ErrInvalidWorkItem
	}
	reconcileLane := isReadyWorkState(in.State) && workItemLaneKey(in) != ""
	if reconcileLane {
		in.State = blockedStateForReadyState(in.State)
	}
	err := r.insertWorkItem(ctx, in, replayOfWorkID)
	if err != nil {
		if isDuplicateKey(err) {
			existing, loadErr := r.GetWorkItemByEvent(ctx, in.EventID, in.WorkType)
			if loadErr != nil {
				return false, loadErr
			}
			if existing == nil {
				return false, err
			}
			return false, nil
		}
		return false, err
	}
	if reconcileLane {
		if err := r.reconcileLaneReadyState(ctx, in); err != nil {
			return false, err
		}
	}
	return true, nil
}

func (r *Repository) EnsureWorkItemDeferredLaneReconcile(ctx context.Context, in CreateWorkItemInput) (bool, error) {
	_, created, err := r.materializeOrdinaryWorkItem(ctx, in, ordinaryWorkItemDeferLane)
	return created, err
}

// EnsureSourceProjectionWorkItem verifies durable coverage for a duplicate
// Event. It never creates, extends, or reopens an executable generation.
func (r *Repository) EnsureSourceProjectionWorkItem(ctx context.Context, in CreateWorkItemInput, deferLaneReconcile bool) (created, selected bool, err error) {
	in = r.defaultCreateWorkItemInput(in)
	if err := validateCreateWorkItemInput(in); err != nil || !IsSourceProjectionWorkID(in.WorkID) || strings.TrimSpace(in.SourceLaneKey) == "" || in.LaneOrdinal < 0 {
		return false, false, ErrInvalidWorkItem
	}
	_, err = r.sourceProjectionCoversEvent(ctx, in)
	return false, false, err
}

// AdmitSourceProjectionWorkItem assigns only newly inserted inputs from the
// admitted pack transaction. Its lower bound is immutable for this generation.
func (r *Repository) AdmitSourceProjectionWorkItem(ctx context.Context, in SourceProjectionWorkItemInput, deferLaneReconcile bool) (bool, bool, error) {
	return r.ensureSourceProjectionWorkItem(ctx, in.WorkItem, deferLaneReconcile, "", false, in.InputStartOrdinal)
}

// EnsureSourceProjectionReplayWorkItem advances a source projection with
// replay semantics and optional immutable lineage to a terminal WorkItem
// generation. Event-only compatibility replays have no WorkItem lineage, but
// still fail closed when another nonterminal generation owns the projection.
func (r *Repository) EnsureSourceProjectionReplayWorkItem(ctx context.Context, in ReplayWorkItemInput, deferLaneReconcile bool) (created, selected bool, err error) {
	// Explicit replay is a cold operation and must not partially use a rolling
	// schema that cannot persist lineage or terminal generation history.
	if err := r.VerifyWorkItemSchema(ctx); err != nil {
		return false, false, err
	}
	return r.ensureSourceProjectionWorkItem(ctx, in.WorkItem, deferLaneReconcile, strings.TrimSpace(in.ReplayOfWorkID), true, in.WorkItem.LaneOrdinal)
}

func (r *Repository) ensureSourceProjectionWorkItem(ctx context.Context, in CreateWorkItemInput, deferLaneReconcile bool, replayOfWorkID string, replay bool, inputStartOrdinal int64) (created, selected bool, err error) {
	results, err := r.admitProjectionBatch(ctx, []SourceProjectionWorkItemInput{{WorkItem: in, InputStartOrdinal: inputStartOrdinal}}, deferLaneReconcile, replayOfWorkID, replay)
	if err != nil {
		return false, false, err
	}
	return results[0], true, nil
}

func (r *Repository) loadQueuedSourceProjectionSuccessorForUpdate(ctx context.Context, projectionKey, activeWorkID string) (*sourceProjectionWorkItemRow, error) {
	rows, err := r.db.QueryContext(ctx, `
SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE(SourceLaneKey, ''), State,
       COALESCE(ProjectionKey, ''), COALESCE(ProjectionActiveKey, ''), COALESCE(InputStartOrdinal,0), LaneOrdinal, Attempt
  FROM _System_ConnectorWorkItem
 WHERE ProjectionKey = ?
   AND ProjectionActiveKey IS NULL
   AND WorkId <> ?
   AND State IN (?, ?, ?, ?)
 ORDER BY CreatedDate DESC, WorkId DESC
 LIMIT 2
 FOR UPDATE`,
		projectionKey, activeWorkID,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var successor *sourceProjectionWorkItemRow
	for rows.Next() {
		if successor != nil {
			return nil, ErrInvalidWorkItem
		}
		var row sourceProjectionWorkItemRow
		if err := rows.Scan(
			&row.workID, &row.eventID, &row.installedPackID, &row.workType,
			&row.sourceLaneKey, &row.state, &row.projectionKey, &row.activeKey, &row.inputStartOrdinal, &row.laneOrdinal, &row.attempt,
		); err != nil {
			return nil, err
		}
		successor = &row
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return successor, nil
}

func (r *Repository) loadActiveSourceProjectionWorkItemForUpdate(ctx context.Context, projectionKey string) (*sourceProjectionWorkItemRow, error) {
	candidate, err := r.loadSourceProjectionWorkItem(ctx, `
SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE(SourceLaneKey, ''), State,
       COALESCE(ProjectionKey, ''), COALESCE(ProjectionActiveKey, ''), COALESCE(InputStartOrdinal,0), LaneOrdinal, Attempt
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey = ?
 LIMIT 1`, projectionKey)
	if err != nil || candidate == nil {
		return candidate, err
	}
	return r.loadSourceProjectionWorkItemForUpdate(ctx, `
SELECT WorkId, EventId, InstalledPackId, WorkType, COALESCE(SourceLaneKey, ''), State,
       COALESCE(ProjectionKey, ''), COALESCE(ProjectionActiveKey, ''), COALESCE(InputStartOrdinal,0), LaneOrdinal, Attempt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND ProjectionActiveKey = ?
 LIMIT 1
 FOR UPDATE`, candidate.workID, projectionKey)
}

func (r *Repository) loadSourceProjectionWorkItemForUpdate(ctx context.Context, query string, args ...any) (*sourceProjectionWorkItemRow, error) {
	return r.loadSourceProjectionWorkItem(ctx, query, args...)
}

func (r *Repository) loadSourceProjectionWorkItem(ctx context.Context, query string, args ...any) (*sourceProjectionWorkItemRow, error) {
	var row sourceProjectionWorkItemRow
	err := r.db.QueryRowContext(ctx, query, args...).Scan(
		&row.workID, &row.eventID, &row.installedPackID, &row.workType,
		&row.sourceLaneKey, &row.state, &row.projectionKey, &row.activeKey, &row.inputStartOrdinal, &row.laneOrdinal, &row.attempt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &row, nil
}

func validateSourceProjectionWorkItemRow(row *sourceProjectionWorkItemRow, in CreateWorkItemInput, projectionKey string) error {
	if row == nil ||
		strings.TrimSpace(row.installedPackID) != strings.TrimSpace(in.InstalledPackID) ||
		strings.TrimSpace(row.workType) != strings.TrimSpace(in.WorkType) ||
		strings.TrimSpace(row.sourceLaneKey) != strings.TrimSpace(in.SourceLaneKey) ||
		(row.projectionKey != "" && row.projectionKey != projectionKey) ||
		(row.activeKey != "" && row.activeKey != projectionKey) {
		return ErrInvalidWorkItem
	}
	return nil
}

func (r *Repository) nextSourceProjectionWorkID(projectionKey string) string {
	return projectionKey + ":" + r.idgen()
}

func (r *Repository) ReconcileLaneReadyState(ctx context.Context, in CreateWorkItemInput) error {
	in = r.defaultCreateWorkItemInput(in)
	if strings.TrimSpace(in.InstalledPackID) == "" || in.Generation < 0 {
		return ErrInvalidWorkItem
	}
	if workItemLaneKey(in) == "" {
		return nil
	}
	return r.reconcileLaneReadyState(ctx, in)
}

func (r *Repository) VerifyWorkItemSchema(ctx context.Context) error {
	rows, err := r.db.QueryContext(ctx, `
SELECT LaneOrdinal, InputStartOrdinal, OwnershipEpoch, FailureCount, FailureClass, FailureSummary,
       LastFailureAt, TerminalAt, ReplayedAt, ReplayOfWorkId,
       ProjectionKey, ProjectionActiveKey,
       DependencyKind, DependencyObjectAPIName, DependencyRecordId,
       DependencyKeyHash, DependencyConnectorOwnerKey,
       DependencyConnectorNamespace, DependencyConnectorSourceObject,
       DependencyDestinationSlotId, DependencyConnectorRemoteId,
	   DependencyBlockedAt, DependencyProbeAfter
  FROM _System_ConnectorWorkItem
 WHERE 1 = 0`)
	if err != nil {
		return err
	}
	return rows.Close()
}

func (r *Repository) insertWorkItem(ctx context.Context, in CreateWorkItemInput, replayOfWorkID string) error {
	if replayOfWorkID != "" {
		_, err := r.db.ExecContext(ctx, `
	INSERT INTO _System_ConnectorWorkItem
	       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
	        Attempt, SourceLaneKey, TargetMutationLaneKey, LaneOrdinal, Generation,
	        ReplayOfWorkId, CreatedDate, LastModifiedDate)
	VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?)`,
			in.WorkID, in.EventID, in.InstalledPackID, in.WorkType, in.QueueClass, in.State, in.RunAfter,
			in.SourceLaneKey, in.TargetMutationLaneKey, in.LaneOrdinal, in.Generation,
			replayOfWorkID, in.Now, in.Now)
		return err
	}
	_, err := r.db.ExecContext(ctx, `
	INSERT INTO _System_ConnectorWorkItem
	       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
	        Attempt, SourceLaneKey, TargetMutationLaneKey, LaneOrdinal, Generation, CreatedDate, LastModifiedDate)
	VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)`,
		in.WorkID, in.EventID, in.InstalledPackID, in.WorkType, in.QueueClass, in.State, in.RunAfter,
		in.SourceLaneKey, in.TargetMutationLaneKey, in.LaneOrdinal, in.Generation, in.Now, in.Now)
	return err
}

func (r *Repository) GetWorkItem(ctx context.Context, workID string) (*WorkItem, error) {
	return r.scanWorkItem(r.db.QueryRowContext(ctx, selectWorkItemSQL+` WHERE WorkId = ?`, workID))
}

func (r *Repository) getWorkItemCurrentForUpdate(ctx context.Context, workID string) (*WorkItem, error) {
	return r.scanWorkItem(r.db.QueryRowContext(ctx, selectWorkItemSQL+` WHERE WorkId = ? FOR UPDATE`, workID))
}

func (r *Repository) getWorkItems(ctx context.Context, workIDs []string) (map[string]*WorkItem, error) {
	if len(workIDs) == 0 {
		return map[string]*WorkItem{}, nil
	}
	placeholders := make([]string, len(workIDs))
	args := make([]any, len(workIDs))
	for i, workID := range workIDs {
		placeholders[i] = "?"
		args[i] = workID
	}
	rows, err := r.db.QueryContext(ctx, selectWorkItemSQL+` WHERE WorkId IN (`+strings.Join(placeholders, ",")+`)`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	works := make(map[string]*WorkItem, len(workIDs))
	for rows.Next() {
		work, err := r.scanWorkItem(rows)
		if err != nil {
			return nil, err
		}
		if work != nil {
			works[work.WorkID] = work
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return works, nil
}

func (r *Repository) GetWorkItemByEvent(ctx context.Context, eventID, workType string) (*WorkItem, error) {
	return r.scanWorkItem(r.db.QueryRowContext(ctx, selectWorkItemSQL+` WHERE EventId = ? AND WorkType = ?`, eventID, workType))
}

func (r *Repository) ClaimWorkItem(ctx context.Context, workID, ownerID, leaseToken string, leaseUntil, now time.Time) (bool, error) {
	if strings.TrimSpace(workID) == "" || strings.TrimSpace(ownerID) == "" || strings.TrimSpace(leaseToken) == "" || !leaseUntil.After(now) {
		return false, ErrInvalidWorkItem
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       Attempt = Attempt + 1,
       LeaseOwner = ?,
       LeaseToken = ?,
       LeaseUntil = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State IN (?, ?)
   AND RunAfter <= ?`,
		WorkStateProcessing, ownerID, leaseToken, leaseUntil, now, workID,
		WorkStatePending, WorkStateRetry, now)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

func (r *Repository) ClaimNextWorkItem(ctx context.Context, in ClaimWorkItemInput) (*WorkItem, bool, error) {
	in = normalizeClaimWorkItemInput(in)
	if err := validateClaimWorkItemInput(in); err != nil {
		return nil, false, err
	}
	candidates, err := r.findClaimableWorkItems(ctx, in, claimWorkItemCandidateWindow)
	if err != nil {
		return nil, false, err
	}
	if len(candidates) == 0 && !in.StaleOnly {
		if _, err := r.promoteBlockedLaneHeadsForClaim(ctx, in, claimWorkItemCandidateWindow); err != nil {
			return nil, false, err
		}
		candidates, err = r.findClaimableWorkItems(ctx, in, claimWorkItemCandidateWindow)
		if err != nil || len(candidates) == 0 {
			return nil, false, err
		}
	}
	for _, candidate := range orderWorkItemCandidates(ctx, candidates, in) {
		claimed, err := r.claimSelectedWorkItem(ctx, candidate.workID, in)
		if err != nil {
			return nil, false, err
		}
		if !claimed {
			if _, err := r.convergeLegacyEventTerminalWorkItem(ctx, candidate.workID, in.Now); err != nil {
				return nil, false, err
			}
			continue
		}
		work, err := r.GetWorkItem(ctx, candidate.workID)
		if err != nil {
			return nil, false, err
		}
		return work, work != nil, nil
	}
	if !in.StaleOnly && len(candidates) > 1 {
		if _, err := r.promoteBlockedLaneHeadsForClaim(ctx, in, claimWorkItemCandidateWindow); err != nil {
			return nil, false, err
		}
		candidates, err := r.findClaimableWorkItems(ctx, in, claimWorkItemCandidateWindow)
		if err != nil {
			return nil, false, err
		}
		for _, candidate := range orderWorkItemCandidates(ctx, candidates, in) {
			claimed, err := r.claimSelectedPromotedWorkItem(ctx, candidate.workID, in)
			if err != nil {
				return nil, false, err
			}
			if !claimed {
				if _, err := r.convergeLegacyEventTerminalWorkItem(ctx, candidate.workID, in.Now); err != nil {
					return nil, false, err
				}
				continue
			}
			work, err := r.GetWorkItem(ctx, candidate.workID)
			if err != nil {
				return nil, false, err
			}
			return work, work != nil, nil
		}
	}
	return nil, false, nil
}

func (r *Repository) PeekClaimableWorkItems(ctx context.Context, in ClaimWorkItemInput, limit int) ([]*WorkItem, error) {
	in = normalizeClaimWorkItemInput(in)
	if err := validateClaimWorkItemInput(in); err != nil {
		return nil, err
	}
	if limit <= 0 || limit > claimWorkItemCandidateWindow {
		limit = claimWorkItemCandidateWindow
	}
	workIDs, err := r.findClaimableLaneHeadWorkItems(ctx, in, limit)
	if err != nil {
		return nil, err
	}
	if len(workIDs) < limit && !in.StaleOnly {
		if _, err := r.promoteBlockedLaneHeadsForClaim(ctx, in, limit-len(workIDs)); err != nil {
			return nil, err
		}
		refreshedWorkIDs, err := r.findClaimableLaneHeadWorkItems(ctx, in, limit)
		if err != nil {
			return nil, err
		}
		if len(refreshedWorkIDs) > len(workIDs) {
			workIDs = refreshedWorkIDs
		}
	}
	worksByID, err := r.getWorkItems(ctx, workIDs)
	if err != nil {
		return nil, err
	}
	works := make([]*WorkItem, 0, len(worksByID))
	for _, workID := range workIDs {
		if work := worksByID[workID]; work != nil {
			works = append(works, work)
		}
	}
	return works, nil
}

func (r *Repository) ClaimWorkItemByEvent(ctx context.Context, in ClaimWorkItemByEventInput) (*WorkItem, bool, error) {
	return r.claimWorkItemByEvent(ctx, in, false)
}

// ClaimWorkItemByEventForCoalescing claims one ordinary successor under both
// the candidate's state and OwnershipEpoch. Both executor WorkTypes are
// inspected because compatibility data can predate the workflow-action type
// split; in.WorkType validates the current mapping, while the stored WorkItem
// type remains authoritative. Multiple claimable executor rows for one Event
// are ambiguous and fail closed. Callers must run this in the same
// transaction that completes the current WorkItem so a lost successor CAS can
// roll back the whole latest_by_source handoff.
func (r *Repository) ClaimWorkItemByEventForCoalescing(ctx context.Context, in ClaimWorkItemByEventInput, allowCrossSourceDelivery bool) (*WorkItem, bool, error) {
	in = normalizeClaimWorkItemByEventInput(in)
	if err := validateClaimWorkItemByEventInput(in); err != nil {
		return nil, false, err
	}
	if !claimWorkTypeAllowed(ctx, in.WorkType) {
		return nil, false, nil
	}
	workID, storedWorkType, ownershipEpoch, ok, err := r.findClaimableWorkItemByEventAtEpoch(ctx, in, allowCrossSourceDelivery)
	if err != nil || !ok {
		return nil, false, err
	}
	if !claimWorkTypeAllowed(ctx, storedWorkType) {
		return nil, false, nil
	}
	claimed, err := r.claimSelectedWorkItemAtEpoch(ctx, workID, ClaimWorkItemInput{
		InstalledPackID: in.InstalledPackID,
		Generation:      in.Generation,
		OwnerID:         in.OwnerID,
		FencingToken:    in.FencingToken,
		LeaseToken:      in.LeaseToken,
		LeaseUntil:      in.LeaseUntil,
		Now:             in.Now,
	}, ownershipEpoch, allowCrossSourceDelivery)
	if err != nil || !claimed {
		return nil, claimed, err
	}
	work, err := r.GetWorkItem(ctx, workID)
	if err != nil {
		return nil, false, err
	}
	return work, work != nil, nil
}

// ClaimStaleWorkItemByEvent reclaims an expired processing WorkItem. It keeps
// the public fresh-claim input source-compatible while making recovery mode
// explicit at the call site.
func (r *Repository) ClaimStaleWorkItemByEvent(ctx context.Context, in ClaimWorkItemByEventInput) (*WorkItem, bool, error) {
	return r.claimWorkItemByEvent(ctx, in, true)
}

func (r *Repository) claimWorkItemByEvent(ctx context.Context, in ClaimWorkItemByEventInput, staleOnly bool) (*WorkItem, bool, error) {
	in = normalizeClaimWorkItemByEventInput(in)
	if err := validateClaimWorkItemByEventInput(in); err != nil {
		return nil, false, err
	}
	if !claimWorkTypeAllowed(ctx, in.WorkType) {
		return nil, false, nil
	}
	workID, ok, err := r.findClaimableWorkItemByEvent(ctx, in, staleOnly)
	if err != nil || !ok {
		return nil, false, err
	}
	claimed, err := r.claimSelectedWorkItem(ctx, workID, ClaimWorkItemInput{
		InstalledPackID: in.InstalledPackID,
		Generation:      in.Generation,
		OwnerID:         in.OwnerID,
		FencingToken:    in.FencingToken,
		LeaseToken:      in.LeaseToken,
		LeaseUntil:      in.LeaseUntil,
		StaleOnly:       staleOnly,
		Now:             in.Now,
	})
	if err != nil {
		return nil, false, err
	}
	if !claimed {
		if _, err := r.convergeLegacyEventTerminalWorkItem(ctx, workID, in.Now); err != nil {
			return nil, false, err
		}
		return nil, false, nil
	}
	work, err := r.GetWorkItem(ctx, workID)
	if err != nil {
		return nil, false, err
	}
	return work, work != nil, nil
}

func (r *Repository) HasNonTerminalLegacyEventWithoutWorkItem(ctx context.Context, in LegacyEventMirrorReadinessInput) (bool, error) {
	in = normalizeLegacyEventMirrorReadinessInput(in)
	if err := validateLegacyEventMirrorReadinessInput(in); err != nil {
		return false, err
	}
	row := r.db.QueryRowContext(ctx, `
SELECT e.Id
  FROM _System_ConnectorEvent e
  LEFT JOIN _System_ConnectorWorkItem w
    ON w.EventId = e.Id
   AND w.WorkType IN (?, ?)
 WHERE e.InstalledPackId = ?
   AND e.Status NOT IN ('processed', 'skipped', 'dead')
   AND w.WorkId IS NULL
   AND NOT EXISTS (
       SELECT 1
         FROM _System_ConnectorWorkItem projection
         JOIN _System_ConnectorEvent selected ON selected.Id = projection.EventId
        WHERE projection.InstalledPackId = e.InstalledPackId
          AND projection.WorkType IN (?, ?)
          AND (
              NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NOT NULL
              OR (
                  NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NULL
                  AND projection.WorkId LIKE 'csp:%'
                  AND projection.State NOT IN ('processed', 'skipped', 'dead')
              )
          )
          AND COALESCE(projection.SourceLaneKey, '') = COALESCE(e.SourceLaneKey, '')
          AND selected.InstalledPackId = e.InstalledPackId
          AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(e.EventKind))
          AND COALESCE(selected.RouteVersion, 0) = COALESCE(e.RouteVersion, 0)
          AND COALESCE(selected.CompanionPackInstallId, '') = COALESCE(e.CompanionPackInstallId, '')
          AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE(e.CompanionFlowVersionId, '')
          AND e.EnqueueOrdinal BETWEEN projection.InputStartOrdinal AND projection.LaneOrdinal
   )
 LIMIT 1`,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		in.InstalledPackID,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply)
	var eventID string
	err := row.Scan(&eventID)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

func (r *Repository) RenewWorkItemLease(ctx context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error) {
	if strings.TrimSpace(workID) == "" || strings.TrimSpace(leaseToken) == "" || !leaseUntil.After(now) {
		return false, ErrInvalidWorkItem
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET LeaseUntil = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
		leaseUntil, now, workID, WorkStateProcessing, leaseToken)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

func (r *Repository) RetryWorkItem(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error) {
	if strings.TrimSpace(workID) == "" || strings.TrimSpace(leaseToken) == "" || runAfter.IsZero() || now.IsZero() {
		return false, ErrInvalidWorkItem
	}
	if _, ok := WorkItemAttemptOutcomeFromContext(ctx); ok {
		return r.retryWorkItemWithAttempt(ctx, workID, leaseToken, runAfter, now)
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
		WorkStateRetry, runAfter, now, workID, WorkStateProcessing, leaseToken)
	if err != nil {
		return false, err
	}
	retried, err := rowsAffected(res)
	if err != nil || !retried {
		return retried, err
	}
	if err := r.promoteNextLaneHeadAfterWithRetry(ctx, workID, now); err != nil {
		return false, err
	}
	return true, nil
}

func (r *Repository) retryWorkItemWithAttempt(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error) {
	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticTx(ctx, db)
		if err != nil {
			return false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}

	var eventID string
	var attempt int
	err := q.QueryRowContext(ctx, `
SELECT EventId, Attempt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?
 LIMIT 1
 FOR UPDATE`, workID, WorkStateProcessing, leaseToken).Scan(&eventID, &attempt)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
		WorkStateRetry, runAfter, now, workID, WorkStateProcessing, leaseToken)
	if err != nil {
		return false, err
	}
	retried, err := rowsAffected(res)
	if err != nil || !retried {
		return retried, err
	}
	if err := appendWorkItemAttempt(ctx, q, workID, attempt, eventID, WorkStateRetry, now); err != nil {
		return false, err
	}
	transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}
	if err := transitionRepo.promoteNextLaneHeadAfterWithRetry(ctx, workID, now); err != nil {
		return false, err
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return false, err
		}
	}
	return true, nil
}

// FailWorkItem consumes one genuine-failure budget unit under the live row
// lease. Attempt remains the claim/reclaim counter; FailureCount alone chooses
// retry versus dead.
func (r *Repository) FailWorkItem(ctx context.Context, in FailWorkItemInput) (*FailWorkItemResult, bool, error) {
	return r.failWorkItem(ctx, in, false)
}

// FailWorkItemDeferredLaneReconcile lets a cross-table caller update its
// compatibility mirror before lane promotion observes the new state.
func (r *Repository) FailWorkItemDeferredLaneReconcile(ctx context.Context, in FailWorkItemInput) (*FailWorkItemResult, bool, error) {
	return r.failWorkItem(ctx, in, true)
}

func (r *Repository) failWorkItem(ctx context.Context, in FailWorkItemInput, deferLaneReconcile bool) (*FailWorkItemResult, bool, error) {
	in = normalizeFailWorkItemInput(in)
	if err := validateFailWorkItemInput(in); err != nil {
		return nil, false, err
	}

	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticTx(ctx, db)
		if err != nil {
			return nil, false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}
	transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}

	var eventID string
	var storedFailureCount int
	err := q.QueryRowContext(ctx, `
SELECT w.EventId, w.FailureCount
  FROM _System_ConnectorWorkItem w
 WHERE w.WorkId = ?
   AND w.EventId = ?
   AND w.State = ?
   AND w.LeaseToken = ?
 LIMIT 1
 FOR UPDATE`,
		in.WorkID, in.EventID, WorkStateProcessing, in.LeaseToken).
		Scan(&eventID, &storedFailureCount)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	var attempt int
	if _, hasAttemptOutcome := WorkItemAttemptOutcomeFromContext(ctx); hasAttemptOutcome {
		if err := q.QueryRowContext(ctx, `
SELECT Attempt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND EventId = ?`, in.WorkID, eventID).Scan(&attempt); err != nil {
			return nil, false, err
		}
	}
	result := failWorkItemResult(in, storedFailureCount+1)
	if result.State == WorkStateDead && IsSourceProjectionWorkID(in.WorkID) {
		completed, err := transitionRepo.completeSourceProjectionWorkItemWithFailure(
			ctx,
			in.WorkID,
			in.LeaseToken,
			WorkStateDead,
			in.Now,
			&sourceProjectionFailureRecord{
				previousCount: storedFailureCount,
				count:         result.FailureCount,
				class:         in.FailureClass,
				summary:       in.FailureSummary,
			},
			deferLaneReconcile,
		)
		if err != nil || !completed {
			return nil, completed, err
		}
	} else {
		terminalAt := any(nil)
		if result.State == WorkStateDead {
			terminalAt = in.Now
		}
		res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       FailureCount = ?,
       FailureClass = ?,
       FailureSummary = ?,
       LastFailureAt = ?,
       TerminalAt = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND State = ?
   AND LeaseToken = ?
   AND FailureCount = ?`,
			result.State, result.RunAfter, result.FailureCount,
			in.FailureClass, in.FailureSummary, in.Now, terminalAt, in.Now,
			in.WorkID, eventID, WorkStateProcessing, in.LeaseToken, storedFailureCount)
		if err != nil {
			return nil, false, err
		}
		updated, err := rowsAffected(res)
		if err != nil || !updated {
			return nil, updated, err
		}
		if !deferLaneReconcile {
			if err := transitionRepo.promoteNextLaneHeadAfterWithRetry(ctx, in.WorkID, in.Now); err != nil {
				return nil, false, err
			}
		}
	}
	if err := appendWorkItemAttempt(ctx, q, in.WorkID, attempt, eventID, result.State, in.Now); err != nil {
		return nil, false, err
	}

	if tx != nil {
		if err := tx.Commit(); err != nil {
			return nil, false, err
		}
	}
	return result, true, nil
}

// ReconcileAfterWorkItemTransition runs lane promotion after any compatibility
// mirror in the caller's transaction has reached the same state.
func (r *Repository) ReconcileAfterWorkItemTransition(ctx context.Context, workID string, now time.Time) error {
	if strings.TrimSpace(workID) == "" || now.IsZero() {
		return ErrInvalidWorkItem
	}
	return r.promoteNextLaneHeadAfterWithRetry(ctx, workID, now)
}

func (r *Repository) CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	if strings.TrimSpace(workID) == "" || strings.TrimSpace(leaseToken) == "" || !isTerminalWorkState(terminalState) || now.IsZero() {
		return false, ErrInvalidTerminalState
	}
	if IsSourceProjectionWorkID(workID) {
		return r.completeSourceProjectionWorkItem(ctx, workID, leaseToken, terminalState, now)
	}
	if _, ok := WorkItemAttemptOutcomeFromContext(ctx); ok {
		return r.completeWorkItemWithAttempt(ctx, workID, leaseToken, terminalState, now)
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
		terminalState, now, now, workID, WorkStateProcessing, leaseToken)
	if isProjectionGenerationSchemaMissing(err) {
		res, err = r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
			terminalState, now, workID, WorkStateProcessing, leaseToken)
	}
	if err != nil {
		return false, err
	}
	completed, err := rowsAffected(res)
	if err != nil || !completed {
		return completed, err
	}
	if err := r.promoteNextLaneHeadAfterWithRetry(ctx, workID, now); err != nil {
		return false, err
	}
	return true, nil
}

func (r *Repository) completeWorkItemWithAttempt(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticTx(ctx, db)
		if err != nil {
			return false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}

	var eventID string
	var attempt int
	err := q.QueryRowContext(ctx, `
SELECT EventId, Attempt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?
 LIMIT 1
 FOR UPDATE`, workID, WorkStateProcessing, leaseToken).Scan(&eventID, &attempt)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
		terminalState, now, now, workID, WorkStateProcessing, leaseToken)
	if isProjectionGenerationSchemaMissing(err) {
		res, err = q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`,
			terminalState, now, workID, WorkStateProcessing, leaseToken)
	}
	if err != nil {
		return false, err
	}
	completed, err := rowsAffected(res)
	if err != nil || !completed {
		return completed, err
	}
	if err := appendWorkItemAttempt(ctx, q, workID, attempt, eventID, terminalState, now); err != nil {
		return false, err
	}
	transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}
	if err := transitionRepo.promoteNextLaneHeadAfterWithRetry(ctx, workID, now); err != nil {
		return false, err
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return false, err
		}
	}
	return true, nil
}

// CompleteExecutorWorkItemForCoalescing terminalizes one ordinary executor
// generation under its lease token and the OwnershipEpoch and WorkType observed
// while the transaction holds the row lock. Reading WorkType from the durable
// WorkItem preserves compatibility rows whose stored type differs from the type
// a current binary would derive from EventKind. Source-projection generations
// keep their dedicated terminal-generation handoff and are rejected here.
func (r *Repository) CompleteExecutorWorkItemForCoalescing(ctx context.Context, workID, eventID, leaseToken string, generation int64, now time.Time) (bool, error) {
	workID = strings.TrimSpace(workID)
	eventID = strings.TrimSpace(eventID)
	leaseToken = strings.TrimSpace(leaseToken)
	if workID == "" || eventID == "" || leaseToken == "" || generation < 0 || now.IsZero() ||
		IsSourceProjectionWorkID(workID) {
		return false, ErrInvalidTerminalState
	}

	var ownershipEpoch int64
	var workType string
	err := r.db.QueryRowContext(ctx, `
SELECT OwnershipEpoch, WorkType
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND EventId = ?
   AND Generation = ?
   AND State = ?
   AND LeaseToken = ?
   AND WorkId NOT LIKE ?
 LIMIT 1
 FOR UPDATE`,
		workID, eventID, generation, WorkStateProcessing, leaseToken,
		SourceProjectionWorkIDPrefix+"%",
	).Scan(&ownershipEpoch, &workType)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	workType = strings.TrimSpace(workType)
	if workType != WorkTypeConnectorEvent && workType != WorkTypeWorkflowActionApply {
		return false, ErrInvalidTerminalState
	}

	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND WorkType = ?
   AND Generation = ?
   AND State = ?
   AND LeaseToken = ?
   AND OwnershipEpoch = ?
   AND WorkId NOT LIKE ?`,
		WorkStateProcessed, now, now,
		workID, eventID, workType, generation, WorkStateProcessing, leaseToken,
		ownershipEpoch, SourceProjectionWorkIDPrefix+"%",
	)
	if isProjectionGenerationSchemaMissing(err) {
		res, err = r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND WorkType = ?
   AND Generation = ?
   AND State = ?
   AND LeaseToken = ?
   AND OwnershipEpoch = ?
   AND WorkId NOT LIKE ?`,
			WorkStateProcessed, now,
			workID, eventID, workType, generation, WorkStateProcessing, leaseToken,
			ownershipEpoch, SourceProjectionWorkIDPrefix+"%",
		)
	}
	if err != nil {
		return false, err
	}
	completed, err := rowsAffected(res)
	if err != nil || !completed {
		return completed, err
	}
	if err := r.promoteNextLaneHeadAfterWithRetry(ctx, workID, now); err != nil {
		return false, err
	}
	return true, nil
}

// PrepareEventMirrorForTerminalization is retained for source compatibility
// with pinned Event-first finalizers. WorkItem-owned generations can no longer
// be prepared or terminalized through that path, so an existing row is returned
// only as a durable fence with prepared=false.
//
// Deprecated: Event-first finalizers must treat any returned WorkID as lease
// loss. Event-only generations have no WorkItem row and remain on the guarded
// legacy fallback.
func (r *Repository) PrepareEventMirrorForTerminalization(ctx context.Context, eventID, leaseToken string, now time.Time) (string, bool, error) {
	if strings.TrimSpace(eventID) == "" || strings.TrimSpace(leaseToken) == "" || now.IsZero() {
		return "", false, ErrInvalidWorkItem
	}

	var workID string
	err := r.db.QueryRowContext(ctx, `
SELECT WorkId
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?
 ORDER BY WorkId
 LIMIT 1
 FOR UPDATE`, eventID).Scan(&workID)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	return workID, false, nil
}

func (r *Repository) completeSourceProjectionWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	return r.completeSourceProjectionWorkItemWithFailure(ctx, workID, leaseToken, terminalState, now, nil, false)
}

type sourceProjectionFailureRecord struct {
	previousCount int
	count         int
	class         string
	summary       string
}

func (r *Repository) completeSourceProjectionWorkItemWithFailure(ctx context.Context, workID, leaseToken, terminalState string, now time.Time, failure *sourceProjectionFailureRecord, deferLaneReconcile bool) (bool, error) {
	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticReadCommittedTx(ctx, db)
		if err != nil {
			return false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}

	var currentEventID, installedPackID, workType, projectionKey, projectionActiveKey, eventKind, companionPack, companionFlow, sourceLaneKey, targetMutationLaneKey string
	var existingTerminalAt sql.NullTime
	var routeVersion, generation int64
	projectionGenerationSchema := true
	err := q.QueryRowContext(ctx, `
SELECT w.EventId, w.InstalledPackId, w.WorkType,
       COALESCE(NULLIF(w.ProjectionKey, ''), w.WorkId),
       COALESCE(w.ProjectionActiveKey, ''), w.TerminalAt,
       COALESCE(w.SourceLaneKey, ''), COALESCE(w.TargetMutationLaneKey, ''), w.Generation
  FROM _System_ConnectorWorkItem w
 WHERE w.WorkId = ?
   AND w.State = ?
   AND w.LeaseToken = ?
 LIMIT 1
 FOR UPDATE`, workID, WorkStateProcessing, leaseToken).Scan(
		&currentEventID, &installedPackID, &workType, &projectionKey, &projectionActiveKey, &existingTerminalAt,
		&sourceLaneKey, &targetMutationLaneKey, &generation,
	)
	if isProjectionGenerationSchemaMissing(err) {
		projectionGenerationSchema = false
		projectionKey = workID
		err = q.QueryRowContext(ctx, `
SELECT w.EventId, e.InstalledPackId, w.WorkType,
       e.EventKind, COALESCE(e.RouteVersion, 0),
       COALESCE(e.CompanionPackInstallId, ''), COALESCE(e.CompanionFlowVersionId, ''),
       COALESCE(e.SourceLaneKey, ''), COALESCE(w.TargetMutationLaneKey, ''), w.Generation
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e ON e.Id = w.EventId
 WHERE w.WorkId = ?
   AND w.State = ?
   AND w.LeaseToken = ?
 LIMIT 1
 FOR UPDATE`, workID, WorkStateProcessing, leaseToken).Scan(
			&currentEventID, &installedPackID, &workType, &eventKind, &routeVersion,
			&companionPack, &companionFlow, &sourceLaneKey, &targetMutationLaneKey, &generation,
		)
	}
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if !IsSourceProjectionWorkID(projectionKey) {
		return false, ErrInvalidWorkItem
	}
	var attempt int
	if failure == nil {
		_, ok := WorkItemAttemptOutcomeFromContext(ctx)
		if ok {
			if err := q.QueryRowContext(ctx, `
SELECT Attempt
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND EventId = ?`, workID, currentEventID).Scan(&attempt); err != nil {
				return false, err
			}
		}
	}

	var successorID, successorQueueClass, successorStatus string
	var successorRunAfter sql.NullTime
	var successorLaneOrdinal int64
	hasSuccessor := false
	var queuedSuccessor *sourceProjectionWorkItemRow
	if projectionGenerationSchema {
		transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}
		queuedSuccessor, err = transitionRepo.loadQueuedSourceProjectionSuccessorForUpdate(ctx, projectionKey, workID)
		if err != nil {
			return false, err
		}
		if queuedSuccessor != nil {
			if err := validateSourceProjectionWorkItemRow(queuedSuccessor, CreateWorkItemInput{
				InstalledPackID: installedPackID,
				WorkType:        workType,
				SourceLaneKey:   sourceLaneKey,
			}, projectionKey); err != nil {
				return false, err
			}
			hasSuccessor = true
		}
	} else {
		// Older schemas have no immutable projection-generation history. Keep
		// their bounded compatibility handoff on the legacy Event lifecycle.
		err = q.QueryRowContext(ctx, `
SELECT Id, QueueClass, Status, NextAttemptAt, EnqueueOrdinal
 FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND LOWER(TRIM(EventKind)) = LOWER(TRIM(?))
   AND COALESCE(RouteVersion, 0) = ?
   AND COALESCE(CompanionPackInstallId, '') = ?
   AND COALESCE(CompanionFlowVersionId, '') = ?
   AND COALESCE(SourceLaneKey, '') = ?
   AND Status IN ('pending', 'failed')
 ORDER BY EnqueueOrdinal DESC, CreatedDate DESC, Id DESC
	 LIMIT 1`, installedPackID, eventKind, routeVersion, companionPack, companionFlow, sourceLaneKey).Scan(
			&successorID, &successorQueueClass, &successorStatus, &successorRunAfter, &successorLaneOrdinal,
		)
		hasSuccessor = err == nil
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return false, err
		}
	}

	reconcileWorkID := workID
	if projectionGenerationSchema {
		var res sql.Result
		var updateErr error
		if failure != nil {
			res, updateErr = q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       FailureCount = ?,
       FailureClass = ?,
       FailureSummary = ?,
       LastFailureAt = ?,
       ProjectionKey = ?,
       ProjectionActiveKey = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND State = ?
   AND LeaseToken = ?
   AND FailureCount = ?`,
				terminalState, failure.count, failure.class, failure.summary, now,
				projectionKey, now, now,
				workID, currentEventID, WorkStateProcessing, leaseToken, failure.previousCount)
		} else {
			res, updateErr = q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       ProjectionKey = ?,
       ProjectionActiveKey = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND State = ?
   AND LeaseToken = ?`,
				terminalState, projectionKey, now, now,
				workID, currentEventID, WorkStateProcessing, leaseToken)
		}
		if updateErr != nil {
			return false, updateErr
		}
		completed, updateErr := rowsAffected(res)
		if updateErr != nil || !completed {
			return completed, updateErr
		}
		reconcileWorkID = ""
		if hasSuccessor {
			res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET ProjectionActiveKey = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND ProjectionKey = ?
   AND ProjectionActiveKey IS NULL
   AND Attempt = 0
   AND State IN (?, ?, ?, ?)`,
				projectionKey, now, queuedSuccessor.workID, projectionKey,
				WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry)
			if err != nil {
				return false, err
			}
			activated, err := rowsAffected(res)
			if err != nil {
				return false, err
			}
			if !activated {
				return false, ErrInvalidWorkItem
			}
			reconcileWorkID = queuedSuccessor.workID
		}
	} else {
		var res sql.Result
		if hasSuccessor {
			nextState, runAfter := sourceProjectionSuccessorReadyState(successorStatus, successorRunAfter, now)
			res, err = q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET EventId = ?, QueueClass = ?, State = ?, RunAfter = ?, Attempt = 0,
       LaneOrdinal = ?, LeaseOwner = NULL, LeaseToken = NULL, LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ? AND EventId = ? AND State = ? AND LeaseToken = ?`,
				successorID, successorQueueClass, nextState, runAfter, successorLaneOrdinal, now,
				workID, currentEventID, WorkStateProcessing, leaseToken)
		} else {
			res, err = q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?, LeaseOwner = NULL, LeaseToken = NULL, LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ? AND EventId = ? AND State = ? AND LeaseToken = ?`,
				terminalState, now, workID, currentEventID, WorkStateProcessing, leaseToken)
		}
		if err != nil {
			return false, err
		}
		completed, updateErr := rowsAffected(res)
		if updateErr != nil || !completed {
			return completed, updateErr
		}
	}
	if failure == nil {
		if err := appendWorkItemAttempt(ctx, q, workID, attempt, currentEventID, terminalState, now); err != nil {
			return false, err
		}
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return false, err
		}
	}
	if !deferLaneReconcile {
		if err := r.reconcileLaneReadyStateWithRetry(ctx, CreateWorkItemInput{
			WorkID: reconcileWorkID, InstalledPackID: installedPackID, Generation: generation,
			SourceLaneKey: sourceLaneKey, TargetMutationLaneKey: targetMutationLaneKey, Now: now,
		}); err != nil {
			return false, err
		}
	}
	return true, nil
}

func sourceProjectionSuccessorReadyState(status string, nextAttemptAt sql.NullTime, now time.Time) (string, time.Time) {
	state := WorkStatePending
	if status == "failed" {
		state = WorkStateRetry
	}
	runAfter := now
	if nextAttemptAt.Valid {
		runAfter = nextAttemptAt.Time
	}
	return state, runAfter
}

func isProjectionGenerationSchemaMissing(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	for _, column := range []string{"projectionkey", "projectionactivekey", "terminalat"} {
		if strings.Contains(msg, column) &&
			(strings.Contains(msg, "1054") || strings.Contains(msg, "unknown column") || strings.Contains(msg, "no such column")) {
			return true
		}
	}
	return false
}

// AcquireSchedulerLease retains the former pack-wide coordination behavior for
// pinned and custom callers during the compatibility window.
//
// Deprecated: in-tree WorkItem workers use row CAS as the sole claim authority
// and do not call this method.
func (r *Repository) AcquireSchedulerLease(ctx context.Context, in SchedulerLeaseInput) (bool, error) {
	if err := validateSchedulerLeaseInput(in); err != nil {
		return false, err
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorSchedulerLease
   SET OwnerId = ?,
       FencingToken = ?,
       LeaseUntil = ?,
       RenewedAt = ?,
       LastModifiedDate = ?
 WHERE InstalledPackId = ?
   AND Generation = ?
   AND LeaseUntil <= ?`,
		in.OwnerID, in.FencingToken, in.LeaseUntil, in.Now, in.Now,
		in.InstalledPackID, in.Generation, in.Now)
	if err != nil {
		return false, err
	}
	acquired, err := rowsAffected(res)
	if err != nil || acquired {
		return acquired, err
	}
	res, err = r.db.ExecContext(ctx, `
UPDATE _System_ConnectorSchedulerLease l
   SET l.OwnerId = ?,
       l.FencingToken = ?,
       l.LeaseUntil = ?,
       l.RenewedAt = ?,
       l.LastModifiedDate = ?
 WHERE l.InstalledPackId = ?
   AND l.Generation = ?
   AND l.LeaseUntil > ?
   AND NOT EXISTS (
       SELECT 1
         FROM _System_ConnectorWorkItem w
        WHERE w.InstalledPackId = l.InstalledPackId
          AND w.Generation = l.Generation
          AND w.State = ?
          AND w.LeaseUntil > ?
        LIMIT 1
   )`,
		in.OwnerID, in.FencingToken, in.LeaseUntil, in.Now, in.Now,
		in.InstalledPackID, in.Generation, in.Now, WorkStateProcessing, in.Now)
	if err != nil {
		return false, err
	}
	acquired, err = rowsAffected(res)
	if err != nil || acquired {
		return acquired, err
	}
	_, err = r.db.ExecContext(ctx, `
INSERT INTO _System_ConnectorSchedulerLease
       (InstalledPackId, Generation, OwnerId, FencingToken, LeaseUntil, RenewedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
		in.InstalledPackID, in.Generation, in.OwnerID, in.FencingToken, in.LeaseUntil, in.Now, in.Now, in.Now)
	if err != nil {
		if isDuplicateKey(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// ReleaseSchedulerLease retains the former pack-wide coordination behavior for
// pinned and custom callers during the compatibility window.
//
// Deprecated: in-tree WorkItem workers do not acquire SchedulerLease rows.
func (r *Repository) ReleaseSchedulerLease(ctx context.Context, in SchedulerLeaseReleaseInput) (bool, error) {
	if err := validateSchedulerLeaseReleaseInput(in); err != nil {
		return false, err
	}
	res, err := r.db.ExecContext(ctx, `
DELETE FROM _System_ConnectorSchedulerLease
 WHERE InstalledPackId = ?
   AND Generation = ?
   AND OwnerId = ?
   AND FencingToken = ?`,
		in.InstalledPackID, in.Generation, in.OwnerID, in.FencingToken)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

// RenewSchedulerLease retains the former pack-wide coordination behavior for
// pinned and custom callers during the compatibility window.
//
// Deprecated: in-tree WorkItem workers do not renew SchedulerLease rows.
func (r *Repository) RenewSchedulerLease(ctx context.Context, in SchedulerLeaseInput) (bool, error) {
	if err := validateSchedulerLeaseInput(in); err != nil {
		return false, err
	}
	freshUntil := in.Now.Add(in.LeaseUntil.Sub(in.Now) / 2)
	fresh, err := r.schedulerLeaseHeldUntil(ctx, in, freshUntil)
	if err != nil {
		return false, err
	}
	if fresh {
		return true, nil
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorSchedulerLease
   SET LeaseUntil = ?,
       RenewedAt = ?,
       LastModifiedDate = ?
 WHERE InstalledPackId = ?
   AND Generation = ?
   AND OwnerId = ?
   AND FencingToken = ?`,
		in.LeaseUntil, in.Now, in.Now,
		in.InstalledPackID, in.Generation, in.OwnerID, in.FencingToken)
	if err != nil {
		return false, err
	}
	renewed, err := rowsAffected(res)
	if err != nil || renewed {
		return renewed, err
	}
	return r.schedulerLeaseHeld(ctx, in)
}

func (r *Repository) schedulerLeaseHeld(ctx context.Context, in SchedulerLeaseInput) (bool, error) {
	return r.schedulerLeaseHeldUntil(ctx, in, in.Now)
}

func (r *Repository) schedulerLeaseHeldUntil(ctx context.Context, in SchedulerLeaseInput, heldUntil time.Time) (bool, error) {
	var held int
	err := r.db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_ConnectorSchedulerLease
 WHERE InstalledPackId = ?
   AND Generation = ?
   AND OwnerId = ?
   AND FencingToken = ?
   AND LeaseUntil > ?
 LIMIT 1`,
		in.InstalledPackID, in.Generation, in.OwnerID, in.FencingToken, heldUntil).Scan(&held)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

const selectWorkItemSQL = `
SELECT WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter,
	       Attempt, COALESCE(LeaseOwner, ''), COALESCE(LeaseToken, ''), LeaseUntil,
	       COALESCE(SourceLaneKey, ''), COALESCE(TargetMutationLaneKey, ''),
	       LaneOrdinal, Generation, CreatedDate, LastModifiedDate
	  FROM _System_ConnectorWorkItem`

type workItemClaimCandidate struct {
	workID     string
	queueClass string
}

func (r *Repository) findClaimableWorkItems(ctx context.Context, in ClaimWorkItemInput, limit int) ([]workItemClaimCandidate, error) {
	if limit <= 0 || limit > claimWorkItemCandidateWindow {
		limit = claimWorkItemCandidateWindow
	}
	preferredQueueClasses := claimQueueClassPreferenceForEligibility(ctx, in.QueueClasses)
	withPreference := !in.StaleOnly && len(preferredQueueClasses) > 0
	selectColumns := "w.WorkId"
	if withPreference {
		selectColumns += ", w.QueueClass"
	}
	queueClassFilter, queueClassArgs := workItemQueueClassFilter("w", in.QueueClasses)
	workTypeFilter, workTypeArgs := workItemWorkTypeFilter("w", ClaimWorkTypes(ctx))
	args := make([]any, 0, 16)
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	args = append(args, WorkTypeConnectorEvent, WorkTypeWorkflowActionApply)
	args = append(args, in.InstalledPackID, in.Generation)
	args = append(args, workTypeArgs...)
	args = append(args, queueClassArgs...)
	args = append(args, claimCandidateWorkItemArgs(in.Now, in.StaleOnly)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventCandidateFenceArgs(in.Now)...)
	preferredOrderSQL, preferredOrderArgs := workItemPreferredQueueClassOrder("w", preferredQueueClasses, in.StaleOnly)
	args = append(args, preferredOrderArgs...)
	rows, err := r.db.QueryContext(ctx, `
SELECT `+selectColumns+`
  FROM _System_ConnectorWorkItem w
	  `+schedulerLeaseJoin("w", distributed)+`
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
   AND w.WorkType IN (?, ?)
 WHERE w.InstalledPackId = ?
   AND w.Generation = ?`+workTypeFilter+queueClassFilter+`
   AND `+claimCandidateWorkItemPredicate("w", in.StaleOnly)+`
   AND `+laneHeadWorkItemPredicate("w")+`
   AND `+workItemEventCandidateFencePredicate("w", "e")+`
	 ORDER BY `+preferredOrderSQL+workItemClaimOrderBy("w")+`
 LIMIT `+fmt.Sprint(limit), args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	candidates := make([]workItemClaimCandidate, 0, limit)
	for rows.Next() {
		var candidate workItemClaimCandidate
		var err error
		if withPreference {
			err = rows.Scan(&candidate.workID, &candidate.queueClass)
		} else {
			err = rows.Scan(&candidate.workID)
		}
		if err != nil {
			return nil, err
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return candidates, nil
}

func (r *Repository) ResolveClaimableWorkItemGeneration(ctx context.Context, in ResolveClaimableGenerationInput) (int64, bool, error) {
	if strings.TrimSpace(in.InstalledPackID) == "" || in.Generation < 0 {
		return 0, false, fmt.Errorf("connector workqueue: invalid claimable generation input")
	}
	now := in.Now
	if now.IsZero() {
		now = r.now().UTC()
	}
	queueClassFilter, queueClassArgs := workItemQueueClassFilter("w", in.QueueClasses)
	workTypeFilter, workTypeArgs := workItemWorkTypeFilter("w", ClaimWorkTypes(ctx))
	var generation int64
	args := []any{in.InstalledPackID}
	args = append(args, workTypeArgs...)
	args = append(args, queueClassArgs...)
	args = append(args, claimCandidateWorkItemArgs(now, in.StaleOnly)...)
	err := r.db.QueryRowContext(ctx, `
SELECT w.Generation
  FROM _System_ConnectorWorkItem w
 WHERE w.InstalledPackId = ?`+workTypeFilter+queueClassFilter+`
   AND `+claimCandidateWorkItemPredicate("w", in.StaleOnly)+`
 ORDER BY w.Generation DESC
 LIMIT 1`, args...).Scan(&generation)
	if err == nil {
		return generation, true, nil
	}
	if err != sql.ErrNoRows {
		return 0, false, err
	}

	activeArgs := []any{in.InstalledPackID}
	activeArgs = append(activeArgs, workTypeArgs...)
	activeArgs = append(activeArgs, queueClassArgs...)
	err = r.db.QueryRowContext(ctx, `
SELECT w.Generation
  FROM _System_ConnectorWorkItem w
 WHERE w.InstalledPackId = ?`+workTypeFilter+queueClassFilter+`
   AND w.State IN (?, ?, ?, ?, ?)
   AND `+workItemExecutionInputPredicate("w")+`
 ORDER BY w.Generation DESC
 LIMIT 1`, append(activeArgs, WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateProcessing)...).Scan(&generation)
	if err == sql.ErrNoRows {
		return in.Generation, false, nil
	}
	if err != nil {
		return 0, false, err
	}
	return generation, true, nil
}

func (r *Repository) findClaimableLaneHeadWorkItems(ctx context.Context, in ClaimWorkItemInput, limit int) ([]string, error) {
	if limit <= 0 || limit > claimWorkItemCandidateWindow {
		limit = claimWorkItemCandidateWindow
	}
	queueClassFilter, queueClassArgs := workItemQueueClassFilter("w", in.QueueClasses)
	workTypeFilter, workTypeArgs := workItemWorkTypeFilter("w", ClaimWorkTypes(ctx))
	args := make([]any, 0, 16)
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	args = append(args, WorkTypeConnectorEvent, WorkTypeWorkflowActionApply)
	args = append(args, in.InstalledPackID, in.Generation)
	args = append(args, workTypeArgs...)
	args = append(args, queueClassArgs...)
	args = append(args, claimCandidateWorkItemArgs(in.Now, in.StaleOnly)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventCandidateFenceArgs(in.Now)...)
	preferredOrderSQL, preferredOrderArgs := workItemPreferredQueueClassOrder("w", claimQueueClassPreferenceForEligibility(ctx, in.QueueClasses), in.StaleOnly)
	args = append(args, preferredOrderArgs...)
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId
  FROM _System_ConnectorWorkItem w
	  `+schedulerLeaseJoin("w", distributed)+`
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
   AND w.WorkType IN (?, ?)
 WHERE w.InstalledPackId = ?
   AND w.Generation = ?`+workTypeFilter+queueClassFilter+`
   AND `+claimCandidateWorkItemPredicate("w", in.StaleOnly)+`
   AND `+laneHeadWorkItemPredicate("w")+`
   AND `+workItemEventCandidateFencePredicate("w", "e")+`
	 ORDER BY `+preferredOrderSQL+workItemClaimOrderBy("w")+`
 LIMIT `+fmt.Sprint(limit), args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	workIDs := make([]string, 0, limit)
	for rows.Next() {
		var workID string
		if err := rows.Scan(&workID); err != nil {
			return nil, err
		}
		workIDs = append(workIDs, workID)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return workIDs, nil
}

func (r *Repository) promoteBlockedLaneHeadsForClaim(ctx context.Context, in ClaimWorkItemInput, limit int) (int, error) {
	if limit <= 0 {
		limit = claimWorkItemCandidateWindow
	}
	queueClassFilter, queueClassArgs := workItemQueueClassFilter("w", in.QueueClasses)
	workTypeFilter, workTypeArgs := workItemWorkTypeFilter("w", ClaimWorkTypes(ctx))
	args := []any{in.InstalledPackID, in.Generation}
	args = append(args, workTypeArgs...)
	args = append(args, queueClassArgs...)
	args = append(args, WorkStateBlocked, WorkStateBlockedRetry, in.Now)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	preferredOrderSQL, preferredOrderArgs := workItemPreferredQueueClassOrder("w", claimQueueClassPreferenceForEligibility(ctx, in.QueueClasses), false)
	args = append(args, preferredOrderArgs...)
	args = append(args, limit)
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId,
	       w.EventId,
	       w.WorkType,
	       w.QueueClass,
       w.RunAfter,
	       COALESCE(w.SourceLaneKey, ''),
	       COALESCE(w.TargetMutationLaneKey, ''),
	       w.LaneOrdinal
	  FROM _System_ConnectorWorkItem w
	 WHERE w.InstalledPackId = ?
	   AND w.Generation = ?`+workTypeFilter+queueClassFilter+`
	   AND w.State IN (?, ?)
	   AND w.RunAfter <= ?
	   AND `+workItemExecutionInputPredicate("w")+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	 ORDER BY `+preferredOrderSQL+workItemClaimOrderBy("w")+`
	 LIMIT ?`, args...)
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	inputs := make([]CreateWorkItemInput, 0, limit)
	seenLanes := make(map[string]struct{}, limit)
	for rows.Next() {
		var candidate CreateWorkItemInput
		if err := rows.Scan(
			&candidate.WorkID,
			&candidate.EventID,
			&candidate.WorkType,
			&candidate.QueueClass,
			&candidate.RunAfter,
			&candidate.SourceLaneKey,
			&candidate.TargetMutationLaneKey,
			&candidate.LaneOrdinal,
		); err != nil {
			return 0, err
		}
		laneKey := workItemLaneKey(candidate)
		if laneKey == "" {
			continue
		}
		if _, ok := seenLanes[laneKey]; ok {
			continue
		}
		seenLanes[laneKey] = struct{}{}
		candidate.InstalledPackID = in.InstalledPackID
		candidate.Generation = in.Generation
		candidate.State = WorkStateBlocked
		candidate.Now = in.Now
		inputs = append(inputs, candidate)
	}
	if err := rows.Err(); err != nil {
		return 0, err
	}
	promoted := 0
	var lastErr error
	for _, candidate := range inputs {
		if err := r.reconcileLaneReadyState(ctx, candidate); err != nil {
			lastErr = err
			if ctx.Err() != nil {
				return promoted, err
			}
			continue
		}
		promoted++
	}
	if promoted == 0 && lastErr != nil {
		return 0, lastErr
	}
	return promoted, nil
}

func rotateCandidateSlice[T any](candidates []T, leaseToken string) []T {
	if len(candidates) <= 1 {
		return candidates
	}
	offset := stableClaimOffset(leaseToken, len(candidates))
	if offset == 0 {
		return candidates
	}
	rotated := make([]T, 0, len(candidates))
	rotated = append(rotated, candidates[offset:]...)
	rotated = append(rotated, candidates[:offset]...)
	return rotated
}

func orderWorkItemCandidates(ctx context.Context, candidates []workItemClaimCandidate, in ClaimWorkItemInput) []workItemClaimCandidate {
	preferredQueueClasses := claimQueueClassPreferenceForEligibility(ctx, in.QueueClasses)
	if in.StaleOnly || len(preferredQueueClasses) == 0 {
		return rotateCandidateSlice(candidates, in.LeaseToken)
	}
	preferredClasses := make(map[string]struct{}, len(preferredQueueClasses))
	for _, class := range preferredQueueClasses {
		preferredClasses[class] = struct{}{}
	}
	preferred := make([]workItemClaimCandidate, 0, len(candidates))
	fallback := make([]workItemClaimCandidate, 0, len(candidates))
	for _, candidate := range candidates {
		if _, ok := preferredClasses[candidate.queueClass]; ok {
			preferred = append(preferred, candidate)
			continue
		}
		fallback = append(fallback, candidate)
	}
	preferred = rotateCandidateSlice(preferred, in.LeaseToken)
	fallback = rotateCandidateSlice(fallback, in.LeaseToken)
	return append(preferred, fallback...)
}

func stableClaimOffset(seed string, n int) int {
	if n <= 1 {
		return 0
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(seed))
	return int(h.Sum32() % uint32(n))
}

func (r *Repository) claimSelectedWorkItem(ctx context.Context, workID string, in ClaimWorkItemInput) (bool, error) {
	if IsSourceProjectionWorkID(workID) {
		if _, err := r.GetSourceProjectionRange(ctx, workID); err != nil {
			return false, err
		}
	}
	args := []any{
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStateProcessing, EventExecutionOwnerEvent,
		in.OwnerID, in.LeaseToken, in.LeaseUntil, in.Now,
		EventExecutionOwnerEvent, EventWorkItemCompatibilityFenceStatus,
		EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent,
		EventExecutionOwnerWorkItem,
		workID, in.InstalledPackID, in.Generation,
	}
	args = append(args, claimCandidateWorkItemArgs(in.Now, in.StaleOnly)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventOwnershipFenceArgs(in.Now)...)
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem AS w
  LEFT JOIN _System_ConnectorEvent AS e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
   AND w.WorkType IN (?, ?)
	   SET w.State = ?,
	       w.Attempt = w.Attempt + 1,
	       w.OwnershipEpoch = w.OwnershipEpoch + 1,
	       w.FailureCount = CASE
	           WHEN e.ExecutionOwner = ? THEN GREATEST(w.FailureCount, COALESCE(e.RetryCount, 0))
	           ELSE w.FailureCount
	       END,
	       w.LeaseOwner = ?,
	       w.LeaseToken = ?,
       w.LeaseUntil = ?,
       w.LastModifiedDate = ?,
       e.Status = CASE WHEN e.ExecutionOwner = ? THEN ? ELSE e.Status END,
       e.NextAttemptAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.NextAttemptAt END,
       e.ClaimedBy = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedBy END,
       e.ClaimedAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedAt END,
       e.ClaimToken = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimToken END,
       e.ExecutionOwner = ?
	 WHERE w.WorkId = ?
	   AND w.InstalledPackId = ?
	   AND w.Generation = ?
	   AND `+claimCandidateWorkItemPredicate("w", in.StaleOnly)+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	   AND `+workItemEventOwnershipFencePredicate("w", "e")+
		schedulerLeaseFilter("w", distributed), args...)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

func (r *Repository) claimSelectedWorkItemAtEpoch(ctx context.Context, workID string, in ClaimWorkItemInput, ownershipEpoch int64, allowCrossSourceDelivery bool) (bool, error) {
	if IsSourceProjectionWorkID(workID) {
		if _, err := r.GetSourceProjectionRange(ctx, workID); err != nil {
			return false, err
		}
	}
	args := []any{
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStateProcessing, EventExecutionOwnerEvent,
		in.OwnerID, in.LeaseToken, in.LeaseUntil, in.Now,
		EventExecutionOwnerEvent, EventWorkItemCompatibilityFenceStatus,
		EventExecutionOwnerEvent, EventExecutionOwnerEvent,
		EventExecutionOwnerEvent, EventExecutionOwnerEvent,
		EventExecutionOwnerWorkItem,
		workID, in.InstalledPackID, in.Generation, ownershipEpoch,
		SourceProjectionWorkIDPrefix + "%",
	}
	args = append(args, claimCandidateWorkItemArgs(in.Now, false)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventOwnershipFenceArgs(in.Now)...)
	if !allowCrossSourceDelivery {
		args = append(args, workItemDeliveryPredecessorArgs()...)
	}
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem AS w
	  LEFT JOIN _System_ConnectorEvent AS e
	    ON e.Id = w.EventId
	   AND e.InstalledPackId = w.InstalledPackId
	   AND w.WorkType IN (?, ?)
	   SET w.State = ?,
	       w.Attempt = w.Attempt + 1,
	       w.OwnershipEpoch = w.OwnershipEpoch + 1,
	       w.FailureCount = CASE
	           WHEN e.ExecutionOwner = ? THEN GREATEST(w.FailureCount, COALESCE(e.RetryCount, 0))
	           ELSE w.FailureCount
	       END,
	       w.LeaseOwner = ?,
	       w.LeaseToken = ?,
       w.LeaseUntil = ?,
       w.LastModifiedDate = ?,
       e.Status = CASE WHEN e.ExecutionOwner = ? THEN ? ELSE e.Status END,
       e.NextAttemptAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.NextAttemptAt END,
       e.ClaimedBy = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedBy END,
       e.ClaimedAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedAt END,
       e.ClaimToken = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimToken END,
       e.ExecutionOwner = ?
	 WHERE w.WorkId = ?
	   AND w.InstalledPackId = ?
	   AND w.Generation = ?
	   AND w.OwnershipEpoch = ?
	   AND w.WorkId NOT LIKE ?
	   AND NULLIF(COALESCE(w.ProjectionKey, ''), '') IS NULL
	   AND NULLIF(COALESCE(w.ProjectionActiveKey, ''), '') IS NULL
	   AND `+claimCandidateWorkItemPredicate("w", false)+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	   AND `+workItemEventOwnershipFencePredicate("w", "e")+
		workItemDeliveryPredecessorFilter("w", "e", allowCrossSourceDelivery)+
		schedulerLeaseFilter("w", distributed), args...)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

func (r *Repository) claimSelectedPromotedWorkItem(ctx context.Context, workID string, in ClaimWorkItemInput) (bool, error) {
	if IsSourceProjectionWorkID(workID) {
		if _, err := r.GetSourceProjectionRange(ctx, workID); err != nil {
			return false, err
		}
	}
	args := []any{
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStateProcessing, EventExecutionOwnerEvent,
		in.OwnerID, in.LeaseToken, in.LeaseUntil, in.Now,
		EventExecutionOwnerEvent, EventWorkItemCompatibilityFenceStatus,
		EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent,
		EventExecutionOwnerWorkItem,
		workID, in.InstalledPackID, in.Generation,
	}
	args = append(args, claimCandidateWorkItemArgs(in.Now, in.StaleOnly)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventOwnershipFenceArgs(in.Now)...)
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem AS w
  LEFT JOIN _System_ConnectorEvent AS e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
   AND w.WorkType IN (?, ?)
	   SET w.State = ?,
	       w.Attempt = w.Attempt + 1,
	       w.OwnershipEpoch = w.OwnershipEpoch + 1,
	       w.FailureCount = CASE
	           WHEN e.ExecutionOwner = ? THEN GREATEST(w.FailureCount, COALESCE(e.RetryCount, 0))
	           ELSE w.FailureCount
	       END,
	       w.LeaseOwner = ?,
	       w.LeaseToken = ?,
       w.LeaseUntil = ?,
       w.LastModifiedDate = ?,
       e.Status = CASE WHEN e.ExecutionOwner = ? THEN ? ELSE e.Status END,
       e.NextAttemptAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.NextAttemptAt END,
       e.ClaimedBy = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedBy END,
       e.ClaimedAt = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimedAt END,
       e.ClaimToken = CASE WHEN e.ExecutionOwner = ? THEN NULL ELSE e.ClaimToken END,
       e.ExecutionOwner = ?
	 WHERE w.WorkId = ?
	   AND w.InstalledPackId = ?
	   AND w.Generation = ?
	   AND `+claimCandidateWorkItemPredicate("w", in.StaleOnly)+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	   AND `+workItemEventOwnershipFencePredicate("w", "e")+
		schedulerLeaseFilter("w", distributed), args...)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

// workItemEventOwnershipFencePredicate is the rolling-version arbiter between
// the WorkItem claimant and binaries that still claim _System_ConnectorEvent.
// The multi-table UPDATE changes Event ownership and wins the WorkItem lease in
// one statement. An older Event claimant either changes Status to processing
// first, or loses its pending/failed/processing CAS after this statement freezes
// Status as skipped. A stale processing Event is reclaimable under the exact
// legacy Event lease timeout.
func workItemEventOwnershipFencePredicate(workAlias, eventAlias string) string {
	return fmt.Sprintf(`(
	       %[1]s.WorkType NOT IN (?, ?)
	       OR (
	           %[2]s.Id IS NOT NULL
	           AND (
	               (%[2]s.ExecutionOwner = ? AND %[2]s.Status = ?)
	               OR (
	                   %[2]s.ExecutionOwner = ?
	                   AND (
		                       %[2]s.Status = ?
		                       OR (
		                           %[2]s.Status = ?
		                           AND (%[2]s.NextAttemptAt IS NULL OR %[2]s.NextAttemptAt <= ?)
		                       )
		                       OR (%[2]s.Status = ? AND %[2]s.LastModifiedDate < ?)
	                   )
	               )
	           )
	       )
	   )`, workAlias, eventAlias)
}

func workItemEventOwnershipFenceArgs(now time.Time) []any {
	return []any{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
		EventExecutionOwnerWorkItem,
		EventWorkItemCompatibilityFenceStatus,
		EventExecutionOwnerEvent,
		"pending",
		"failed", now,
		"processing", now.Add(-claimpolicy.StaleProcessingSeconds * time.Second),
	}
}

// workItemEventCandidateFencePredicate admits rows that can win the execution
// ownership CAS and attempt-zero compatibility mirrors whose legacy Event owner
// already reached terminal state. The latter are selected only for terminal
// convergence; ClaimNextWorkItem never executes them.
func workItemEventCandidateFencePredicate(workAlias, eventAlias string) string {
	return fmt.Sprintf(`(
	       %s
	       OR (
	           %[2]s.WorkType IN (?, ?)
	           AND %[2]s.Attempt = 0
	           AND %[3]s.ExecutionOwner = ?
	           AND %[3]s.Status IN (?, ?, ?)
	       )
	   )`,
		workItemEventOwnershipFencePredicate(workAlias, eventAlias),
		workAlias,
		eventAlias,
	)
}

func workItemEventCandidateFenceArgs(now time.Time) []any {
	args := workItemEventOwnershipFenceArgs(now)
	return append(args,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		EventExecutionOwnerEvent,
		WorkStateProcessed, WorkStateSkipped, WorkStateDead,
	)
}

const legacyEventTerminalConvergenceOwner = "legacy-event-terminal-convergence"

// convergeLegacyEventTerminalWorkItem closes the only rolling-version outcome
// in which an older Event executor wins before WorkItem ownership transfers.
// The old executor remains the durable owner, so this path copies only its
// terminal classification onto an attempt-zero compatibility WorkItem. It
// never executes business behavior and never changes Event execution ownership.
//
// Event is locked before WorkItem to preserve the compatibility transaction
// order used by replay and legacy finalization. Source projections reuse the
// normal terminal-generation handoff so queued successor activation remains
// atomic with terminal history.
func (r *Repository) convergeLegacyEventTerminalWorkItem(ctx context.Context, workID string, now time.Time) (bool, error) {
	db, ok := r.db.(*sql.DB)
	if !ok || strings.TrimSpace(workID) == "" || now.IsZero() {
		return false, nil
	}

	var eventID string
	err := db.QueryRowContext(ctx, `
SELECT w.EventId
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.WorkId = ?
   AND w.WorkType IN (?, ?)
   AND w.Attempt = 0
   AND w.State IN (?, ?, ?, ?)
   AND e.ExecutionOwner = ?
   AND e.Status IN (?, ?, ?)
 LIMIT 1`,
		workID,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
		EventExecutionOwnerEvent,
		WorkStateProcessed, WorkStateSkipped, WorkStateDead,
	).Scan(&eventID)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticReadCommittedTx(ctx, db)
	if err != nil {
		return false, err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	var eventStatus, executionOwner string
	err = tx.QueryRowContext(ctx, `
SELECT Status, ExecutionOwner
  FROM _System_ConnectorEvent
 WHERE Id = ?
 FOR UPDATE`, eventID).Scan(&eventStatus, &executionOwner)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	terminalState, terminal := legacyEventTerminalWorkState(eventStatus)
	if executionOwner != EventExecutionOwnerEvent || !terminal {
		return false, nil
	}

	var (
		storedEventID         string
		installedPackID       string
		workType              string
		state                 string
		sourceLaneKey         string
		targetMutationLaneKey string
		generation            int64
	)
	err = tx.QueryRowContext(ctx, `
SELECT EventId, InstalledPackId, WorkType, State,
       COALESCE(SourceLaneKey, ''), COALESCE(TargetMutationLaneKey, ''), Generation
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND EventId = ?
   AND WorkType IN (?, ?)
   AND Attempt = 0
   AND State IN (?, ?, ?, ?)
 LIMIT 1
 FOR UPDATE`,
		workID, eventID,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
	).Scan(
		&storedEventID, &installedPackID, &workType, &state,
		&sourceLaneKey, &targetMutationLaneKey, &generation,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	transitionRepo := r.WithTx(tx)
	if IsSourceProjectionWorkID(workID) {
		leaseToken := "legacy-event-terminal:" + r.idgen()
		res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = ?,
       LeaseToken = ?,
       LeaseUntil = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND WorkType = ?
   AND Attempt = 0
   AND State = ?`,
			WorkStateProcessing,
			legacyEventTerminalConvergenceOwner,
			leaseToken,
			now,
			now,
			workID,
			storedEventID,
			workType,
			state,
		)
		if err != nil {
			return false, err
		}
		prepared, err := rowsAffected(res)
		if err != nil || !prepared {
			return false, err
		}
		completed, err := transitionRepo.completeSourceProjectionWorkItem(
			ctx,
			workID,
			leaseToken,
			terminalState,
			now,
		)
		if err != nil || !completed {
			return false, err
		}
	} else {
		res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       TerminalAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND WorkType = ?
   AND Attempt = 0
   AND State = ?`,
			terminalState, now, now,
			workID, storedEventID, workType, state,
		)
		if err != nil {
			return false, err
		}
		completed, err := rowsAffected(res)
		if err != nil || !completed {
			return false, err
		}
	}

	if err := tx.Commit(); err != nil {
		return false, err
	}
	committed = true
	if !IsSourceProjectionWorkID(workID) {
		if err := r.reconcileLaneReadyStateWithRetry(ctx, CreateWorkItemInput{
			WorkID:                workID,
			InstalledPackID:       installedPackID,
			WorkType:              workType,
			SourceLaneKey:         sourceLaneKey,
			TargetMutationLaneKey: targetMutationLaneKey,
			Generation:            generation,
			Now:                   now,
		}); err != nil {
			return false, err
		}
	}
	return true, nil
}

func legacyEventTerminalWorkState(status string) (string, bool) {
	switch strings.TrimSpace(status) {
	case WorkStateProcessed:
		return WorkStateProcessed, true
	case WorkStateSkipped:
		return WorkStateSkipped, true
	case WorkStateDead:
		return WorkStateDead, true
	default:
		return "", false
	}
}

func (r *Repository) reconcileLaneReadyState(ctx context.Context, in CreateWorkItemInput) error {
	laneKey := workItemLaneKey(in)
	if laneKey == "" {
		return nil
	}
	now := in.Now
	if now.IsZero() {
		now = r.now().UTC()
	}
	headID, err := r.nextLaneReadyHead(ctx, in, laneKey, now)
	if err != nil {
		return err
	}
	if err := r.demoteNonHeadLaneReadyRows(ctx, in, laneKey, headID, now); err != nil {
		return err
	}
	if headID == "" {
		return nil
	}
	return r.promoteLaneReadyHead(ctx, in, laneKey, headID, now)
}

func (r *Repository) nextLaneReadyHead(ctx context.Context, in CreateWorkItemInput, laneKey string, now time.Time) (string, error) {
	heads, err := r.nextLaneReadyHeads(ctx, in, []string{laneKey}, now)
	if err != nil || len(heads) == 0 {
		return "", err
	}
	return heads[0], nil
}

func (r *Repository) demoteNonHeadLaneReadyRows(ctx context.Context, in CreateWorkItemInput, laneKey, headID string, now time.Time) error {
	var heads []string
	if strings.TrimSpace(headID) != "" {
		heads = []string{headID}
	}
	return r.demoteNonHeadLaneReadyBatch(ctx, in, []string{laneKey}, heads, now)
}

func (r *Repository) promoteLaneReadyHead(ctx context.Context, in CreateWorkItemInput, laneKey, workID string, now time.Time) error {
	return r.promoteLaneReadyBatch(ctx, in, []string{laneKey}, []string{workID}, now)
}

// These disjoint point lookups preserve target-first/source-fallback identity
// without hiding lane equality inside COALESCE or a cross-index OR.
const laneProcessingTargetSQL = `SELECT 1
 FROM _System_ConnectorWorkItem processing USE INDEX (idx_connector_work_item_target_lane)
 WHERE processing.InstalledPackId = %s AND processing.Generation = %s
   AND processing.TargetMutationLaneKey = %s AND processing.State = ?`

const laneProcessingSourceSQL = `SELECT 1
 FROM _System_ConnectorWorkItem processing USE INDEX (idx_connector_work_item_source_lane_gen)
 WHERE processing.InstalledPackId = %s AND processing.Generation = %s
   AND processing.SourceLaneKey = %s AND processing.State = ?
   AND (processing.TargetMutationLaneKey IS NULL OR processing.TargetMutationLaneKey = '')`

func (r *Repository) promoteNextLaneHeadAfter(ctx context.Context, completedWorkID string, now time.Time) error {
	var installedPackID, sourceLaneKey, targetMutationLaneKey string
	var generation int64
	err := r.db.QueryRowContext(ctx, `
SELECT InstalledPackId,
       Generation,
       COALESCE(SourceLaneKey, ''),
       COALESCE(TargetMutationLaneKey, '')
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
 LIMIT 1`, completedWorkID).Scan(&installedPackID, &generation, &sourceLaneKey, &targetMutationLaneKey)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return err
	}
	laneKey := normalizedLaneKey(sourceLaneKey, targetMutationLaneKey)
	if laneKey == "" {
		return nil
	}
	return r.reconcileLaneReadyState(ctx, CreateWorkItemInput{
		InstalledPackID:       installedPackID,
		WorkType:              WorkTypeConnectorEvent,
		SourceLaneKey:         sourceLaneKey,
		TargetMutationLaneKey: targetMutationLaneKey,
		Generation:            generation,
		Now:                   now,
	})
}

func (r *Repository) promoteNextLaneHeadAfterWithRetry(ctx context.Context, completedWorkID string, now time.Time) error {
	return retryLanePromotion(ctx, func() error {
		return r.promoteNextLaneHeadAfter(ctx, completedWorkID, now)
	})
}

func (r *Repository) reconcileLaneReadyStateWithRetry(ctx context.Context, in CreateWorkItemInput) error {
	return retryLanePromotion(ctx, func() error {
		return r.reconcileLaneReadyState(ctx, in)
	})
}

func retryLanePromotion(ctx context.Context, promote func() error) error {
	var lastErr error
	for attempt := 0; attempt < lanePromotionRetryAttempts; attempt++ {
		if err := promote(); err != nil {
			lastErr = err
			if ctx.Err() != nil || attempt == lanePromotionRetryAttempts-1 {
				return err
			}
			timer := time.NewTimer(time.Duration(attempt+1) * 25 * time.Millisecond)
			select {
			case <-ctx.Done():
				if !timer.Stop() {
					select {
					case <-timer.C:
					default:
					}
				}
				return ctx.Err()
			case <-timer.C:
			}
			continue
		}
		return nil
	}
	return lastErr
}

func (r *Repository) findClaimableWorkItemByEvent(ctx context.Context, in ClaimWorkItemByEventInput, staleOnly bool) (string, bool, error) {
	args := []any{
		in.EventID,
		in.WorkType,
		in.InstalledPackID,
		in.Generation,
	}
	args = append(args, claimCandidateWorkItemArgs(in.Now, staleOnly)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventCandidateFenceArgs(in.Now)...)
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	row := r.db.QueryRowContext(ctx, `
SELECT w.WorkId
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.EventId = ?
   AND w.WorkType = ?
	   AND w.InstalledPackId = ?
	   AND w.Generation = ?
	   AND `+claimCandidateWorkItemPredicate("w", staleOnly)+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	   AND `+workItemEventCandidateFencePredicate("w", "e")+
		schedulerLeaseFilter("w", distributed)+`
 LIMIT 1`, args...)
	var workID string
	err := row.Scan(&workID)
	if err == sql.ErrNoRows {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	return workID, true, nil
}

func (r *Repository) findClaimableWorkItemByEventAtEpoch(ctx context.Context, in ClaimWorkItemByEventInput, allowCrossSourceDelivery bool) (string, string, int64, bool, error) {
	args := []any{
		in.EventID,
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
		in.InstalledPackID,
		in.Generation,
		SourceProjectionWorkIDPrefix + "%",
	}
	args = append(args, claimCandidateWorkItemArgs(in.Now, false)...)
	args = append(args, laneHeadWorkItemArgs(in.InstalledPackID, in.Generation, in.Now)...)
	args = append(args, workItemEventOwnershipFenceArgs(in.Now)...)
	if !allowCrossSourceDelivery {
		args = append(args, workItemDeliveryPredecessorArgs()...)
	}
	distributed := DistributedClaim(ctx)
	args = appendSchedulerLeaseArgs(args, distributed, in.OwnerID, in.FencingToken, in.Now)
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId, w.WorkType, w.OwnershipEpoch
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.EventId = ?
	   AND w.WorkType IN (?, ?)
	   AND w.InstalledPackId = ?
	   AND w.Generation = ?
	   AND w.WorkId NOT LIKE ?
	   AND NULLIF(COALESCE(w.ProjectionKey, ''), '') IS NULL
	   AND NULLIF(COALESCE(w.ProjectionActiveKey, ''), '') IS NULL
	   AND `+claimCandidateWorkItemPredicate("w", false)+`
	   AND `+laneHeadWorkItemPredicate("w")+`
	   AND `+workItemEventOwnershipFencePredicate("w", "e")+
		workItemDeliveryPredecessorFilter("w", "e", allowCrossSourceDelivery)+
		schedulerLeaseFilter("w", distributed)+`
	ORDER BY w.WorkId
 LIMIT 2
 FOR UPDATE`, args...)
	if err != nil {
		return "", "", 0, false, err
	}
	defer rows.Close()
	if !rows.Next() {
		return "", "", 0, false, rows.Err()
	}
	var workID string
	var workType string
	var ownershipEpoch int64
	if err := rows.Scan(&workID, &workType, &ownershipEpoch); err != nil {
		return "", "", 0, false, err
	}
	if rows.Next() {
		return "", "", 0, false, ErrInvalidWorkItem
	}
	if err := rows.Err(); err != nil {
		return "", "", 0, false, err
	}
	return workID, strings.TrimSpace(workType), ownershipEpoch, true, nil
}

func claimCandidateWorkItemPredicate(alias string, staleOnly bool) string {
	var statePredicate string
	if staleOnly {
		statePredicate = fmt.Sprintf("%s.State = ? AND %s.LeaseUntil IS NOT NULL AND %s.LeaseUntil <= ?", alias, alias, alias)
	} else {
		statePredicate = fmt.Sprintf("%s.State IN (?, ?) AND %s.RunAfter <= ?", alias, alias)
	}
	return fmt.Sprintf("(%s) AND %s", statePredicate, workItemExecutionInputPredicate(alias))
}

func claimCandidateWorkItemArgs(now time.Time, staleOnly bool) []any {
	if staleOnly {
		return []any{WorkStateProcessing, now}
	}
	return []any{WorkStatePending, WorkStateRetry, now}
}

func schedulerLeasePredicate(workAlias string) string {
	return fmt.Sprintf(`EXISTS (
       SELECT 1
         FROM _System_ConnectorSchedulerLease l
        WHERE l.InstalledPackId = %s.InstalledPackId
          AND l.Generation = %s.Generation
          AND l.OwnerId = ?
          AND l.FencingToken = ?
          AND l.LeaseUntil > ?
   )`, workAlias, workAlias)
}

func schedulerLeaseJoin(workAlias string, distributed bool) string {
	if distributed {
		return ""
	}
	return fmt.Sprintf(`JOIN _System_ConnectorSchedulerLease l
    ON l.InstalledPackId = %s.InstalledPackId
   AND l.Generation = %s.Generation
   AND l.OwnerId = ?
   AND l.FencingToken = ?
   AND l.LeaseUntil > ?`, workAlias, workAlias)
}

func schedulerLeaseFilter(workAlias string, distributed bool) string {
	if distributed {
		return ""
	}
	return "\n\t   AND " + schedulerLeasePredicate(workAlias)
}

func appendSchedulerLeaseArgs(args []any, distributed bool, ownerID, fencingToken string, now time.Time) []any {
	if distributed {
		return args
	}
	return append(args, ownerID, fencingToken, now)
}

func laneHeadWorkItemPredicate(workAlias string) string {
	return fmt.Sprintf(`NOT EXISTS (
	       SELECT 1
	         FROM _System_ConnectorWorkItem older USE INDEX (idx_connector_work_item_lease_due)
	        WHERE older.InstalledPackId = ?
	          AND older.Generation = ?
	          AND older.WorkId <> %s.WorkId
	          AND %s
	          AND %s
	          AND %s
		   )`, workAlias, activeWorkItemBlockerPredicate("older"), workItemLaneIdentityMatchesPredicate("older", workAlias), workItemPrecedesPredicate("older", workAlias))
}

// activeWorkItemBlockerPredicate treats a lease-free dependency suspension as
// ordinary processing because WorkStateDependencyBlocked is a compatibility
// alias for processing.  Keep this claim path limited to baseline columns:
// dependency metadata may be unavailable until the feature schema is applied,
// but it must not stop existing WorkItem claims before then.
func activeWorkItemBlockerPredicate(workAlias string) string {
	return fmt.Sprintf(`%s
	      AND (
	          %s.State = ?
	          OR (
	              %s.State IN (?, ?, ?, ?)
	              AND %s.RunAfter <= ?
	          )
	      )`, claimableProjectionGenerationPredicate(workAlias), workAlias, workAlias, workAlias)
}

func workItemLaneIdentityMatchesPredicate(olderWorkAlias, currentWorkAlias string) string {
	return fmt.Sprintf(`COALESCE(NULLIF(%s.TargetMutationLaneKey, ''), %s.SourceLaneKey, '') = COALESCE(NULLIF(%s.TargetMutationLaneKey, ''), %s.SourceLaneKey, '')`, olderWorkAlias, olderWorkAlias, currentWorkAlias, currentWorkAlias)
}

func workItemLaneKey(in CreateWorkItemInput) string {
	return normalizedLaneKey(in.SourceLaneKey, in.TargetMutationLaneKey)
}

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

func isReadyWorkState(state string) bool {
	return state == WorkStatePending || state == WorkStateRetry
}

func blockedStateForReadyState(state string) string {
	if state == WorkStateRetry {
		return WorkStateBlockedRetry
	}
	return WorkStateBlocked
}

func laneHeadWorkItemArgs(installedPackID string, generation int64, now time.Time) []any {
	args := []any{installedPackID, generation}
	return append(args, laneHeadWorkItemBlockerArgs(now)...)
}

func laneHeadWorkItemBlockerArgs(now time.Time) []any {
	return []any{
		WorkStateProcessing,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
		now,
	}
}

// workItemDeliveryPredecessorFilter preserves strict delivery FIFO after Event
// ownership transfers to WorkItem. Event.Status is frozen on that path, so the
// predecessor's WorkItem state is the authoritative nonterminal signal. Unlike
// source-lane readiness, delivery FIFO remains blocked during retry backoff.
func workItemDeliveryPredecessorFilter(workAlias, eventAlias string, allowCrossSourceDelivery bool) string {
	if allowCrossSourceDelivery {
		return ""
	}
	return fmt.Sprintf(`
	   AND NOT EXISTS (
	       SELECT 1
	         FROM _System_ConnectorWorkItem delivery_older
	         JOIN _System_ConnectorEvent delivery_event
	           ON delivery_event.Id = delivery_older.EventId
	          AND delivery_event.InstalledPackId = delivery_older.InstalledPackId
	        WHERE delivery_older.InstalledPackId = %[1]s.InstalledPackId
	          AND delivery_older.Generation = %[1]s.Generation
	          AND delivery_older.WorkId <> %[1]s.WorkId
	          AND %[3]s
	          AND %[4]s
	          AND NULLIF(%[2]s.DeliveryId, '') IS NOT NULL
	          AND delivery_event.DeliveryId = %[2]s.DeliveryId
	          AND delivery_event.DeliverySequence < %[2]s.DeliverySequence
	   )`, workAlias, eventAlias, eventBackedWorkTypePredicate("delivery_older"), strictDeliveryWorkItemBlockerPredicate("delivery_older"))
}

func workItemDeliveryPredecessorArgs() []any {
	return []any{WorkStateProcessed, WorkStateSkipped, WorkStateDead}
}

func strictDeliveryWorkItemBlockerPredicate(workAlias string) string {
	return fmt.Sprintf(`%s
	      AND %s.State NOT IN (?, ?, ?)`, claimableProjectionGenerationPredicate(workAlias), workAlias)
}

func eventBackedWorkTypePredicate(workAlias string) string {
	return fmt.Sprintf("%s.WorkType IN ('%s', '%s')", workAlias, WorkTypeConnectorEvent, WorkTypeWorkflowActionApply)
}

func immutableEventInputPredicate(workAlias string) string {
	return fmt.Sprintf(`(NOT (%s) OR EXISTS (
	       SELECT 1
	         FROM _System_ConnectorEvent immutable_event
	        WHERE immutable_event.Id = %s.EventId
	   ))`, eventBackedWorkTypePredicate(workAlias), workAlias)
}

// A source-projection successor is durable before it becomes executable.
// ProjectionActiveKey is the single activation pointer; a row with a
// ProjectionKey and no active key cannot participate in readiness, lane
// blocking, or claim CAS until the prior generation terminalizes it.
func claimableProjectionGenerationPredicate(workAlias string) string {
	return fmt.Sprintf(`(
	       NULLIF(COALESCE(%s.ProjectionKey, ''), '') IS NULL
	       OR NULLIF(COALESCE(%s.ProjectionActiveKey, ''), '') IS NOT NULL
	   )`, workAlias, workAlias)
}

func workItemExecutionInputPredicate(workAlias string) string {
	return fmt.Sprintf("(%s AND %s)",
		immutableEventInputPredicate(workAlias),
		claimableProjectionGenerationPredicate(workAlias),
	)
}

func workItemPrecedesPredicate(olderWorkAlias, currentWorkAlias string) string {
	return fmt.Sprintf(`(
	          %s.LaneOrdinal < %s.LaneOrdinal
	          OR (
	              %s.LaneOrdinal = %s.LaneOrdinal
	              AND (
	                  %s.CreatedDate < %s.CreatedDate
	                  OR (%s.CreatedDate = %s.CreatedDate AND %s.WorkId < %s.WorkId)
	              )
	          )
	      )`, olderWorkAlias, currentWorkAlias, olderWorkAlias, currentWorkAlias, olderWorkAlias, currentWorkAlias, olderWorkAlias, currentWorkAlias, olderWorkAlias, currentWorkAlias)
}

func laneReadyHeadOrderBy(workAlias, eventAlias string) string {
	eventBacked := eventBackedWorkTypePredicate(workAlias)
	return fmt.Sprintf(`%[1]s.LaneOrdinal ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL THEN %[2]s.CreatedDate ELSE %[1]s.CreatedDate END ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL AND %[2]s.EnqueueOrdinal <> 0 THEN 0 ELSE 1 END ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL AND %[2]s.EnqueueOrdinal <> 0 THEN %[2]s.EnqueueOrdinal ELSE 0 END ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL AND %[2]s.EnqueueOrdinal = 0 AND %[2]s.DeliverySequence <> 0 THEN 0 ELSE 1 END ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL AND %[2]s.EnqueueOrdinal = 0 THEN %[2]s.DeliverySequence ELSE 0 END ASC,
	         CASE WHEN %[3]s AND %[2]s.Id IS NOT NULL THEN %[2]s.Id ELSE '' END ASC,
	         %[1]s.CreatedDate ASC,
	         %[1]s.WorkId ASC`, workAlias, eventAlias, eventBacked)
}

func workItemClaimOrderBy(workAlias string) string {
	return fmt.Sprintf(`%[1]s.LaneOrdinal ASC,
	         %[1]s.RunAfter ASC,
	         %[1]s.CreatedDate ASC,
	         %[1]s.WorkId ASC`, workAlias)
}

func workItemPreferredQueueClassOrder(workAlias string, preferredQueueClasses []string, staleOnly bool) (string, []any) {
	if staleOnly || len(preferredQueueClasses) == 0 {
		return "", nil
	}
	if len(preferredQueueClasses) == 1 {
		return fmt.Sprintf("CASE WHEN %s.QueueClass = ? THEN 0 ELSE 1 END ASC,\n\t         ", workAlias), []any{preferredQueueClasses[0]}
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(preferredQueueClasses)), ",")
	args := make([]any, 0, len(preferredQueueClasses))
	for _, class := range preferredQueueClasses {
		args = append(args, class)
	}
	return fmt.Sprintf("CASE WHEN %s.QueueClass IN (%s) THEN 0 ELSE 1 END ASC,\n\t         ", workAlias, placeholders), args
}

func (r *Repository) scanWorkItem(row interface {
	Scan(dest ...any) error
}) (*WorkItem, error) {
	var work WorkItem
	err := row.Scan(
		&work.WorkID, &work.EventID, &work.InstalledPackID, &work.WorkType, &work.QueueClass, &work.State, &work.RunAfter,
		&work.Attempt, &work.LeaseOwner, &work.LeaseToken, &work.LeaseUntil,
		&work.SourceLaneKey, &work.TargetMutationLaneKey, &work.LaneOrdinal, &work.Generation, &work.CreatedDate, &work.LastModifiedDate,
	)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &work, nil
}

func normalizeFailWorkItemInput(in FailWorkItemInput) FailWorkItemInput {
	in.WorkID = strings.TrimSpace(in.WorkID)
	in.EventID = strings.TrimSpace(in.EventID)
	in.LeaseToken = strings.TrimSpace(in.LeaseToken)
	in.FailureClass = strings.TrimSpace(in.FailureClass)
	in.FailureSummary = strings.TrimSpace(in.FailureSummary)
	return in
}

func validateFailWorkItemInput(in FailWorkItemInput) error {
	if in.WorkID == "" || in.EventID == "" || in.LeaseToken == "" ||
		in.FailureClass == "" || len(in.FailureClass) > 64 ||
		in.FailureSummary == "" || in.RetryThreshold <= 0 ||
		len(in.RetryDelays) == 0 || in.Now.IsZero() {
		return ErrInvalidWorkItem
	}
	for _, delay := range in.RetryDelays {
		if delay < 0 {
			return ErrInvalidWorkItem
		}
	}
	return nil
}

func failWorkItemResult(in FailWorkItemInput, failureCount int) *FailWorkItemResult {
	result := &FailWorkItemResult{
		State:        WorkStateDead,
		FailureCount: failureCount,
		RunAfter:     in.Now,
	}
	if failureCount >= in.RetryThreshold {
		return result
	}
	delayIndex := failureCount - 1
	if delayIndex >= len(in.RetryDelays) {
		delayIndex = len(in.RetryDelays) - 1
	}
	result.State = WorkStateRetry
	result.RunAfter = in.Now.Add(in.RetryDelays[delayIndex])
	return result
}

func (r *Repository) defaultCreateWorkItemInput(in CreateWorkItemInput) CreateWorkItemInput {
	in.WorkID = strings.TrimSpace(in.WorkID)
	in.EventID = strings.TrimSpace(in.EventID)
	in.InstalledPackID = strings.TrimSpace(in.InstalledPackID)
	in.WorkType = strings.TrimSpace(in.WorkType)
	in.QueueClass = strings.TrimSpace(in.QueueClass)
	in.State = strings.TrimSpace(in.State)
	in.SourceLaneKey = strings.TrimSpace(in.SourceLaneKey)
	in.TargetMutationLaneKey = strings.TrimSpace(in.TargetMutationLaneKey)
	if strings.TrimSpace(in.WorkID) == "" {
		in.WorkID = r.idgen()
	}
	if strings.TrimSpace(in.QueueClass) == "" {
		in.QueueClass = DefaultQueueClass
	}
	if strings.TrimSpace(in.State) == "" {
		in.State = WorkStatePending
	}
	if in.RunAfter.IsZero() {
		in.RunAfter = r.now().UTC()
	}
	if in.Now.IsZero() {
		in.Now = r.now().UTC()
	}
	return in
}

func normalizeClaimWorkItemByEventInput(in ClaimWorkItemByEventInput) ClaimWorkItemByEventInput {
	in.EventID = strings.TrimSpace(in.EventID)
	in.WorkType = strings.TrimSpace(in.WorkType)
	in.InstalledPackID = strings.TrimSpace(in.InstalledPackID)
	in.OwnerID = strings.TrimSpace(in.OwnerID)
	in.FencingToken = strings.TrimSpace(in.FencingToken)
	in.LeaseToken = strings.TrimSpace(in.LeaseToken)
	if in.WorkType == "" {
		in.WorkType = WorkTypeConnectorEvent
	}
	return in
}

func validateClaimWorkItemByEventInput(in ClaimWorkItemByEventInput) error {
	if in.EventID == "" ||
		in.WorkType == "" ||
		in.InstalledPackID == "" ||
		in.OwnerID == "" ||
		in.FencingToken == "" ||
		in.LeaseToken == "" ||
		in.Generation < 0 ||
		!in.LeaseUntil.After(in.Now) ||
		in.Now.IsZero() {
		return ErrInvalidWorkItem
	}
	return nil
}

func normalizeLegacyEventMirrorReadinessInput(in LegacyEventMirrorReadinessInput) LegacyEventMirrorReadinessInput {
	in.InstalledPackID = strings.TrimSpace(in.InstalledPackID)
	return in
}

func validateLegacyEventMirrorReadinessInput(in LegacyEventMirrorReadinessInput) error {
	if in.InstalledPackID == "" || in.Generation < 0 {
		return ErrInvalidWorkItem
	}
	return nil
}

func validateCreateWorkItemInput(in CreateWorkItemInput) error {
	if strings.TrimSpace(in.WorkID) == "" ||
		strings.TrimSpace(in.EventID) == "" ||
		strings.TrimSpace(in.InstalledPackID) == "" ||
		strings.TrimSpace(in.WorkType) == "" ||
		strings.TrimSpace(in.QueueClass) == "" ||
		strings.TrimSpace(in.State) == "" ||
		(in.State != WorkStatePending && in.State != WorkStateRetry) ||
		in.RunAfter.IsZero() ||
		in.Now.IsZero() {
		return ErrInvalidWorkItem
	}
	return nil
}

func normalizeClaimWorkItemInput(in ClaimWorkItemInput) ClaimWorkItemInput {
	in.InstalledPackID = strings.TrimSpace(in.InstalledPackID)
	in.OwnerID = strings.TrimSpace(in.OwnerID)
	in.FencingToken = strings.TrimSpace(in.FencingToken)
	in.LeaseToken = strings.TrimSpace(in.LeaseToken)
	in.QueueClass = strings.TrimSpace(in.QueueClass)
	in.QueueClasses = normalizeClaimQueueClasses(in.QueueClass, in.QueueClasses)
	return in
}

func normalizeWorkTypes(workTypes []string) []string {
	seen := make(map[string]struct{}, len(workTypes))
	out := make([]string, 0, len(workTypes))
	for _, raw := range workTypes {
		workType := strings.TrimSpace(raw)
		if workType == "" {
			continue
		}
		if _, ok := seen[workType]; ok {
			continue
		}
		seen[workType] = struct{}{}
		out = append(out, workType)
	}
	return out
}

func normalizeClaimQueueClasses(queueClass string, queueClasses []string) []string {
	seen := make(map[string]struct{}, 1+len(queueClasses))
	out := make([]string, 0, 1+len(queueClasses))
	add := func(raw string) {
		class := strings.TrimSpace(raw)
		if class == "" {
			return
		}
		if _, ok := seen[class]; ok {
			return
		}
		seen[class] = struct{}{}
		out = append(out, class)
	}
	add(queueClass)
	for _, class := range queueClasses {
		add(class)
	}
	return out
}

func workItemQueueClassFilter(alias string, queueClasses []string) (string, []any) {
	switch len(queueClasses) {
	case 0:
		return "", nil
	case 1:
		return fmt.Sprintf(" AND %s.QueueClass = ?", alias), []any{queueClasses[0]}
	default:
		placeholders := strings.TrimRight(strings.Repeat("?,", len(queueClasses)), ",")
		args := make([]any, 0, len(queueClasses))
		for _, class := range queueClasses {
			args = append(args, class)
		}
		return fmt.Sprintf(" AND %s.QueueClass IN (%s)", alias, placeholders), args
	}
}

func workItemWorkTypeFilter(alias string, workTypes []string) (string, []any) {
	switch len(workTypes) {
	case 0:
		return "", nil
	case 1:
		return fmt.Sprintf(" AND %s.WorkType = ?", alias), []any{workTypes[0]}
	default:
		placeholders := strings.TrimRight(strings.Repeat("?,", len(workTypes)), ",")
		args := make([]any, 0, len(workTypes))
		for _, workType := range workTypes {
			args = append(args, workType)
		}
		return fmt.Sprintf(" AND %s.WorkType IN (%s)", alias, placeholders), args
	}
}

func validateClaimWorkItemInput(in ClaimWorkItemInput) error {
	if strings.TrimSpace(in.InstalledPackID) == "" ||
		strings.TrimSpace(in.OwnerID) == "" ||
		strings.TrimSpace(in.FencingToken) == "" ||
		strings.TrimSpace(in.LeaseToken) == "" ||
		in.Now.IsZero() ||
		!in.LeaseUntil.After(in.Now) {
		return ErrInvalidWorkItem
	}
	return nil
}

func validateSchedulerLeaseInput(in SchedulerLeaseInput) error {
	if strings.TrimSpace(in.InstalledPackID) == "" ||
		strings.TrimSpace(in.OwnerID) == "" ||
		strings.TrimSpace(in.FencingToken) == "" ||
		!in.LeaseUntil.After(in.Now) {
		return ErrInvalidSchedulerLease
	}
	return nil
}

func validateSchedulerLeaseReleaseInput(in SchedulerLeaseReleaseInput) error {
	if strings.TrimSpace(in.InstalledPackID) == "" ||
		strings.TrimSpace(in.OwnerID) == "" ||
		strings.TrimSpace(in.FencingToken) == "" ||
		in.Generation < 0 ||
		in.Now.IsZero() {
		return ErrInvalidSchedulerLease
	}
	return nil
}

func rowsAffected(res sql.Result) (bool, error) {
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

func isDuplicateKey(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062
}