package persistence

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	mysqldriver "github.com/go-sql-driver/mysql"
	"github.com/google/uuid"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// ErrOutboxEventAlreadyClaimed means the requested claim transition could not
// prove current attempt authority. ClaimEvent returns it after a competing
// transition; scoped refresh/finalization paths also return it for a replaced
// or expired claim. Callers must treat it as a benign lost race, not a failure.
var ErrOutboxEventAlreadyClaimed = errors.New("outbox event already claimed")

// ErrOutboxRecordKeyUnavailable is returned when the pull-worker FIFO path is
// asked to use RecordKey before the physical column has been applied.
var ErrOutboxRecordKeyUnavailable = errors.New("outbox RecordKey column unavailable")

// outboxStatusProcessing is the transient status written by ClaimEvent when
// a pending row is successfully locked for dispatch. Defined locally rather
// than in shared/pkg/constants so this bug fix stays scoped to the outbox
// module; it is only ever written/read inside this file and its paired
// service. The column is VARCHAR so the value round-trips unchanged.
const outboxStatusProcessing = "processing"

// OutboxStaleProcessingSeconds is the legacy repository's default claim
// lease. Independently deployed workers install an explicit role-owned lease;
// both variants are evaluated against the database clock.
const OutboxStaleProcessingSeconds = 300

// OutboxRetryVisibleAfterSeconds prevents independent pull workers from
// immediately reclaiming the same just-failed row in a tight loop. It preserves
// the old worker's "next poll" retry cadence without introducing a new
// scheduler column or long retry backoff.
const OutboxRetryVisibleAfterSeconds = 1

const outboxAttemptSavepointName = "outbox_attempt"
const tableRecordEventOutbox = "_System_RecordEventOutbox"
const outboxDatabaseNowSQL = "CURRENT_TIMESTAMP(6)"
const recordEventExecutionScopeDigestColumn = "ExecutionScopeDigest"
const recordEventClaimIndex = "idx_record_event_outbox_claim"

// outboxLeaseRefreshNowSQL advances a live lease even if two consecutive
// database statements observe the same microsecond. RowsAffected is used as
// the compare-and-transition result, so a matched refresh must not become a
// false lost-claim solely because its timestamp assignment was a no-op.
// Both operands are evaluated by TiDB; no process clock participates.
const outboxLeaseRefreshNowSQL = "GREATEST(CURRENT_TIMESTAMP(6), TIMESTAMPADD(MICROSECOND, 1, `LastModifiedDate`))"
const recordEventClaimTokenColumn = "ClaimToken"

var recordEventWorkerRequiredColumns = map[string]string{
	"Id":                   "varchar(255)",
	"OrgId":                "varchar(255)",
	"AppInstallationId":    "varchar(255)",
	"RuntimeCapabilityId":  "varchar(255)",
	"ExecutionScopeDigest": "char(64)",
	"EventType":            "varchar(255)",
	"Payload":              "json",
	"Status":               "varchar(255)",
	"RetryCount":           "int",
	"RecordKey":            "varchar(512)",
	"ErrorMessage":         "text",
	"ProcessedDate":        "datetime",
	"RepairId":             "varchar(255)",
	"RepairReason":         "varchar(1024)",
	"RepairCount":          "int",
	"LastRepairAt":         "datetime(6)",
	"CreatedDate":          "datetime",
	"LastModifiedDate":     "datetime(6)",
	"ClaimToken":           "varchar(36)",
}

// OutboxRepository handles database operations for the outbox pattern
type OutboxRepository struct {
	db          *sql.DB
	tableName   string
	scope       *RuntimeWorkScope
	scopeDigest string
	claimLease  time.Duration

	recordKeyColumnMu      sync.Mutex
	recordKeyColumnKnown   bool
	recordKeyColumnPresent bool
}

// NewOutboxRepository creates a new OutboxRepository
func NewOutboxRepository(db *sql.DB) *OutboxRepository {
	return newOutboxRepositoryForTable(db, constants.TableOutboxEvent)
}

// NewRecordEventOutboxRepository creates an outbox repository backed by the
// record-trigger queue table. The schema mirrors _System_OutboxEvent for this
// MVP so the service can preserve publish/retry semantics while isolating
// record-trigger write pressure from the generic event outbox.
func NewRecordEventOutboxRepository(db *sql.DB) *OutboxRepository {
	return newOutboxRepositoryForTable(db, tableRecordEventOutbox)
}

// NewScopedRecordEventOutboxRepository constructs the independently deployed
// worker's queue view. Every read and durable transition is constrained by the
// declared scope in SQL; immutable metadata binding remains the execution
// authority after selection.
func NewScopedRecordEventOutboxRepository(db *sql.DB, scope RuntimeWorkScope, claimLease time.Duration) (*OutboxRepository, error) {
	if db == nil || scope.Validate() != nil || claimLease <= 0 {
		return nil, errors.New("record-event outbox: database, scope, and claim lease are required")
	}
	scopeDigest, err := scope.Digest()
	if err != nil {
		return nil, fmt.Errorf("record-event outbox: derive execution scope digest: %w", err)
	}
	repo := newOutboxRepositoryForTable(db, tableRecordEventOutbox)
	repo.scope = &scope
	repo.scopeDigest = scopeDigest
	repo.claimLease = claimLease
	return repo, nil
}

func newOutboxRepositoryForTable(db *sql.DB, tableName string) *OutboxRepository {
	if tableName == "" {
		tableName = constants.TableOutboxEvent
	}
	return &OutboxRepository{db: db, tableName: tableName}
}

func (r *OutboxRepository) outboxTableName() string {
	if r == nil || r.tableName == "" {
		return constants.TableOutboxEvent
	}
	return r.tableName
}

func (r *OutboxRepository) processingLease() time.Duration {
	if r != nil && r.claimLease > 0 {
		return r.claimLease
	}
	return defaultOutboxProcessingLease
}

// ProcessingClaimHeartbeatInterval is an optional worker policy surface. It
// keeps the heartbeat derived from the same lease used by SQL reclaim logic.
func (r *OutboxRepository) ProcessingClaimHeartbeatInterval() time.Duration {
	interval := r.processingLease() / 3
	if interval <= 0 {
		return time.Millisecond
	}
	return interval
}

func (r *OutboxRepository) constrainToRuntimeWorkScope(qb *query.Builder) *query.Builder {
	if r == nil || r.scope == nil {
		return qb
	}
	return qb.Where(recordEventExecutionScopeDigestColumn, r.scopeDigest).
		Where("OrgId", r.scope.OrgID).
		Where("AppInstallationId", r.scope.AppInstallationID).
		Where("RuntimeCapabilityId", r.scope.RuntimeCapabilityID)
}

// CheckRecordEventWorkerReady proves the dedicated queue has the full
// dispatch and exact-repair schema before an independently deployed worker is
// allowed to become ready. The legacy outbox deliberately cannot satisfy this
// check and therefore cannot become a fallback.
func (r *OutboxRepository) CheckRecordEventWorkerReady(ctx context.Context) error {
	if r == nil || r.db == nil || r.outboxTableName() != tableRecordEventOutbox || r.scope == nil || r.scope.Validate() != nil || r.claimLease <= 0 {
		return errors.New("record-event outbox readiness: dedicated scoped repository and claim lease are required")
	}
	if err := verifyRecordEventOutboxSchema(ctx, r.db); err != nil {
		return fmt.Errorf("record-event outbox readiness: %w", err)
	}
	if err := verifyRecordEventOutboxScope(ctx, r.db, *r.scope, r.scopeDigest); err != nil {
		return fmt.Errorf("record-event outbox readiness: %w", err)
	}
	return nil
}

func verifyRecordEventOutboxSchema(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return errors.New("dedicated record-event database is required")
	}
	if err := VerifyRequiredTableColumns(ctx, db, tableRecordEventOutbox, recordEventWorkerRequiredColumns); err != nil {
		return err
	}
	return VerifyExactTableIndex(ctx, db, tableRecordEventOutbox, recordEventClaimIndex, false, []string{
		recordEventExecutionScopeDigestColumn,
		constants.FieldSysOutboxEvent_Status,
		constants.FieldCreatedDate,
		constants.FieldID,
	})
}

func verifyRecordEventOutboxScope(ctx context.Context, db *sql.DB, scope RuntimeWorkScope, scopeDigest string) error {
	var invalid int64
	err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_RecordEventOutbox
WHERE (ExecutionScopeDigest = ? OR (OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?))
  AND NOT (ExecutionScopeDigest <=> ? AND OrgId <=> ? AND AppInstallationId <=> ? AND RuntimeCapabilityId <=> ?)`,
		scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
		scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
	).Scan(&invalid)
	if err != nil {
		return fmt.Errorf("verify exact execution scope projection: %w", err)
	}
	if invalid != 0 {
		return fmt.Errorf("execution scope projection has %d mismatched row(s); run schema apply after quiescing legacy producers", invalid)
	}
	return nil
}

// VerifyRuntimeRecordEventQueueSchema proves the producer/consumer payload
// table can preserve the scoped, immutable execution binding and repair state.
// It neither selects a scope nor claims worker ownership.
func VerifyRuntimeRecordEventQueueSchema(ctx context.Context, db *sql.DB) error {
	return verifyRecordEventOutboxSchema(ctx, db)
}

// GetExecutor returns the transactional executor if present in context, or the DB connection
func (r *OutboxRepository) GetExecutor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

func (r *OutboxRepository) CreateOutboxAttemptSavepoint(ctx context.Context) error {
	_, err := r.GetExecutor(ctx).ExecContext(ctx, "SAVEPOINT "+outboxAttemptSavepointName)
	return err
}

func (r *OutboxRepository) RollbackToOutboxAttemptSavepoint(ctx context.Context) error {
	_, err := r.GetExecutor(ctx).ExecContext(ctx, "ROLLBACK TO SAVEPOINT "+outboxAttemptSavepointName)
	return err
}

func (r *OutboxRepository) ReleaseOutboxAttemptSavepoint(ctx context.Context) error {
	_, err := r.GetExecutor(ctx).ExecContext(ctx, "RELEASE SAVEPOINT "+outboxAttemptSavepointName)
	return err
}

// RefreshOutboxProcessingClaim atomically validates and renews a processing
// claim. For an independently deployed scoped worker, the token, scope, status,
// and database-owned lease must all still be current; an expired claim cannot
// be revived even if no competitor has reclaimed it yet.
func (r *OutboxRepository) RefreshOutboxProcessingClaim(ctx context.Context, claim models.OutboxClaim) error {
	qb := query.Update(r.outboxTableName()).
		Set(map[string]interface{}{
			constants.FieldLastModifiedDate: query.Raw(outboxLeaseRefreshNowSQL),
		}).
		Where(constants.FieldID, claim.EventID).
		Where(constants.FieldSysOutboxEvent_Status, outboxStatusProcessing)
	qb = r.constrainToClaim(qb, claim)
	qb = r.constrainToRuntimeWorkScope(qb)
	if r.scope != nil {
		// A scoped claim is authority only while both its generation token and
		// database-owned lease are live. Token equality alone is insufficient:
		// before a competing reclaim rotates the token, a paused worker could
		// otherwise renew an already-expired attempt and regain effect authority.
		liveSQL, liveArgs := outboxProcessingLeasePredicate(r.processingLease(), true)
		qb = qb.WhereRaw(liveSQL, liveArgs)
	}
	qr := qb.Build()

	res, err := r.GetExecutor(ctx).ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrOutboxEventAlreadyClaimed
	}
	return nil
}

// HasRecordKeyColumn reports whether the physical RecordKey column is present.
// It is deliberately based on INFORMATION_SCHEMA, not the generated schema
// catalog, because supported restarts may skip apply on already-bootstrapped
// databases and therefore run new code against an older physical table.
func (r *OutboxRepository) HasRecordKeyColumn(ctx context.Context) (bool, error) {
	r.recordKeyColumnMu.Lock()
	if r.recordKeyColumnKnown {
		present := r.recordKeyColumnPresent
		r.recordKeyColumnMu.Unlock()
		return present, nil
	}
	r.recordKeyColumnMu.Unlock()

	var count int
	err := r.db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
		r.outboxTableName(),
		constants.FieldSysOutboxEvent_RecordKey,
	).Scan(&count)
	if err != nil {
		return false, err
	}

	present := count > 0
	r.recordKeyColumnMu.Lock()
	r.recordKeyColumnKnown = true
	r.recordKeyColumnPresent = present
	r.recordKeyColumnMu.Unlock()
	return present, nil
}

// ObserveOutboxWork returns only bounded aggregate state; no payload or tenant
// identifier crosses the worker metrics boundary.
func (r *OutboxRepository) ObserveOutboxWork(ctx context.Context) (int64, int64, time.Duration, error) {
	if r == nil || r.db == nil {
		return 0, 0, 0, errors.New("outbox repository is unavailable")
	}
	// Rows outside these three statuses contribute zero to every result. Keep
	// that equivalence in SQL so retained processed history is excluded by the
	// existing (scope, status, created date) index before aggregation.
	queryText := fmt.Sprintf(`SELECT
  COALESCE(SUM(CASE WHEN Status IN ('pending', 'processing') THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN Status = 'failed' THEN 1 ELSE 0 END), 0),
  COALESCE(TIMESTAMPDIFF(MICROSECOND, MIN(CASE WHEN Status IN ('pending', 'processing') THEN CreatedDate END), NOW(6)), 0)
FROM %s
WHERE Status IN ('pending', 'processing', 'failed')`, quoteOutboxTable(r.outboxTableName()))
	args := []any{}
	if r.scope != nil {
		queryText += " AND ExecutionScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?"
		args = append(args, r.scopeDigest, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	}
	var depth, failed, oldestMicros int64
	if err := r.db.QueryRowContext(ctx, queryText, args...).Scan(&depth, &failed, &oldestMicros); err != nil {
		return 0, 0, 0, fmt.Errorf("observe outbox work: %w", err)
	}
	if depth < 0 || failed < 0 || oldestMicros < 0 {
		return 0, 0, 0, errors.New("observe outbox work: invalid aggregate")
	}
	return depth, failed, time.Duration(oldestMicros) * time.Microsecond, nil
}

// HasOutboxWorkHint is the bounded hot-poll read for independently deployed
// workers. It deliberately reuses the authoritative eligibility predicate but
// does not attempt to prove FIFO claimability or ownership: false may defer a
// concurrently arriving row until the next poll, while true is rechecked by
// ClaimNextEventWithRecordFIFO under the deployment and row fences.
//
// Keeping this separate from ObserveOutboxWork avoids running a full
// SUM/CASE/MIN queue-health aggregate every second merely to learn whether a
// claim attempt is worth opening.
func (r *OutboxRepository) HasOutboxWorkHint(ctx context.Context) (bool, error) {
	if r == nil || r.db == nil {
		return false, errors.New("outbox repository is unavailable")
	}
	if r.scope != nil && r.outboxTableName() == tableRecordEventOutbox {
		// Keep the two status ranges separate and force the readiness-verified
		// (scope, status, created date, id) index. TiDB otherwise flattens the
		// eligibility OR and may choose a table scan; an idle scoped worker then
		// rereads retained processed history every poll merely to prove no work.
		// This remains one SQL statement, and the fenced FIFO claim is still the
		// sole positive authority.
		queryText, args := r.scopedRecordEventWorkHintQuery()
		var present bool
		err := r.db.QueryRowContext(ctx, queryText, args...).Scan(&present)
		if err != nil {
			return false, fmt.Errorf("probe scoped record-event work hint: %w", err)
		}
		return present, nil
	}
	eligibilitySQL, args := outboxEligibilityPredicate(r.processingLease())
	queryText := "SELECT EXISTS(SELECT 1 FROM " + quoteOutboxTable(r.outboxTableName()) +
		" WHERE " + eligibilitySQL
	if r.scope != nil {
		queryText += " AND `ExecutionScopeDigest` = ? AND `OrgId` = ? AND `AppInstallationId` = ? AND `RuntimeCapabilityId` = ?"
		args = append(args, r.scopeDigest, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	}
	queryText += " LIMIT 1)"
	var present bool
	if err := r.db.QueryRowContext(ctx, queryText, args...).Scan(&present); err != nil {
		return false, fmt.Errorf("probe outbox work hint: %w", err)
	}
	return present, nil
}

func (r *OutboxRepository) scopedRecordEventWorkHintQuery() (string, []interface{}) {
	pendingSQL, pendingArgs := outboxPendingEligibilityPredicate()
	staleSQL, staleArgs := outboxStaleProcessingEligibilityPredicate(r.processingLease())
	queryText := "SELECT EXISTS(\n" +
		"  SELECT 1 FROM `" + tableRecordEventOutbox + "` FORCE INDEX (`" + recordEventClaimIndex + "`)\n" +
		"  WHERE `ExecutionScopeDigest` = ? AND " + pendingSQL + "\n" +
		"    AND `OrgId` = ? AND `AppInstallationId` = ? AND `RuntimeCapabilityId` = ?\n" +
		"  LIMIT 1\n" +
		") OR EXISTS(\n" +
		"  SELECT 1 FROM `" + tableRecordEventOutbox + "` FORCE INDEX (`" + recordEventClaimIndex + "`)\n" +
		"  WHERE `ExecutionScopeDigest` = ? AND " + staleSQL + "\n" +
		"    AND `OrgId` = ? AND `AppInstallationId` = ? AND `RuntimeCapabilityId` = ?\n" +
		"  LIMIT 1\n" +
		")"
	args := []interface{}{r.scopeDigest}
	args = append(args, pendingArgs...)
	args = append(args, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID, r.scopeDigest)
	args = append(args, staleArgs...)
	args = append(args, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	return queryText, args
}

func quoteOutboxTable(table string) string {
	if table == tableRecordEventOutbox {
		return "`" + tableRecordEventOutbox + "`"
	}
	return "`" + constants.TableOutboxEvent + "`"
}

// Enqueue inserts a new event into the outbox
func (r *OutboxRepository) Enqueue(ctx context.Context, eventType string, payload interface{}) (string, error) {
	executor := r.GetExecutor(ctx)
	id := utils.GenerateID()

	stageStart := time.Now()
	payloadJSON, err := json.Marshal(payload)
	logOutboxRepositoryStageTiming(ctx, "marshalPayload", eventType, 0, stageStart, err)
	if err != nil {
		return "", apperrors.NewInternalError("failed to marshal event payload", err)
	}

	stageStart = time.Now()
	recordKey := outboxRecordKeyFromPayload(payloadJSON, id)
	logOutboxRepositoryStageTiming(ctx, "deriveRecordKey", eventType, 0, stageStart, nil)

	// Status defaults to 'pending'
	status := constants.OutboxStatusPending

	stageStart = time.Now()
	insertValues := map[string]interface{}{
		constants.FieldID:                        id,
		constants.FieldSysOutboxEvent_EventType:  eventType,
		constants.FieldSysOutboxEvent_Payload:    payloadJSON,
		constants.FieldSysOutboxEvent_Status:     status,
		constants.FieldSysOutboxEvent_RetryCount: 0,
		constants.FieldCreatedDate:               query.Raw(outboxDatabaseNowSQL),
		constants.FieldLastModifiedDate:          query.Raw(outboxDatabaseNowSQL),
	}
	if r.outboxTableName() == tableRecordEventOutbox {
		scope, scoped := RuntimeWorkScopeFromContext(ctx)
		if r.scope != nil {
			if !scoped || !sameRuntimeWorkScope(scope, *r.scope) {
				return "", errors.New("record-event outbox: enqueue scope does not match repository scope")
			}
		}
		if scoped {
			scopeDigest, err := scope.Digest()
			if err != nil {
				return "", fmt.Errorf("record-event outbox: derive enqueue execution scope digest: %w", err)
			}
			insertValues["OrgId"] = scope.OrgID
			insertValues["AppInstallationId"] = scope.AppInstallationID
			insertValues["RuntimeCapabilityId"] = scope.RuntimeCapabilityID
			insertValues[recordEventExecutionScopeDigestColumn] = scopeDigest
		}
	}
	logOutboxRepositoryStageTiming(ctx, "buildInsertValues", eventType, len(insertValues), stageStart, nil)

	stageStart = time.Now()
	hasRecordKey, probeErr := r.HasRecordKeyColumn(ctx)
	logOutboxRepositoryStageTiming(ctx, "probeRecordKeyColumn", eventType, len(insertValues), stageStart, probeErr)
	if probeErr != nil {
		log.Printf("⚠️ [Outbox] RecordKey column probe failed; enqueueing with legacy column set: %v", probeErr)
	} else if hasRecordKey {
		insertValues[constants.FieldSysOutboxEvent_RecordKey] = recordKey
	}

	stageStart = time.Now()
	tableName := r.outboxTableName()
	qr := query.Insert(tableName, insertValues).Build()
	logOutboxRepositoryStageTiming(ctx, "buildInsertSQL", eventType, len(insertValues), stageStart, nil)

	stageStart = time.Now()
	_, err = executor.ExecContext(ctx, qr.SQL, qr.Params...)
	logRepositorySQLTiming(ctx, "OutboxRepository.Enqueue", tableName, "write", qr.SQL, len(qr.Params), len(insertValues), 1, stageStart, err)
	if err != nil {
		return "", apperrors.NewInternalError("failed to enqueue event", err)
	}

	return id, nil
}

type outboxRecordPayload struct {
	ObjectAPIName string         `json:"object_api_name"`
	Record        models.SObject `json:"record"`
}

type outboxRecordEnvelope struct {
	PublicPayload outboxRecordPayload `json:"public_payload"`
}

func outboxRecordKeyFromPayload(payloadJSON []byte, fallbackID string) string {
	if len(payloadJSON) == 0 {
		return fallbackID
	}
	var envelope outboxRecordEnvelope
	if err := json.Unmarshal(payloadJSON, &envelope); err == nil {
		if key := outboxRecordKey(envelope.PublicPayload); key != "" {
			return key
		}
	}

	var legacy outboxRecordPayload
	if err := json.Unmarshal(payloadJSON, &legacy); err == nil {
		if key := outboxRecordKey(legacy); key != "" {
			return key
		}
	}
	return fallbackID
}

func outboxRecordKey(payload outboxRecordPayload) string {
	if payload.ObjectAPIName == "" {
		return ""
	}
	recordID := payload.Record.GetString(constants.FieldID)
	if recordID == "" {
		return ""
	}
	return payload.ObjectAPIName + "/" + recordID
}

// BackfillRecordKeys populates RecordKey for legacy outbox rows written before
// the column existed. Only non-terminal rows are backfilled: retained processed
// history is irrelevant to future FIFO decisions and must not make worker
// startup rewrite an unbounded table.
func (r *OutboxRepository) BackfillRecordKeys(ctx context.Context, limit int) (int, error) {
	if limit <= 0 {
		return 0, nil
	}
	hasRecordKey, err := r.HasRecordKeyColumn(ctx)
	if err != nil {
		return 0, apperrors.NewInternalError("failed to probe outbox record key column", err)
	}
	if !hasRecordKey {
		return 0, nil
	}

	qb := query.From(r.outboxTableName()).
		Select([]string{
			constants.FieldID,
			constants.FieldSysOutboxEvent_Payload,
		}).
		WhereRaw("`RecordKey` IS NULL", nil).
		WhereRaw("`Status` IN (?, ?)", []interface{}{
			constants.OutboxStatusPending,
			outboxStatusProcessing,
		})
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.
		OrderBy(constants.FieldCreatedDate, "ASC").
		AddOrderBy(constants.FieldID, "ASC").
		Limit(limit).
		Build()

	rows, err := r.db.QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return 0, apperrors.NewInternalError("failed to query outbox rows missing record key", err)
	}
	defer rows.Close()

	type missingKeyRow struct {
		id      string
		payload []byte
	}
	var missing []missingKeyRow
	for rows.Next() {
		var row missingKeyRow
		if err := rows.Scan(&row.id, &row.payload); err != nil {
			return 0, apperrors.NewInternalError("failed to scan outbox row missing record key", err)
		}
		missing = append(missing, row)
	}
	if err := rows.Err(); err != nil {
		return 0, apperrors.NewInternalError("error iterating outbox rows missing record key", err)
	}

	updated := 0
	for _, row := range missing {
		recordKey := outboxRecordKeyFromPayload(row.payload, row.id)
		qb := query.Update(r.outboxTableName()).
			Set(map[string]interface{}{
				constants.FieldSysOutboxEvent_RecordKey: recordKey,
			}).
			Where(constants.FieldID, row.id).
			WhereRaw("`RecordKey` IS NULL", nil)
		qb = r.constrainToRuntimeWorkScope(qb)
		qr := qb.Build()
		res, err := r.db.ExecContext(ctx, qr.SQL, qr.Params...)
		if err != nil {
			return updated, apperrors.NewInternalError("failed to backfill outbox record key", err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return updated, apperrors.NewInternalError("failed to count outbox record key backfill", err)
		}
		updated += int(n)
	}
	return updated, nil
}

// GetPendingEvents retrieves eligible events ordered by stable creation order.
//
// Rows stuck in Status='processing' past this repository's claim lease are
// included in the result set and reclaimable. The database clock owns the
// lease boundary so API/worker host clock skew cannot reclaim live work.
// ClaimEvent mirrors the predicate so only one worker wins the reclaim race.
func (r *OutboxRepository) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	return r.getPendingEvents(ctx, limit, false)
}

// GetPendingEventsWithRecordFIFO retrieves eligible events for the pull-worker
// path. Same-record FIFO is enforced here rather than in worker memory. A row
// is ineligible while any older row with the same RecordKey is non-terminal,
// including a stale 'processing' row. Legacy rows with null RecordKey are
// unknown-key heads: they conservatively block younger work until they reach a
// terminal state or the background backfill gives them a key. The stale head
// itself is reclaimable; younger siblings wait until it reaches 'processed' or
// 'failed'.
func (r *OutboxRepository) GetPendingEventsWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	hasRecordKey, err := r.HasRecordKeyColumn(ctx)
	if err != nil {
		return nil, apperrors.NewInternalError("failed to probe outbox record key column", err)
	}
	if !hasRecordKey {
		return nil, ErrOutboxRecordKeyUnavailable
	}
	return r.getPendingEvents(ctx, limit, true)
}

func (r *OutboxRepository) getPendingEvents(ctx context.Context, limit int, recordFIFO bool) ([]models.OutboxEvent, error) {
	cols := []string{
		constants.FieldID,
		constants.FieldSysOutboxEvent_EventType,
		constants.FieldSysOutboxEvent_Payload,
		constants.FieldSysOutboxEvent_RetryCount,
	}
	qr := r.pendingEventsQuery(limit, recordFIFO, cols)

	rows, err := r.db.QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return nil, apperrors.NewInternalError("failed to query pending events", err)
	}
	defer rows.Close()

	var events []models.OutboxEvent
	for rows.Next() {
		var e models.OutboxEvent

		if err := rows.Scan(&e.ID, &e.EventType, &e.Payload, &e.RetryCount); err != nil {
			log.Printf("Warning: failed to scan outbox event: %v", err)
			continue
		}
		events = append(events, e)
	}
	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError("error iterating outbox events", err)
	}

	return events, nil
}

func (r *OutboxRepository) pendingEventsQuery(limit int, recordFIFO bool, cols []string) query.QueryResult {
	eligibilitySQL, eligibilityArgs := outboxEligibilityPredicate(r.processingLease())
	return r.fifoEventsQuery(limit, recordFIFO, cols, eligibilitySQL, eligibilityArgs)
}

func (r *OutboxRepository) pendingOnlyEventsQuery(limit int, recordFIFO bool, cols []string) query.QueryResult {
	predicate, args := outboxPendingEligibilityPredicate()
	return r.fifoEventsQuery(
		limit,
		recordFIFO,
		cols,
		predicate,
		args,
	)
}

func (r *OutboxRepository) staleProcessingOnlyEventsQuery(limit int, recordFIFO bool, cols []string) query.QueryResult {
	predicate, args := outboxStaleProcessingEligibilityPredicate(r.processingLease())
	return r.fifoEventsQuery(
		limit,
		recordFIFO,
		cols,
		predicate,
		args,
	)
}

func (r *OutboxRepository) hasStaleProcessingWorkHint(ctx context.Context) (bool, error) {
	if r == nil || r.scope == nil || r.outboxTableName() != tableRecordEventOutbox {
		return false, errors.New("stale record-event hint requires a scoped repository")
	}
	predicate, predicateArgs := outboxStaleProcessingEligibilityPredicate(r.processingLease())
	queryText := "SELECT EXISTS(" +
		"SELECT 1 FROM `" + tableRecordEventOutbox + "` FORCE INDEX (`" + recordEventClaimIndex + "`) " +
		"WHERE `ExecutionScopeDigest` = ? AND " + predicate + " " +
		"AND `OrgId` = ? AND `AppInstallationId` = ? AND `RuntimeCapabilityId` = ? LIMIT 1)"
	args := []interface{}{r.scopeDigest}
	args = append(args, predicateArgs...)
	args = append(args, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	var present bool
	err := r.GetExecutor(ctx).QueryRowContext(ctx, queryText, args...).Scan(&present)
	if err != nil {
		return false, fmt.Errorf("probe stale record-event work: %w", err)
	}
	return present, nil
}

func (r *OutboxRepository) fifoEventsQuery(limit int, recordFIFO bool, cols []string, eligibilitySQL string, eligibilityArgs []interface{}) query.QueryResult {
	tableName := r.outboxTableName()
	qb := query.From(tableName).
		Select(cols).
		WhereRaw(eligibilitySQL, eligibilityArgs)
	if r.scope != nil && tableName == tableRecordEventOutbox {
		// Keep both FIFO arms on the readiness-owned scope/status index. The
		// explicit scope tuple remains a residual fail-closed authority check.
		qb = qb.UseIndex(recordEventClaimIndex)
	}
	qb = r.constrainToRuntimeWorkScope(qb)

	if recordFIFO {
		scopePredicate := ""
		scopeArgs := []interface{}{}
		if r.scope != nil {
			scopePredicate = "AND older.`ExecutionScopeDigest` = ? AND older.`OrgId` = ? AND older.`AppInstallationId` = ? AND older.`RuntimeCapabilityId` = ? "
			scopeArgs = append(scopeArgs, r.scopeDigest, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
		}
		olderTable := "`" + tableName + "` older"
		if r.scope != nil && tableName == tableRecordEventOutbox {
			olderTable += " USE INDEX (`" + recordEventClaimIndex + "`)"
		}
		qb = qb.WhereRaw(
			"NOT EXISTS ("+
				"SELECT 1 FROM "+olderTable+" "+
				"WHERE (older.`CreatedDate` < `"+tableName+"`.`CreatedDate` "+
				"OR (older.`CreatedDate` = `"+tableName+"`.`CreatedDate` AND older.`Id` < `"+tableName+"`.`Id`)) "+
				"AND older.`Status` NOT IN (?, ?) "+scopePredicate+
				"AND (`"+tableName+"`.`RecordKey` IS NULL "+
				"OR older.`RecordKey` IS NULL "+
				"OR older.`RecordKey` = `"+tableName+"`.`RecordKey`))",
			append([]interface{}{
				constants.OutboxStatusProcessed,
				constants.OutboxStatusFailed,
			}, scopeArgs...),
		)
	}

	qr := qb.
		OrderBy(constants.FieldCreatedDate, "ASC").
		AddOrderBy(constants.FieldID, "ASC").
		Limit(limit).
		Build()
	return qr
}

// ClaimEventBatchWithRecordFIFO selects and claims up to limit FIFO-eligible
// heads inside the caller's pessimistic admission transaction. The correlated
// FIFO predicate returns at most one non-terminal head per RecordKey, so one
// set read preserves same-record order while admitting independent keys.
//
// Normal pending rows retain priority. When they do not fill the requested
// capacity, one separate NOWAIT set read may fill the remainder with stale
// processing heads. ClaimEvent remains the compare-and-transition proof for
// each selected row. Selection and every transition commit atomically, and no
// unleased in-memory item leaves this method.
func (r *OutboxRepository) ClaimEventBatchWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, []models.OutboxClaim, error) {
	if r == nil || r.scope == nil {
		return nil, nil, errors.New("atomic record-event claim requires a scoped repository")
	}
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); !ok || tx == nil {
		return nil, nil, errors.New("atomic record-event claim requires an active transaction")
	}
	if pessimistic, _ := ctx.Value(pessimisticTxMarkerKey{}).(bool); !pessimistic {
		return nil, nil, errors.New("atomic record-event claim requires a pessimistic transaction")
	}
	if limit <= 0 {
		return nil, nil, errors.New("atomic record-event claim batch limit must be positive")
	}

	cols := []string{
		constants.FieldID,
		constants.FieldSysOutboxEvent_EventType,
		constants.FieldSysOutboxEvent_Payload,
		constants.FieldSysOutboxEvent_RetryCount,
		constants.FieldSysOutboxEvent_RecordKey,
	}
	pendingQuery := r.pendingOnlyEventsQuery(limit, true, cols)
	pendingQuery.SQL += " FOR UPDATE NOWAIT"
	events, contended, err := r.readFIFOClaimCandidates(ctx, pendingQuery)
	if err != nil {
		return nil, nil, apperrors.NewInternalError("failed to select pending FIFO outbox claim batch", err)
	}
	if contended {
		return nil, nil, nil
	}
	claims, err := r.claimFIFOEvents(ctx, events)
	if err != nil {
		return nil, nil, err
	}
	if len(events) < limit {
		stalePresent, err := r.hasStaleProcessingWorkHint(ctx)
		if err != nil {
			return nil, nil, apperrors.NewInternalError("failed to probe stale FIFO outbox work", err)
		}
		if !stalePresent {
			return events, claims, nil
		}
		staleQuery := r.staleProcessingOnlyEventsQuery(limit-len(events), true, cols)
		staleQuery.SQL += " FOR UPDATE NOWAIT"
		staleEvents, staleContended, err := r.readFIFOClaimCandidates(ctx, staleQuery)
		if err != nil {
			return nil, nil, apperrors.NewInternalError("failed to select stale FIFO outbox claim batch", err)
		}
		if !staleContended {
			staleClaims, err := r.claimFIFOEvents(ctx, staleEvents)
			if err != nil {
				return nil, nil, err
			}
			events = append(events, staleEvents...)
			claims = append(claims, staleClaims...)
		}
	}
	return events, claims, nil
}

func (r *OutboxRepository) readFIFOClaimCandidates(ctx context.Context, qr query.QueryResult) ([]models.OutboxEvent, bool, error) {
	rows, err := r.GetExecutor(ctx).QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		if isOutboxNowaitConflict(err) {
			return nil, true, nil
		}
		return nil, false, err
	}
	defer rows.Close()
	events := make([]models.OutboxEvent, 0)
	for rows.Next() {
		var event models.OutboxEvent
		var recordKey sql.NullString
		if err := rows.Scan(&event.ID, &event.EventType, &event.Payload, &event.RetryCount, &recordKey); err != nil {
			return nil, false, err
		}
		event.RecordKey = recordKey.String
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		if isOutboxNowaitConflict(err) {
			return nil, true, nil
		}
		return nil, false, err
	}
	return events, false, nil
}

func isOutboxNowaitConflict(err error) bool {
	var mysqlErr *mysqldriver.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == 3572
}

func (r *OutboxRepository) claimFIFOEvents(ctx context.Context, events []models.OutboxEvent) ([]models.OutboxClaim, error) {
	claims := make([]models.OutboxClaim, 0, len(events))
	for _, event := range events {
		claim, err := r.ClaimEvent(ctx, event.ID)
		if err != nil {
			return nil, fmt.Errorf("claim selected FIFO outbox event %s: %w", event.ID, err)
		}
		claim.RecordKey = event.RecordKey
		claims = append(claims, claim)
	}
	return claims, nil
}

// ClaimNextEventWithRecordFIFO is the compatibility single-item view of the
// same set-claim path. It exists for deterministic callers; independently
// deployed workers use ClaimEventBatchWithRecordFIFO directly.
func (r *OutboxRepository) ClaimNextEventWithRecordFIFO(ctx context.Context) (models.OutboxEvent, models.OutboxClaim, bool, error) {
	events, claims, err := r.ClaimEventBatchWithRecordFIFO(ctx, 1)
	if err != nil || len(events) == 0 {
		return models.OutboxEvent{}, models.OutboxClaim{}, false, err
	}
	if len(events) != 1 || len(claims) != 1 {
		return models.OutboxEvent{}, models.OutboxClaim{}, false, errors.New("atomic record-event single claim returned inconsistent batch")
	}
	return events[0], claims[0], true, nil
}

// ClaimEvent atomically transitions a pending (or stale 'processing') event
// to 'processing' so only one worker dispatches it.
//
// The original implementation used SELECT ... FOR UPDATE SKIP LOCKED, which
// silently degrades to a non-blocking read under TiDB optimistic txn mode
// (the cluster default on some TiDB Cloud Serverless branches; see
// backend/internal/features/automation/autonumber/reserve.go for the
// precedent fix). In optimistic mode two concurrent workers could each
// believe they had exclusively claimed the row and double-dispatch.
//
// The UPDATE ... WHERE (Status='pending' OR stale-processing) form below
// takes an exclusive row lock in BOTH txn modes. Of the N concurrent
// callers, exactly one sees RowsAffected()==1; the others see 0 because
// the WHERE clause no longer matches once the first UPDATE commits (or
// blocks until it does under pessimistic mode). The sentinel
// ErrOutboxEventAlreadyClaimed signals the benign-skip case so the
// service layer can distinguish it from a genuine DB error.
//
// The stale-processing predicate mirrors GetPendingEvents so a row whose
// previous worker died between claim and status-finalise becomes
// reclaimable after this repository's processing lease — no separate sweeper
// goroutine is needed. LastModifiedDate is updated on every successful
// claim so a long-running-but-alive dispatch (unlikely; each event has a
// short publish timeout) does NOT get its row reclaimed mid-flight.
func (r *OutboxRepository) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	eligibilitySQL, eligibilityArgs := outboxEligibilityPredicate(r.processingLease())
	claim := models.OutboxClaim{EventID: id}
	updates := map[string]interface{}{
		constants.FieldSysOutboxEvent_Status: outboxStatusProcessing,
		constants.FieldLastModifiedDate:      query.Raw(outboxDatabaseNowSQL),
	}
	if r.scope != nil {
		token, err := uuid.NewRandom()
		if err != nil {
			return models.OutboxClaim{}, fmt.Errorf("record-event outbox: generate claim token: %w", err)
		}
		claim.Token = token.String()
		updates[recordEventClaimTokenColumn] = claim.Token
	}
	qb := query.Update(r.outboxTableName()).
		Set(updates).
		Where(constants.FieldID, id).
		WhereRaw(eligibilitySQL, eligibilityArgs)
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.Build()

	exec := r.GetExecutor(ctx)
	res, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return models.OutboxClaim{}, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return models.OutboxClaim{}, err
	}
	if n == 0 {
		return models.OutboxClaim{}, ErrOutboxEventAlreadyClaimed
	}
	retryCount, err := r.claimRetryCount(ctx, claim)
	if err != nil {
		return models.OutboxClaim{}, err
	}
	claim.RetryCount = retryCount
	return claim, nil
}

// LockOutboxClaimCandidate acquires the candidate row before a Runtime worker
// takes the singleton admission gate. NOWAIT is essential: an expired lease can
// still belong to a live effect transaction that holds this row, and waiting
// for it while holding the gate would recreate global head-of-line blocking.
func (r *OutboxRepository) LockOutboxClaimCandidate(ctx context.Context, id string) error {
	if r == nil || id == "" {
		return ErrOutboxEventAlreadyClaimed
	}
	statement := "SELECT `Id` FROM `" + r.outboxTableName() + "` WHERE `Id` = ?"
	args := []interface{}{id}
	if r.scope != nil {
		statement += " AND `ExecutionScopeDigest` = ? AND `OrgId` = ? AND `AppInstallationId` = ? AND `RuntimeCapabilityId` = ?"
		args = append(args, r.scopeDigest, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	}
	statement += " FOR UPDATE NOWAIT"
	var lockedID string
	err := r.GetExecutor(ctx).QueryRowContext(ctx, statement, args...).Scan(&lockedID)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrOutboxEventAlreadyClaimed
	}
	var mysqlErr *mysqldriver.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 3572 {
		return ErrOutboxEventAlreadyClaimed
	}
	if err != nil {
		return fmt.Errorf("lock outbox claim candidate: %w", err)
	}
	if lockedID != id {
		return ErrOutboxEventAlreadyClaimed
	}
	return nil
}

func (r *OutboxRepository) claimRetryCount(ctx context.Context, claim models.OutboxClaim) (int, error) {
	qb := query.From(r.outboxTableName()).
		Select([]string{constants.FieldSysOutboxEvent_RetryCount}).
		Where(constants.FieldID, claim.EventID).
		Where(constants.FieldSysOutboxEvent_Status, outboxStatusProcessing)
	qb = r.constrainToClaim(qb, claim)
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.Build()

	var claimedID string
	var retryCount int
	if err := r.GetExecutor(ctx).QueryRowContext(ctx, qr.SQL, qr.Params...).Scan(&claimedID, &retryCount); err != nil {
		return 0, fmt.Errorf("read claimed outbox retry count: %w", err)
	}
	if claimedID != claim.EventID {
		return 0, errors.New("read claimed outbox retry count: row identity mismatch")
	}
	if retryCount < 0 {
		return 0, errors.New("read claimed outbox retry count: negative durable count")
	}
	return retryCount, nil
}

func (r *OutboxRepository) constrainToClaim(qb *query.Builder, claim models.OutboxClaim) *query.Builder {
	if r == nil || r.scope == nil {
		return qb
	}
	if claim.EventID == "" || claim.Token == "" {
		// An impossible predicate makes an incomplete externally deployed claim
		// fail closed without teaching the query builder a special error path.
		return qb.WhereRaw("1 = 0", nil)
	}
	return qb.Where(recordEventClaimTokenColumn, claim.Token)
}

// UpdateStatus updates the status and related fields of an event
func (r *OutboxRepository) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	qb := query.Update(r.outboxTableName()).
		Where(constants.FieldID, claim.EventID).
		Where(constants.FieldSysOutboxEvent_Status, outboxStatusProcessing)
	qb = r.constrainToClaim(qb, claim)

	updates := map[string]interface{}{
		constants.FieldSysOutboxEvent_Status: status,
		constants.FieldLastModifiedDate:      query.Raw(outboxDatabaseNowSQL),
	}

	if status == string(constants.OutboxStatusProcessed) {
		updates[constants.FieldSysOutboxEvent_ProcessedDate] = query.Raw(outboxDatabaseNowSQL)
	} else if status == string(constants.OutboxStatusFailed) {
		updates[constants.FieldSysOutboxEvent_ErrorMessage] = errMessage
	} else {
		return apperrors.NewValidationError("status", fmt.Sprintf("unsupported status update: %s", status))
	}
	if r.scope != nil {
		updates[recordEventClaimTokenColumn] = nil
	}
	qb = r.constrainToRuntimeWorkScope(qb)

	qr := qb.Set(updates).Build()

	exec := r.GetExecutor(ctx)
	res, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrOutboxEventAlreadyClaimed
	}
	return nil
}

// MarkFailed atomically records the terminal failed state and increments the
// durable attempt count. A Runtime poison repair is compare-and-set against
// both that count and ErrorMessage, so a caller's stale discovery value must
// never be allowed to author the receipt.
func (r *OutboxRepository) MarkFailed(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	updates := map[string]interface{}{
		constants.FieldSysOutboxEvent_Status:       constants.OutboxStatusFailed,
		constants.FieldSysOutboxEvent_RetryCount:   query.Raw("`RetryCount` + 1"),
		constants.FieldSysOutboxEvent_ErrorMessage: errMessage,
		constants.FieldLastModifiedDate:            query.Raw(outboxDatabaseNowSQL),
	}
	if r.scope != nil {
		updates[recordEventClaimTokenColumn] = nil
	}
	qb := query.Update(r.outboxTableName()).
		Set(updates).
		Where(constants.FieldID, claim.EventID).
		Where(constants.FieldSysOutboxEvent_Status, outboxStatusProcessing)
	qb = r.constrainToClaim(qb, claim)
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.Build()

	res, err := r.GetExecutor(ctx).ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrOutboxEventAlreadyClaimed
	}
	return nil
}

// IncrementRetry increments the retry count, updates the error message,
// and returns the row to 'pending' so the next poll picks it up. The row
// was moved to 'processing' by ClaimEvent; without this reset it would
// stay stuck and never retry.
func (r *OutboxRepository) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	updates := map[string]interface{}{
		constants.FieldSysOutboxEvent_Status:       constants.OutboxStatusPending,
		constants.FieldSysOutboxEvent_RetryCount:   query.Raw("`RetryCount` + 1"),
		constants.FieldSysOutboxEvent_ErrorMessage: errMessage,
		constants.FieldLastModifiedDate:            query.Raw(outboxDatabaseNowSQL),
	}
	if r.scope != nil {
		updates[recordEventClaimTokenColumn] = nil
	}
	qb := query.Update(r.outboxTableName()).
		Set(updates).
		Where(constants.FieldID, claim.EventID).
		Where(constants.FieldSysOutboxEvent_Status, outboxStatusProcessing)
	qb = r.constrainToClaim(qb, claim)
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.Build()

	exec := r.GetExecutor(ctx)
	res, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrOutboxEventAlreadyClaimed
	}
	return nil
}

// outboxEligibilityPredicate is the single durable eligibility rule shared by
// candidate discovery and the compare-and-transition claim. Both lease expiry
// and retry visibility are evaluated by TiDB so worker clock skew cannot make
// work eligible early, defer it indefinitely, or let discovery disagree with
// the atomic claim.
func outboxEligibilityPredicate(processingLease time.Duration) (string, []interface{}) {
	expiredSQL, expiredArgs := outboxProcessingLeasePredicate(processingLease, false)
	return "((`Status` = ? OR (`Status` = ? AND " + expiredSQL + ")) AND (`Status` <> ? OR `RetryCount` = 0 OR `LastModifiedDate` < TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6))))", []interface{}{
		constants.OutboxStatusPending,
		outboxStatusProcessing,
		expiredArgs[0],
		constants.OutboxStatusPending,
		-(time.Duration(OutboxRetryVisibleAfterSeconds) * time.Second).Microseconds(),
	}
}

func outboxPendingEligibilityPredicate() (string, []interface{}) {
	return "`Status` = ? AND (`RetryCount` = 0 OR `LastModifiedDate` < TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)))", []interface{}{
		constants.OutboxStatusPending,
		-(time.Duration(OutboxRetryVisibleAfterSeconds) * time.Second).Microseconds(),
	}
}

func outboxStaleProcessingEligibilityPredicate(processingLease time.Duration) (string, []interface{}) {
	expiredSQL, expiredArgs := outboxProcessingLeasePredicate(processingLease, false)
	return "`Status` = ? AND " + expiredSQL, append([]interface{}{outboxStatusProcessing}, expiredArgs...)
}

// outboxProcessingLeasePredicate is the sole boundary between a live claim and
// an expired claim. Both sides use TiDB time and complementary comparisons, so
// refresh and reclaim cannot each accept (or each reject) the same timestamp.
// The existing reclaim contract treats equality as live: expired is < cutoff,
// live is >= cutoff.
func outboxProcessingLeasePredicate(processingLease time.Duration, live bool) (string, []interface{}) {
	comparison := "<"
	if live {
		comparison = ">="
	}
	return "`LastModifiedDate` " + comparison + " TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6))", []interface{}{
		-processingLease.Microseconds(),
	}
}

// CleanupProcessed deletes old processed events
func (r *OutboxRepository) CleanupProcessed(ctx context.Context, cutoff time.Time) (int64, error) {
	qb := query.Delete(r.outboxTableName()).
		Where(constants.FieldSysOutboxEvent_Status, constants.OutboxStatusProcessed).
		Where(constants.FieldSysOutboxEvent_ProcessedDate, cutoff, "<")
	qb = r.constrainToRuntimeWorkScope(qb)
	qr := qb.Build()

	result, err := r.db.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}

// CleanupProcessedBatch deletes one deterministic, bounded batch of completed
// rows using database time. Independently deployed workers call this method
// inside the same pessimistic transaction as their deployment-ownership fence;
// using GetExecutor is therefore essential. The legacy CleanupProcessed path
// is retained for the monolith until its separate ownership cutover.
func (r *OutboxRepository) CleanupProcessedBatch(ctx context.Context, retention time.Duration, batchSize int) (int64, error) {
	if r == nil || r.db == nil {
		return 0, errors.New("outbox cleanup: repository is unavailable")
	}
	if retention < time.Second || batchSize <= 0 {
		return 0, errors.New("outbox cleanup: retention and batch size must be positive and bounded")
	}
	queryText := fmt.Sprintf(`DELETE FROM %s
WHERE Status = ?
  AND ProcessedDate < TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6))
%s
ORDER BY ProcessedDate ASC, Id ASC
LIMIT ?`, quoteOutboxTable(r.outboxTableName()), func() string {
		if r.scope != nil {
			return "  AND ExecutionScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?"
		}
		return ""
	}())
	args := []any{constants.OutboxStatusProcessed, -retention.Microseconds()}
	if r.scope != nil {
		args = append(args, r.scopeDigest, r.scope.OrgID, r.scope.AppInstallationID, r.scope.RuntimeCapabilityID)
	}
	args = append(args, batchSize)
	result, err := r.GetExecutor(ctx).ExecContext(ctx, queryText, args...)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}