package events

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"log"
	"os"
	"strconv"
	"sync"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

// OutboxEvent status constants
const (
	OutboxStatusPending   = "pending"
	OutboxStatusProcessed = "processed"
	OutboxStatusFailed    = "failed"
	MaxRetryAttempts      = 5
	// outboxEffectTransactionRetries bounds ordinary lock-contention replay.
	// A late field-history prelock plan converges independently inside
	// WithPessimisticRetry and therefore does not consume this budget.
	outboxEffectTransactionRetries = 3

	// defaultOutboxParallelism bounds concurrent outbox dispatch: legacy batch
	// slots when OUTBOX_PULL_WORKER is off, or long-lived pull workers when it
	// is on. ClaimEvent atomically transitions each row from 'pending' ->
	// 'processing' before handlers run, so concurrent dispatch of distinct
	// eligible events is safe; this cap just limits goroutine / DB-connection
	// fan-out. The production DB pool is 100 conns, so 32 leaves headroom for
	// foreground API requests while preventing record-triggered automation from
	// building multi-minute outbox lag under watchdog/e2e bursts.
	defaultOutboxParallelism         = 32
	outboxRecordKeyBackfillBatchSize = 1000
	// minOutboxParallelism / maxOutboxParallelism clamp the env-override
	// so a typo in OUTBOX_WORKER_PARALLELISM can't collapse throughput
	// to zero or exhaust the DB pool.
	minOutboxParallelism = 1
	maxOutboxParallelism = 64

	// outboxParallelismEnv is the env var that overrides
	// defaultOutboxParallelism. Parsed once at service construction.
	outboxParallelismEnv = "OUTBOX_WORKER_PARALLELISM"

	// outboxPullWorkerEnv gates the new independent pull-worker drainer.
	// Until the rollout gate is green, production can keep the legacy batch
	// drainer while the new code path is exercised explicitly.
	outboxPullWorkerEnv = "OUTBOX_PULL_WORKER"

	// recordEventOutboxEnv routes record-trigger events to a dedicated physical
	// outbox table when a composition root wires one. Default off keeps the
	// legacy single-table outbox path.
	recordEventOutboxEnv = "NEXUSCRM_RECORD_EVENT_OUTBOX_ENABLED"

	// RuntimeRecordEventDispatchWorkFamily owns metadata-driven record-event
	// interpretation from the dedicated transactional outbox.
	RuntimeRecordEventDispatchWorkFamily = "runtime.record-event-dispatch.v1"
)

var outboxProcessingHeartbeatInterval = time.Duration(persistence.OutboxStaleProcessingSeconds/3) * time.Second

// OutboxMetadataLookup is the minimal MetadataService surface that the
// outbox needs to validate the dynamic arm of registered no-subscriber objects.
// Defined as an interface (not a hard dependency on *MetadataService) so
// production wiring stays in service_manager_init.go and tests can mock
// the lookup without dragging in the whole metadata layer.
type OutboxMetadataLookup interface {
	HasActiveRecordTriggeredFlow(ctx context.Context, objectAPIName string) bool
}

// OutboxService handles transactional event storage and async publishing.
// It implements the Outbox Pattern for guaranteed event delivery.
type OutboxService struct {
	repo              ports.OutboxRepository
	workerRepo        ports.OutboxRepository
	recordEventOutbox *OutboxService
	// dedicatedRecordEventRouting is composition-owned activation. The
	// independent Runtime mutation role sets it explicitly so its durable
	// record-trigger work can never fall back to the legacy monolith's generic
	// outbox table. The legacy host continues to use the rollout environment
	// gate until its own ownership flip.
	dedicatedRecordEventRouting bool
	eventBus                    ports.EventPublisher
	txManager                   ports.TransactionManager
	recordPayloadLoader         RecordPayloadLoader

	// metadata is OPTIONAL. When nil, the no-subscriber object
	// filter falls through to the enqueue path (fail-safe: do not
	// silently drop events when we cannot prove there is no subscriber).
	// Production wires this via WithMetadataLookup; most unit tests
	// leave it nil because they do not exercise the noise path.
	metadata OutboxMetadataLookup

	// durableExecutionBinder is configured only by an independently owned
	// metadata-driven worker. The envelope remains opaque to events; the
	// Catalog/Runtime composition verifies and installs the complete exact
	// execution envelope (metadata view plus any durable routing projection).
	// durableExecutionBinder opens and verifies immutable metadata outside the
	// database effect/ack transaction. The returned installer performs no I/O;
	// it only binds the already verified execution state to the transaction
	// context. This keeps artifact-store latency out of the row-lock window.
	durableExecutionBinder DurableExecutionBinder
	durableAdmissionFence  func(context.Context) error
	durableAttemptFence    func(context.Context, string) error

	// parallelism bounds legacy batch slots or long-lived pull workers. Read
	// from OUTBOX_WORKER_PARALLELISM at construction, clamped to
	// [minOutboxParallelism, maxOutboxParallelism].
	parallelism int
	maxAttempts int

	// Worker control
	stopCh   chan struct{}
	stopOnce sync.Once
	wg       sync.WaitGroup
}

// DurableExecutionInstaller installs already-verified durable execution state
// into the effect transaction context. It must not perform I/O, but it may
// fail closed instead of silently dropping one required context projection.
type DurableExecutionInstaller func(context.Context) (context.Context, error)

// DurableExecutionBinder opens immutable execution inputs before the effect
// transaction, then returns an installer for the complete execution envelope.
// Events deliberately treats the envelope as opaque; its owner is responsible
// for making every required projection available to downstream effects.
type DurableExecutionBinder func(context.Context, string) (DurableExecutionInstaller, error)

type OutboxWorkState struct {
	QueueDepth int64
	Failed     int64
	OldestAge  time.Duration
}

type outboxWorkObserver interface {
	ObserveOutboxWork(context.Context) (int64, int64, time.Duration, error)
}

type outboxWorkHinter interface {
	HasOutboxWorkHint(context.Context) (bool, error)
}

type recordEventWorkerReadiness interface {
	CheckRecordEventWorkerReady(context.Context) error
}

type processedOutboxBatchCleaner interface {
	CleanupProcessedBatch(context.Context, time.Duration, int) (int64, error)
}

type processingClaimHeartbeatPolicy interface {
	ProcessingClaimHeartbeatInterval() time.Duration
}

type outboxClaimCandidateLocker interface {
	LockOutboxClaimCandidate(context.Context, string) error
}

type outboxFIFOAtomicBatchClaimer interface {
	ClaimEventBatchWithRecordFIFO(context.Context, int) ([]models.OutboxEvent, []models.OutboxClaim, error)
}

// DurableClaimedRecordEvent is an opaque processing lease returned only after
// the deployment-fenced claim transaction commits. The independently deployed
// Worker may keep it in memory only while an immediately available effect slot
// starts; a process crash leaves the durable processing lease reclaimable by
// the repository's existing stale-claim path.
type DurableClaimedRecordEvent struct {
	event models.OutboxEvent
	claim models.OutboxClaim
}

func (os *OutboxService) ObserveWork(ctx context.Context) (OutboxWorkState, error) {
	observer, ok := os.processingRepo().(outboxWorkObserver)
	if !ok {
		return OutboxWorkState{}, errors.New("outbox work observation is unavailable")
	}
	depth, failed, oldest, err := observer.ObserveOutboxWork(ctx)
	return OutboxWorkState{QueueDepth: depth, Failed: failed, OldestAge: oldest}, err
}

// HasWork is a negative-only admission hint. The repository's fenced claim
// remains the sole authority, so an observation failure must cause callers to
// fall through to that claim rather than suppress work.
func (os *OutboxService) HasWork(ctx context.Context) (bool, error) {
	hinter, ok := os.processingRepo().(outboxWorkHinter)
	if !ok {
		return false, errors.New("outbox work hint is unavailable")
	}
	return hinter.HasOutboxWorkHint(ctx)
}

// CheckDurableRecordEventReady ensures an independent worker has every
// fail-closed dependency before it advertises readiness. In particular, a
// legacy/global outbox repository, absent immutable binding, or absent
// deployment fence cannot silently become the serving path.
func (os *OutboxService) CheckDurableRecordEventReady(ctx context.Context) error {
	if os == nil || os.txManager == nil || os.durableExecutionBinder == nil || os.durableAdmissionFence == nil || os.durableAttemptFence == nil || os.maxAttempts <= 0 {
		return errors.New("record-event worker readiness: transaction manager, binding, fence, and poison threshold are required")
	}
	if _, ok := os.processingRepo().(outboxFIFOAtomicBatchClaimer); !ok {
		return errors.New("record-event worker readiness: atomic FIFO claim is unavailable")
	}
	ready, ok := os.processingRepo().(recordEventWorkerReadiness)
	if !ok {
		return errors.New("record-event worker readiness: dedicated repository check is unavailable")
	}
	return ready.CheckRecordEventWorkerReady(ctx)
}

// WithDurableExecutionBinder requires every processed row to carry one exact
// metadata binding. It does not affect enqueue-only Runtime API composition.
func (os *OutboxService) WithDurableExecutionBinder(binder DurableExecutionBinder) *OutboxService {
	if os != nil {
		os.durableExecutionBinder = binder
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithDurableExecutionBinder(binder)
		}
	}
	return os
}

// WithDurableAdmissionFence installs the singleton transfer gate used only by
// the short claim transaction. Long-running effects use keyed fence slots.
func (os *OutboxService) WithDurableAdmissionFence(fence func(context.Context) error) *OutboxService {
	if os != nil && fence != nil {
		os.durableAdmissionFence = fence
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithDurableAdmissionFence(fence)
		}
	}
	return os
}

// WithDurableAttemptFence retains the single-fence composition used by legacy
// independently owned work. Record-event Runtime composition uses the keyed
// variant below so long effects do not share the admission row.
func (os *OutboxService) WithDurableAttemptFence(fence func(context.Context) error) *OutboxService {
	if os != nil && fence != nil {
		os.durableAdmissionFence = fence
		os.durableAttemptFence = func(ctx context.Context, _ string) error { return fence(ctx) }
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithDurableAttemptFence(fence)
		}
	}
	return os
}

// WithDurableKeyedAttemptFence installs the fixed-slot effect fence used by
// independent record-event lanes.
func (os *OutboxService) WithDurableKeyedAttemptFence(fence func(context.Context, string) error) *OutboxService {
	if os != nil && fence != nil {
		os.durableAttemptFence = fence
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithDurableKeyedAttemptFence(fence)
		}
	}
	return os
}

// NewOutboxService creates a new OutboxService
func NewOutboxService(repo ports.OutboxRepository, eventBus ports.EventPublisher, txManager ports.TransactionManager) *OutboxService {
	return &OutboxService{
		repo:        repo,
		eventBus:    eventBus,
		txManager:   txManager,
		parallelism: resolveOutboxParallelism(),
		maxAttempts: MaxRetryAttempts,
		stopCh:      make(chan struct{}),
	}
}

// WithMaxRetryAttempts sets an explicit bounded poison threshold for one
// independently deployed worker role. Legacy composition retains the existing
// default.
func (os *OutboxService) WithMaxRetryAttempts(max int) *OutboxService {
	if os != nil && max > 0 && max <= 20 {
		os.maxAttempts = max
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithMaxRetryAttempts(max)
		}
	}
	return os
}

// WithMetadataLookup wires the OutboxMetadataLookup that arms the
// dynamic side of the no-subscriber object filter. Returns the
// receiver so callers can chain it onto NewOutboxService.
//
// When unset, the filter degrades to the safe fallback (enqueue, never
// drop). Production must call this with the live MetadataService so the
// filter actually fires; tests can leave it unset OR pass a stub when
// they specifically exercise the drop path.
func (os *OutboxService) WithMetadataLookup(m OutboxMetadataLookup) *OutboxService {
	os.metadata = m
	if os.recordEventOutbox != nil {
		os.recordEventOutbox.WithMetadataLookup(m)
	}
	return os
}

// WithWorkerRepository routes background outbox draining/bookkeeping through a
// dedicated repository. Foreground enqueue keeps using repo so it can
// participate in the caller's write transaction.
func (os *OutboxService) WithWorkerRepository(repo ports.OutboxRepository) *OutboxService {
	if repo != nil {
		os.workerRepo = repo
	}
	return os
}

// WithRecordEventRepositories wires a dedicated outbox table for record-trigger
// events. The route is still gated by NEXUSCRM_RECORD_EVENT_OUTBOX_ENABLED so
// production can deploy the code and schema before moving traffic.
func (os *OutboxService) WithRecordEventRepositories(repo, workerRepo ports.OutboxRepository) *OutboxService {
	if repo == nil {
		return os
	}
	child := NewOutboxService(repo, os.eventBus, os.txManager).
		WithMetadataLookup(os.metadata).
		WithWorkerRepository(workerRepo).
		WithDurableExecutionBinder(os.durableExecutionBinder).
		WithDurableAdmissionFence(os.durableAdmissionFence).
		WithDurableKeyedAttemptFence(os.durableAttemptFence).
		WithMaxRetryAttempts(os.maxAttempts)
	child.recordPayloadLoader = os.recordPayloadLoader
	os.recordEventOutbox = child
	return os
}

// RecordEventProcessor returns the explicitly configured dedicated child.
// Merely obtaining it starts no goroutine and changes no routing; Runtime Worker
// composition uses it to avoid the legacy environment-selected main outbox.
func (os *OutboxService) RecordEventProcessor() (*OutboxService, error) {
	if os == nil || os.recordEventOutbox == nil {
		return nil, errors.New("dedicated record-event outbox is unavailable")
	}
	return os.recordEventOutbox, nil
}

// WithDedicatedRecordEventRouting makes the already-wired record-event table
// the only durable destination for after-save record events. This is a generic
// composition mechanism, not a second queue authority: enqueue still occurs
// through this OutboxService inside the caller's transaction, while one
// separately fenced worker role owns consumption.
func (os *OutboxService) WithDedicatedRecordEventRouting() *OutboxService {
	if os != nil {
		os.dedicatedRecordEventRouting = true
	}
	return os
}

// WithRecordPayloadLoader wires the worker-side loader used by the minimal
// RecordCreated payload MVP. Foreground enqueue still writes the durable event
// row in the caller's transaction; the worker expands the public payload before
// normal subscribers see it.
func (os *OutboxService) WithRecordPayloadLoader(loader RecordPayloadLoader) *OutboxService {
	if loader != nil {
		os.recordPayloadLoader = loader
		if os.recordEventOutbox != nil {
			os.recordEventOutbox.WithRecordPayloadLoader(loader)
		}
	}
	return os
}

func (os *OutboxService) processingRepo() ports.OutboxRepository {
	if os != nil && os.workerRepo != nil {
		return os.workerRepo
	}
	if os == nil {
		return nil
	}
	return os.repo
}

func (os *OutboxService) routeRecordEventOutbox(eventType events.EventType) *OutboxService {
	if os == nil || os.recordEventOutbox == nil || (!os.dedicatedRecordEventRouting && !recordEventOutboxEnabled()) {
		return nil
	}
	switch eventType {
	case events.RecordCreated, events.RecordUpdated, events.RecordDeleted:
		return os.recordEventOutbox
	default:
		return nil
	}
}

func recordEventOutboxEnabled() bool {
	enabled, err := strconv.ParseBool(os.Getenv(recordEventOutboxEnv))
	return err == nil && enabled
}

// resolveOutboxParallelism reads the env override, clamps it, and falls
// back to defaultOutboxParallelism on parse failure. Centralised so the
// constructor stays short and the clamp logic is easy to unit-test.
func resolveOutboxParallelism() int {
	raw := os.Getenv(outboxParallelismEnv)
	if raw == "" {
		return defaultOutboxParallelism
	}
	n, err := strconv.Atoi(raw)
	if err != nil {
		log.Printf("⚠️ [Outbox] invalid %s=%q, falling back to %d: %v", outboxParallelismEnv, raw, defaultOutboxParallelism, err)
		return defaultOutboxParallelism
	}
	if n < minOutboxParallelism {
		return minOutboxParallelism
	}
	if n > maxOutboxParallelism {
		return maxOutboxParallelism
	}
	return n
}

// EnqueueEvent stores an event in the outbox table within the current transaction.
// The transaction is extracted from the context if present.
//
// Events for registered no-subscriber objects
// are dropped silently when no active record-triggered flow is wired
// against the object. The registration is only the static candidate arm; the
// metadata lookup below is the dynamic subscriber check.
func (os *OutboxService) EnqueueEvent(ctx context.Context, eventType events.EventType, payload RecordEventPayload) error {
	if routed := os.routeRecordEventOutbox(eventType); routed != nil {
		return routed.EnqueueEvent(ctx, eventType, payload)
	}
	stageStart := time.Now()
	if isNoSubscriberRecordEventObject(payload.ObjectAPIName) {
		// Dynamic arm: even though this object is on the noise candidate
		// list, the frontend TriggerConfigPanel exposes every schema as
		// a record-trigger choice, so a user could legitimately wire a
		// flow against any registered object. Drop ONLY when the metadata
		// lookup confirms no active flow is listening.
		//
		// Fail-safe: if the lookup is unwired (most unit tests) or the
		// cache cannot answer, we treat the event as needed and let it
		// through — silently dropping a wired user flow's events is a
		// far worse outcome than carrying a few noise rows through the
		// outbox.
		if os.metadata != nil && !os.metadata.HasActiveRecordTriggeredFlow(ctx, payload.ObjectAPIName) {
			// Per-event volume is bounded: the filter only matches
			// pure-noise objects, so this fires at the writer's rate
			// (typically scheduler / REST flow invokes). The log is the
			// breadcrumb operators need when debugging "why didn't my
			// record publish to WS / CDC?".
			log.Printf("⏭ [Outbox] dropped %s for object=%s (no active record-triggered flow; see registered no-subscriber event objects)", eventType, payload.ObjectAPIName)
			logOutboxServiceStageTiming(ctx, "noSubscriberCheck.drop", string(eventType), payload.ObjectAPIName, "", stageStart, nil)
			return nil
		}
	}
	logOutboxServiceStageTiming(ctx, "noSubscriberCheck", string(eventType), payload.ObjectAPIName, "", stageStart, nil)

	// Capture registered internal metadata into the serialized envelope.
	stageStart = time.Now()
	if payload.Metadata == nil {
		payload.Metadata = make(map[string]string)
	}
	internalMetadata := make(map[string]string)
	envelopeCtx, commitInternalMetadata := captureRegisteredInternalMetadata(ctx, payload.Metadata, internalMetadata)
	logOutboxServiceStageTiming(ctx, "captureInternalMetadata", string(eventType), payload.ObjectAPIName, "", stageStart, nil)

	// Repository automatically handles transaction extraction from context
	stageStart = time.Now()
	id, err := os.repo.Enqueue(envelopeCtx, string(eventType), InternalRecordEventEnvelope{
		PublicPayload:    payload,
		InternalMetadata: internalMetadata,
	})
	logOutboxServiceStageTiming(ctx, "repo.Enqueue", string(eventType), payload.ObjectAPIName, id, stageStart, err)
	if err != nil {
		return err
	}
	stageStart = time.Now()
	commitInternalMetadata()
	logOutboxServiceStageTiming(ctx, "commitInternalMetadata", string(eventType), payload.ObjectAPIName, id, stageStart, nil)
	log.Printf("✅ [Outbox] Enqueued event %s (Type: %s, ID: %s)", eventType, string(eventType), id)
	return nil
}

// StartWorker starts the configured outbox worker mode.
func (os *OutboxService) StartWorker(interval time.Duration) {
	if os.recordEventOutbox != nil && recordEventOutboxEnabled() {
		os.recordEventOutbox.StartWorker(interval)
	}
	if outboxPullWorkerEnabled() {
		os.startPullWorkers(interval)
		return
	}
	os.startBatchWorker(interval)
}

func outboxPullWorkerEnabled() bool {
	enabled, err := strconv.ParseBool(os.Getenv(outboxPullWorkerEnv))
	return err == nil && enabled
}

// startPullWorkers starts long-lived pull workers that process pending outbox
// events. Each worker polls independently with the specified idle interval.
func (os *OutboxService) startPullWorkers(interval time.Duration) {
	ready, err := os.outboxRecordKeyColumnReady(context.Background())
	if err != nil {
		log.Printf("⚠️ Outbox pull workers not started; RecordKey column probe failed, falling back to batch worker: %v", err)
		os.startBatchWorker(interval)
		return
	}
	if !ready {
		log.Printf("⚠️ Outbox pull workers not started; RecordKey column is not applied, falling back to batch worker")
		os.startBatchWorker(interval)
		return
	}

	log.Printf("📤 Outbox pull workers starting with %v idle interval (parallelism=%d)", interval, os.parallelism)
	os.startOutboxRecordKeyBackfillWorker(interval)
	for i := 0; i < os.parallelism; i++ {
		workerID := i
		os.wg.Add(1)
		go os.runOutboxPullWorker(workerID, interval)
	}
}

func (os *OutboxService) startBatchWorker(interval time.Duration) {
	os.wg.Add(1)
	go func() {
		defer os.wg.Done()
		defer func() {
			if r := recover(); r != nil {
				log.Printf("🔥 CRITICAL: Outbox batch worker panicked: %v", r)
			}
		}()

		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		log.Printf("📤 Outbox batch worker started with %v interval (parallelism=%d)", interval, os.parallelism)

		for {
			select {
			case <-os.stopCh:
				log.Printf("📤 Outbox batch worker stopping...")
				return
			case <-ticker.C:
				func() {
					defer func() {
						if r := recover(); r != nil {
							log.Printf("🔥 Outbox batch iteration panic: %v", r)
						}
					}()
					workerCtx, cancel := outboxWorkerProcessContext()
					err := os.processOutboxBatch(workerCtx)
					cancel()
					if err != nil {
						log.Printf("⚠️ Outbox batch worker error: %v", err)
					}
				}()
			}
		}
	}()
}

type outboxRecordKeyBackfiller interface {
	BackfillRecordKeys(ctx context.Context, limit int) (int, error)
}

type outboxRecordKeyColumnProbe interface {
	HasRecordKeyColumn(ctx context.Context) (bool, error)
}

func (os *OutboxService) outboxRecordKeyColumnReady(ctx context.Context) (bool, error) {
	probe, ok := os.processingRepo().(outboxRecordKeyColumnProbe)
	if !ok {
		return true, nil
	}
	return probe.HasRecordKeyColumn(ctx)
}

func (os *OutboxService) backfillOutboxRecordKeyBatch(ctx context.Context) (int, error) {
	backfiller, ok := os.processingRepo().(outboxRecordKeyBackfiller)
	if !ok {
		return 0, nil
	}
	return backfiller.BackfillRecordKeys(ctx, outboxRecordKeyBackfillBatchSize)
}

func (os *OutboxService) startOutboxRecordKeyBackfillWorker(interval time.Duration) {
	if _, ok := os.processingRepo().(outboxRecordKeyBackfiller); !ok {
		return
	}

	os.wg.Add(1)
	go os.runOutboxRecordKeyBackfillWorker(interval)
}

func (os *OutboxService) runOutboxRecordKeyBackfillWorker(interval time.Duration) {
	defer os.wg.Done()
	defer func() {
		if r := recover(); r != nil {
			log.Printf("🔥 CRITICAL: Outbox RecordKey backfill worker panicked: %v", r)
		}
	}()

	timer := time.NewTimer(0)
	defer timer.Stop()

	for {
		select {
		case <-os.stopCh:
			log.Printf("📤 Outbox RecordKey backfill worker stopping...")
			return
		case <-timer.C:
		}

		ctx, cancel := outboxBookkeepingContext()
		n, err := os.backfillOutboxRecordKeyBatch(ctx)
		cancel()
		if err != nil {
			log.Printf("⚠️ Outbox RecordKey backfill failed; pull workers continue and will retry: %v", err)
			timer.Reset(interval)
			continue
		}
		if n == 0 {
			return
		}
		timer.Reset(interval)
	}
}

func (os *OutboxService) runOutboxPullWorker(workerID int, interval time.Duration) {
	defer os.wg.Done()
	defer func() {
		if r := recover(); r != nil {
			log.Printf("🔥 CRITICAL: Outbox pull worker %d panicked: %v", workerID, r)
		}
	}()

	timer := time.NewTimer(0)
	defer timer.Stop()

	for {
		select {
		case <-os.stopCh:
			log.Printf("📤 Outbox pull worker %d stopping...", workerID)
			return
		case <-timer.C:
		}

		os.drainOutboxWorker(workerID)

		timer.Reset(interval)
	}
}

func (os *OutboxService) drainOutboxWorker(workerID int) {
	for {
		select {
		case <-os.stopCh:
			return
		default:
		}

		workerCtx, cancel := outboxWorkerProcessContext()
		processed, err := os.processNextOutboxEvent(workerCtx, workerID)
		cancel()
		if err != nil {
			log.Printf("⚠️ Outbox pull worker %d error: %v", workerID, err)
			return
		}
		if !processed {
			return
		}
	}
}

func outboxWorkerProcessContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(context.Background())
}

func outboxBookkeepingContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(context.Background())
}

// StopWorker stops the background workers gracefully.
func (os *OutboxService) StopWorker() {
	if os.recordEventOutbox != nil {
		os.recordEventOutbox.StopWorker()
	}
	os.stopOnce.Do(func() {
		close(os.stopCh)
	})
	os.wg.Wait()
	log.Printf("📤 Outbox worker stopped")
}

// ProcessOutbox drains eligible events from the outbox table until no row is
// currently claimable or an error stops the deterministic drain.
//
// Production uses StartWorker, which runs multiple independent pull workers.
// Tests and deterministic callers can invoke this method to synchronously drain
// without constructing the background worker fleet.
func (os *OutboxService) ProcessOutbox(ctx context.Context) error {
	if os.recordEventOutbox != nil && recordEventOutboxEnabled() {
		if err := os.recordEventOutbox.ProcessOutbox(ctx); err != nil {
			return err
		}
	}
	useRecordFIFO, err := os.outboxRecordKeyColumnReady(ctx)
	if err != nil {
		return err
	}
	for {
		processed, err := os.processNextOutboxEventWithFIFO(ctx, -1, useRecordFIFO)
		if err != nil {
			return err
		}
		if !processed {
			return nil
		}
	}
}

// ProcessNext processes at most one row from this service's own repository.
// It never traverses an environment-selected child and is therefore suitable
// for one explicitly fenced Runtime work-family controller.
func (os *OutboxService) ProcessNext(ctx context.Context) (bool, error) {
	useRecordFIFO, err := os.outboxRecordKeyColumnReady(ctx)
	if err != nil {
		return false, err
	}
	return os.processNextOutboxEventWithFIFO(ctx, -1, useRecordFIFO)
}

// ClaimDurableRecordEventBatch claims no more than limit distinct FIFO heads
// under one deployment-admission transaction. A row claimed earlier in this
// transaction remains non-terminal, so the existing FIFO anti-chain predicate
// excludes every younger row for that RecordKey while still admitting heads
// for independent keys.
func (os *OutboxService) ClaimDurableRecordEventBatch(ctx context.Context, limit int) ([]DurableClaimedRecordEvent, error) {
	if os == nil || os.durableAdmissionFence == nil {
		return nil, errors.New("durable record-event batch claim requires an admission fence")
	}
	claimer, ok := os.processingRepo().(outboxFIFOAtomicBatchClaimer)
	if !ok {
		return nil, errors.New("durable record-event batch claim requires atomic FIFO admission")
	}
	claimed, err := os.claimDurableRecordEventBatch(ctx, limit, claimer)
	if err != nil {
		return nil, apperrors.NewInternalError("failed to atomically claim FIFO record-event batch", err)
	}
	return claimed, nil
}

func (os *OutboxService) claimDurableRecordEventBatch(ctx context.Context, limit int, claimer outboxFIFOAtomicBatchClaimer) ([]DurableClaimedRecordEvent, error) {
	if limit <= 0 || limit > maxOutboxParallelism {
		return nil, fmt.Errorf("durable record-event batch claim limit must be between 1 and %d", maxOutboxParallelism)
	}
	claimed := make([]DurableClaimedRecordEvent, 0, limit)
	err := os.withDurableAdmissionWrite(ctx, func(context.Context) error { return nil }, func(txCtx context.Context) error {
		events, claims, err := claimer.ClaimEventBatchWithRecordFIFO(txCtx, limit)
		if err != nil {
			return err
		}
		if len(events) != len(claims) || len(events) > limit {
			return errors.New("atomic FIFO batch claim returned inconsistent capacity")
		}
		for index, event := range events {
			claim := claims[index]
			if event.ID == "" || claim.EventID == "" || event.ID != claim.EventID {
				return errors.New("atomic FIFO claim returned inconsistent event and lease identity")
			}
			if claim.RecordKey == "" {
				claim.RecordKey = recordKeyFromPayload(event.Payload)
				if claim.RecordKey == "" {
					claim.RecordKey = event.ID
				}
			}
			claimed = append(claimed, DurableClaimedRecordEvent{event: event, claim: claim})
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return claimed, nil
}

// ProcessDurableRecordEventClaim executes one already committed processing
// lease. Claim admission and effects remain separate transactions: the effect
// path revalidates the claim token and keyed ownership fence before publishing
// any subscriber side effect.
func (os *OutboxService) ProcessDurableRecordEventClaim(ctx context.Context, claimed DurableClaimedRecordEvent) error {
	if os == nil || claimed.event.ID == "" || claimed.claim.EventID != claimed.event.ID || claimed.claim.RecordKey == "" {
		return errors.New("durable record-event processing lease is incomplete")
	}
	return os.processDurableRecordEventClaim(ctx, claimed, -1)
}

func (os *OutboxService) processOutboxBatch(ctx context.Context) error {
	batchID := utils.GenerateID()
	if len(batchID) >= 8 {
		batchID = batchID[:8]
	}

	pending, err := os.processingRepo().GetPendingEvents(ctx, 100)
	if err != nil {
		return err
	}
	if len(pending) == 0 {
		return nil
	}

	log.Printf("🔄 [Outbox][%s] Processing %d pending events (parallelism=%d)", batchID, len(pending), os.parallelism)

	slotChans := make([]chan OutboxEventEnvelope, os.parallelism)
	for i := range slotChans {
		slotChans[i] = make(chan OutboxEventEnvelope, 16)
	}

	var wg sync.WaitGroup
	wg.Add(os.parallelism)
	for i, ch := range slotChans {
		go func(slot int, in <-chan OutboxEventEnvelope) {
			defer wg.Done()
			for env := range in {
				func(env OutboxEventEnvelope) {
					defer func() {
						if r := recover(); r != nil {
							log.Printf("🔥 [Outbox][%s] slot %d event %s panicked: %v", batchID, slot, env.ID, r)
						}
					}()
					if err := os.processEventAtomic(ctx, env.ID, env.EventType, env.Payload); err != nil {
						log.Printf("⚠️ [Outbox][%s] slot %d failed event %s: %v", batchID, slot, env.ID, err)
					}
				}(env)
			}
		}(i, ch)
	}

	dispatchErr := func() error {
		for i := range pending {
			ev := pending[i]
			slot := slotForEvent(ev.ID, ev.Payload, os.parallelism)
			env := OutboxEventEnvelope{
				ID:        ev.ID,
				EventType: ev.EventType,
				Payload:   ev.Payload,
			}
			select {
			case slotChans[slot] <- env:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	}()

	for _, ch := range slotChans {
		close(ch)
	}
	wg.Wait()

	if dispatchErr != nil {
		log.Printf("⚠️ [Outbox][%s] dispatch interrupted: %v", batchID, dispatchErr)
		return dispatchErr
	}
	return nil
}

// OutboxEventEnvelope carries the subset of fields processEventAtomic needs in
// the legacy batch drainer.
type OutboxEventEnvelope struct {
	ID        string
	EventType string
	Payload   string
}

func slotForEvent(id, payloadJSON string, parallelism int) int {
	if parallelism <= 1 {
		return 0
	}
	key := recordKeyFromPayload(payloadJSON)
	if key == "" {
		key = id
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(key))
	return int(h.Sum32() % uint32(parallelism))
}

func recordKeyFromPayload(payloadJSON string) string {
	if payloadJSON == "" {
		return ""
	}
	var envelope InternalRecordEventEnvelope
	if err := json.Unmarshal([]byte(payloadJSON), &envelope); err == nil {
		if envelope.PublicPayload.ObjectAPIName != "" && envelope.PublicPayload.Record.GetString("Id") != "" {
			return envelope.PublicPayload.ObjectAPIName + "/" + envelope.PublicPayload.Record.GetString("Id")
		}
	}
	var legacy RecordEventPayload
	if err := json.Unmarshal([]byte(payloadJSON), &legacy); err != nil {
		return ""
	}
	recordID := legacy.Record.GetString("Id")
	if legacy.ObjectAPIName == "" || recordID == "" {
		return ""
	}
	return legacy.ObjectAPIName + "/" + recordID
}

func (os *OutboxService) processNextOutboxEvent(ctx context.Context, workerID int) (processed bool, err error) {
	return os.processNextOutboxEventWithFIFO(ctx, workerID, true)
}

func (os *OutboxService) processNextOutboxEventWithFIFO(ctx context.Context, workerID int, useRecordFIFO bool) (processed bool, err error) {
	repo := os.processingRepo()
	if useRecordFIFO && os.durableAdmissionFence != nil {
		if claimer, ok := repo.(outboxFIFOAtomicBatchClaimer); ok {
			return os.processNextAtomicallyClaimedFIFOEvent(ctx, workerID, claimer)
		}
	}

	var pending []models.OutboxEvent
	if useRecordFIFO {
		pending, err = repo.GetPendingEventsWithRecordFIFO(ctx, 1)
	} else {
		pending, err = repo.GetPendingEvents(ctx, 1)
	}
	if err != nil {
		return false, err
	}
	if len(pending) == 0 {
		return false, nil
	}

	ev := pending[0]
	label := "manual"
	if workerID >= 0 {
		label = fmt.Sprintf("worker-%d", workerID)
	}
	log.Printf("🔄 [Outbox][%s] Processing event %s (Type: %s)", label, ev.ID, ev.EventType)

	defer func() {
		if r := recover(); r != nil {
			processed = true
			err = fmt.Errorf("panic while processing outbox event %s: %v", ev.ID, r)
			log.Printf("🔥 [Outbox][%s] event %s panicked: %v", label, ev.ID, r)
		}
	}()
	if useRecordFIFO {
		recordKey := recordKeyFromPayload(ev.Payload)
		if recordKey == "" {
			recordKey = ev.ID
		}
		claim, err := os.claimEvent(ctx, ev.ID, recordKey)
		if err != nil {
			if errors.Is(err, persistence.ErrOutboxEventAlreadyClaimed) {
				return true, nil
			}
			return true, apperrors.NewInternalError("failed to claim event", err)
		}
		if claim.EventID == "" {
			return true, nil
		}
		if err := os.processClaimedEventAtomic(ctx, claim, ev.EventType, ev.Payload); err != nil {
			log.Printf("⚠️ [Outbox][%s] failed event %s: %v", label, ev.ID, err)
			return true, err
		}
		return true, nil
	}
	if err := os.processEventAtomic(ctx, ev.ID, ev.EventType, ev.Payload); err != nil {
		log.Printf("⚠️ [Outbox][%s] failed event %s: %v", label, ev.ID, err)
		return true, err
	}
	return true, nil
}

func (os *OutboxService) processNextAtomicallyClaimedFIFOEvent(ctx context.Context, workerID int, claimer outboxFIFOAtomicBatchClaimer) (processed bool, err error) {
	claimed, err := os.claimDurableRecordEventBatch(ctx, 1, claimer)
	if err != nil {
		return false, apperrors.NewInternalError("failed to atomically claim next FIFO event", err)
	}
	if len(claimed) == 0 {
		return false, nil
	}
	return true, os.processDurableRecordEventClaim(ctx, claimed[0], workerID)
}

func (os *OutboxService) processDurableRecordEventClaim(ctx context.Context, claimed DurableClaimedRecordEvent, workerID int) (err error) {
	event := claimed.event
	claim := claimed.claim
	label := "manual"
	if workerID >= 0 {
		label = fmt.Sprintf("worker-%d", workerID)
	}
	log.Printf("🔄 [Outbox][%s] Processing event %s (Type: %s)", label, event.ID, event.EventType)

	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("panic while processing outbox event %s: %v", event.ID, recovered)
			log.Printf("🔥 [Outbox][%s] event %s panicked: %v", label, event.ID, recovered)
		}
	}()
	if err := os.processClaimedEventAtomic(ctx, claim, event.EventType, event.Payload); err != nil {
		log.Printf("⚠️ [Outbox][%s] failed event %s: %v", label, event.ID, err)
		return err
	}
	return nil
}

// claimEvent makes the visible processing lease part of the same short
// transaction as the deployment-ownership fence. A preflight-only fence is
// insufficient: ownership may flip between VerifyActive and ClaimEvent,
// allowing a stale worker to hide work from the new owner for the full stale
// claim interval even though the later effect transaction correctly refuses
// to publish. Legacy in-process outbox consumers have no deployment fence and
// preserve their existing single-statement claim path.
func (os *OutboxService) claimEvent(ctx context.Context, id string, recordKeys ...string) (models.OutboxClaim, error) {
	repo := os.processingRepo()
	recordKey := id
	if len(recordKeys) > 0 && recordKeys[0] != "" {
		recordKey = recordKeys[0]
	}
	var claim models.OutboxClaim
	err := os.withDurableAdmissionWrite(ctx, func(txCtx context.Context) error {
		if locker, ok := repo.(outboxClaimCandidateLocker); ok {
			return locker.LockOutboxClaimCandidate(txCtx, id)
		}
		return nil
	}, func(txCtx context.Context) error {
		var err error
		claim, err = repo.ClaimEvent(txCtx, id)
		claim.RecordKey = recordKey
		return err
	})
	return claim, err
}

func (os *OutboxService) withDurableAdmissionWrite(ctx context.Context, beforeFence, write func(context.Context) error) error {
	if beforeFence == nil || write == nil {
		return errors.New("durable outbox admission write is required")
	}
	if os.durableAdmissionFence == nil {
		return write(ctx)
	}
	if os.txManager == nil {
		return errors.New("durable outbox admission requires a transaction manager")
	}
	return os.txManager.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		if err := os.durableAdmissionFence(txCtx); err != nil {
			return err
		}
		// The singleton gate must precede the candidate row in the lock order:
		// transfer also owns singleton -> slots, while effects own slot -> row.
		// NOWAIT makes this row acquisition non-blocking, so it cannot recreate
		// admission HOL while the singleton is held.
		if err := beforeFence(txCtx); err != nil {
			return err
		}
		return write(txCtx)
	})
}

// withDurableQueueWrite makes the deployment fence and every durable queue
// mutation one transaction. The effect/ack path already holds this fence; the
// same rule must also cover claim leases and retry/poison reconciliation or a
// stale deployment could still hide or rewrite work after ownership moved.
// Legacy in-process consumers have no deployment fence and preserve their
// existing direct repository path.
func (os *OutboxService) withDurableQueueWrite(ctx context.Context, recordKey string, write func(context.Context) error) error {
	if write == nil {
		return errors.New("durable outbox write is required")
	}
	if os.durableAttemptFence == nil {
		return write(ctx)
	}
	if os.txManager == nil {
		return errors.New("durable outbox write requires a transaction manager")
	}
	return os.txManager.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		if err := os.durableAttemptFence(txCtx, recordKey); err != nil {
			return err
		}
		return write(txCtx)
	})
}

func (os *OutboxService) refreshProcessingClaim(ctx context.Context, claim models.OutboxClaim) error {
	return os.withDurableQueueWrite(ctx, claim.RecordKey, func(txCtx context.Context) error {
		return os.processingRepo().RefreshOutboxProcessingClaim(txCtx, claim)
	})
}

func decodeInternalRecordEventEnvelope(payloadJSON string) (InternalRecordEventEnvelope, error) {
	var envelope InternalRecordEventEnvelope
	if err := json.Unmarshal([]byte(payloadJSON), &envelope); err == nil && envelope.PublicPayload.ObjectAPIName != "" {
		if envelope.InternalMetadata == nil {
			envelope.InternalMetadata = map[string]string{}
		}
		return envelope, nil
	}

	var legacy RecordEventPayload
	if err := json.Unmarshal([]byte(payloadJSON), &legacy); err != nil {
		return InternalRecordEventEnvelope{}, err
	}

	return InternalRecordEventEnvelope{
		PublicPayload:    legacy,
		InternalMetadata: cloneStringMap(legacy.Metadata),
	}, nil
}

func cloneStringMap(in map[string]string) map[string]string {
	if len(in) == 0 {
		return map[string]string{}
	}
	out := make(map[string]string, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

var errInvalidOutboxPayload = errors.New("invalid outbox payload")

type preparedClaimedEvent struct {
	envelope         InternalRecordEventEnvelope
	installExecution DurableExecutionInstaller
}

// prepareClaimedEvent opens all immutable execution inputs before the
// database effect/ack transaction begins. The returned value is request-local
// and contains no selector or fallback: it can only reinstall the complete
// execution envelope verified from the durable row.
func (os *OutboxService) prepareClaimedEvent(ctx context.Context, id, payloadJSON string) (*preparedClaimedEvent, error) {
	envelope, err := decodeInternalRecordEventEnvelope(payloadJSON)
	if err != nil {
		log.Printf("❌ [Outbox] Event %s failed payload unmarshal: %v", id, err)
		return nil, fmt.Errorf("%w: %v", errInvalidOutboxPayload, err)
	}
	prepared := &preparedClaimedEvent{envelope: envelope}
	if os.durableExecutionBinder == nil {
		return prepared, nil
	}
	binding := envelope.InternalMetadata[durableExecutionBindingMetadataKey]
	if binding == "" {
		return nil, errors.New("outbox event has no durable metadata binding")
	}
	openCtx := restoreInternalMetadata(ctx, envelope.InternalMetadata)
	prepared.installExecution, err = os.durableExecutionBinder(openCtx, binding)
	if err != nil {
		return nil, fmt.Errorf("bind durable outbox metadata: %w", err)
	}
	if prepared.installExecution == nil {
		return nil, errors.New("bind durable outbox metadata: binder returned no execution installer")
	}
	return prepared, nil
}

func (os *OutboxService) publishClaimedEvent(ctx context.Context, id, eventType, payloadJSON string) error {
	prepared, err := os.prepareClaimedEvent(ctx, id, payloadJSON)
	if err != nil {
		return err
	}
	pubCtx, publishPayload, err := os.prepareClaimedEventPublish(ctx, id, eventType, prepared)
	if err != nil {
		return err
	}

	// Publish via EventBus. Do not wrap this call with a local timeout:
	// after-save subscribers do not uniformly propagate handler timeout
	// failures back through Publish, so outbox retry/accounting must not
	// infer durable success/failure from a deadline at this layer.
	return os.publishOutboxEvent(pubCtx, eventType, publishPayload)
}

func (os *OutboxService) prepareClaimedEventPublish(ctx context.Context, id, eventType string, prepared *preparedClaimedEvent) (context.Context, RecordEventPayload, error) {
	if prepared == nil {
		return nil, RecordEventPayload{}, errors.New("outbox event was not prepared")
	}
	pubCtx := restoreInternalMetadata(ctx, prepared.envelope.InternalMetadata)
	if prepared.installExecution != nil {
		var installErr error
		pubCtx, installErr = prepared.installExecution(pubCtx)
		if installErr != nil {
			return nil, RecordEventPayload{}, fmt.Errorf("install durable outbox execution envelope: %w", installErr)
		}
	}
	pubCtx = WithCurrentOutboxEventID(pubCtx, id)
	publishPayload := prepared.envelope.PublicPayload
	if expanded, expandErr := os.expandMinimalRecordPayload(pubCtx, eventType, publishPayload, prepared.envelope.InternalMetadata); expandErr != nil {
		return nil, RecordEventPayload{}, expandErr
	} else {
		publishPayload = expanded
	}
	// Strip internal recursion-control and active-projection keys from the
	// public payload before dispatch. The active-projection snapshots are
	// still accessible from pubCtx via ActiveProjectionRecordFromCtx /
	// ActiveProjectionOldRecordFromCtx (restored by the registered internal
	// metadata hook); they must not be visible to general event subscribers
	// through payload.Metadata.
	publishPayload.Metadata = stripRegisteredInternalMetadata(publishPayload.Metadata)

	return pubCtx, publishPayload, nil
}

func (os *OutboxService) runClaimedEventAttempt(txCtx context.Context, claim models.OutboxClaim, eventType string, prepared *preparedClaimedEvent, prepareErr error, claimNeedsRevalidation, useSavepoint bool) error {
	repo := os.processingRepo()
	id := claim.EventID
	if os.durableAttemptFence != nil {
		if err := os.durableAttemptFence(txCtx, claim.RecordKey); err != nil {
			return err
		}
	}
	// Pull workers commit the claim before opening the effect transaction, so
	// they must revalidate the visible lease before invoking any subscriber.
	// The legacy batch path creates its claim inside this same transaction and
	// already holds the row lock. Refreshing it immediately would write the same
	// database timestamp and TiDB reports RowsAffected()==0 for that no-op,
	// falsely classifying the fresh claim as lost.
	if claimNeedsRevalidation {
		if err := repo.RefreshOutboxProcessingClaim(txCtx, claim); err != nil {
			return err
		}
	}
	if prepareErr != nil {
		if errors.Is(prepareErr, errInvalidOutboxPayload) {
			if markErr := repo.MarkFailed(txCtx, claim, fmt.Sprintf("invalid payload: %v", prepareErr)); markErr != nil {
				return apperrors.NewInternalError("failed to mark event as failed", markErr)
			}
			return nil
		}
		return os.handleClaimedEventFailure(txCtx, claim, prepareErr)
	}
	pubCtx, publishPayload, err := os.prepareClaimedEventPublish(txCtx, id, eventType, prepared)
	if err != nil {
		if errors.Is(err, errInvalidOutboxPayload) {
			if markErr := repo.MarkFailed(txCtx, claim, fmt.Sprintf("invalid payload: %v", err)); markErr != nil {
				return apperrors.NewInternalError("failed to mark event as failed", markErr)
			}
			return nil
		}
		return os.handleClaimedEventFailure(txCtx, claim, err)
	}

	if useSavepoint {
		if err := repo.CreateOutboxAttemptSavepoint(txCtx); err != nil {
			return apperrors.NewInternalError("failed to create outbox attempt savepoint", err)
		}
	}

	if err := os.publishOutboxEvent(pubCtx, eventType, publishPayload); err != nil {
		// Late field-history lock discovery is transaction-control flow. The
		// root retry owner must roll back the whole subscriber effect and replay
		// it with the complete ordered prelock set; converting it into an outbox
		// retry would repeat the same incomplete plan on every attempt.
		if _, retryRequired := ports.HistorySequencePrelockFields(err); retryRequired {
			return err
		}
		if useSavepoint {
			if rollbackErr := repo.RollbackToOutboxAttemptSavepoint(txCtx); rollbackErr != nil {
				return apperrors.NewInternalError("failed to roll back outbox attempt savepoint", rollbackErr)
			}
			if releaseErr := repo.ReleaseOutboxAttemptSavepoint(txCtx); releaseErr != nil {
				return apperrors.NewInternalError("failed to release outbox attempt savepoint", releaseErr)
			}
		}
		return os.handleClaimedEventFailure(txCtx, claim, err)
	}

	if useSavepoint {
		if releaseErr := repo.ReleaseOutboxAttemptSavepoint(txCtx); releaseErr != nil {
			return apperrors.NewInternalError("failed to release outbox attempt savepoint", releaseErr)
		}
	}

	if err := repo.UpdateStatus(txCtx, claim, OutboxStatusProcessed, ""); err != nil {
		return apperrors.NewInternalError("failed to mark as processed", err)
	}

	log.Printf("✅ [Outbox] Successfully processed event %s (Type: %s)", id, eventType)
	return nil
}

// processEventAtomic claims an event, publishes it, and updates status inside
// one pessimistic transaction. The outbox row lock stays held for the live
// attempt on the legacy batch path. Pull workers may claim first so the visible
// processing lease can heartbeat, but publish + final status still share one
// transaction so handler DML is rolled back if outbox success bookkeeping fails.
func (os *OutboxService) processEventAtomic(ctx context.Context, id, eventType, payloadJSON string) error {
	repo := os.processingRepo()
	prepared, prepareErr := os.prepareClaimedEvent(ctx, id, payloadJSON)
	run := func(txCtx context.Context, useSavepoint bool) error {
		// Claim this specific event. ClaimEvent returns
		// ErrOutboxEventAlreadyClaimed when the row is no longer pending; that is
		// a benign race outcome.
		claim, err := repo.ClaimEvent(txCtx, id)
		if err != nil {
			if errors.Is(err, persistence.ErrOutboxEventAlreadyClaimed) {
				return nil
			}
			return apperrors.NewInternalError("failed to claim event", err)
		}
		if claim.EventID == "" {
			return nil // Legacy "already claimed" signal; retained for safety.
		}

		return os.runClaimedEventAttempt(txCtx, claim, eventType, prepared, prepareErr, false, useSavepoint)
	}
	if os.txManager == nil {
		return run(ctx, false)
	}
	return os.txManager.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		return run(txCtx, true)
	}, outboxEffectTransactionRetries)
}

// processClaimedEventAtomic handles an event already claimed in a short,
// committed claim statement. Pull workers use this path so Status='processing'
// is visible to other workers while the handler runs; the heartbeat keeps that
// visible claim fresh until final status or retry bookkeeping lands.
func (os *OutboxService) processClaimedEventAtomic(ctx context.Context, claim models.OutboxClaim, eventType, payloadJSON string) error {
	id := claim.EventID
	stopHeartbeat := os.startOutboxProcessingHeartbeat(ctx, claim)
	defer stopHeartbeat()
	prepared, prepareErr := os.prepareClaimedEvent(ctx, id, payloadJSON)

	run := func(txCtx context.Context, useSavepoint bool) error {
		return os.runClaimedEventAttempt(txCtx, claim, eventType, prepared, prepareErr, true, useSavepoint)
	}

	var err error
	if os.txManager == nil {
		err = run(ctx, false)
	} else {
		err = os.txManager.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
			return run(txCtx, true)
		}, outboxEffectTransactionRetries)
	}
	if err == nil {
		return nil
	}
	if errors.Is(err, persistence.ErrOutboxEventAlreadyClaimed) {
		return nil
	}
	if errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		// A stale deployment may have claimed a row before the ownership flip,
		// but it must neither publish effects nor mutate retry/poison state.
		return err
	}

	bookkeepingCtx, cancel := outboxBookkeepingContext()
	defer cancel()
	if retryErr := os.withDurableQueueWrite(bookkeepingCtx, claim.RecordKey, func(txCtx context.Context) error {
		// The failed effect transaction rolled back its claim refresh and row lock.
		// Reacquire both before changing retry/poison state: a token that has not
		// yet been rotated is not authority once its database lease has expired.
		if refreshErr := os.processingRepo().RefreshOutboxProcessingClaim(txCtx, claim); refreshErr != nil {
			return refreshErr
		}
		return os.handleClaimedEventFailure(txCtx, claim, err)
	}); retryErr != nil {
		if errors.Is(retryErr, persistence.ErrOutboxEventAlreadyClaimed) {
			return nil
		}
		return retryErr
	}
	return nil
}

func (os *OutboxService) startOutboxProcessingHeartbeat(parent context.Context, claim models.OutboxClaim) func() {
	id := claim.EventID
	interval := outboxProcessingHeartbeatInterval
	if policy, ok := os.processingRepo().(processingClaimHeartbeatPolicy); ok {
		interval = policy.ProcessingClaimHeartbeatInterval()
	}
	if interval <= 0 {
		return func() {}
	}

	done := make(chan struct{})
	var once sync.Once
	var wg sync.WaitGroup
	heartbeatCtx, cancelHeartbeat := context.WithCancel(parent)
	wg.Add(1)
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-done:
				return
			case <-heartbeatCtx.Done():
				return
			case <-ticker.C:
			}

			err := os.refreshProcessingClaim(heartbeatCtx, claim)
			if err == nil {
				continue
			}
			if errors.Is(err, persistence.ErrOutboxEventAlreadyClaimed) || errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
				return
			}
			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
				return
			}
			log.Printf("⚠️ [Outbox] failed to refresh processing claim for event %s: %v", id, err)
		}
	}()

	return func() {
		once.Do(func() {
			cancelHeartbeat()
			close(done)
			wg.Wait()
		})
	}
}

func (os *OutboxService) publishOutboxEvent(ctx context.Context, eventType string, payload interface{}) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic while processing event: %v", r)
		}
	}()
	return os.eventBus.Publish(ctx, events.EventType(eventType), payload)
}

func (os *OutboxService) handleClaimedEventFailure(ctx context.Context, claim models.OutboxClaim, cause error) error {
	id := claim.EventID
	newRetryCount := claim.RetryCount + 1
	repo := os.processingRepo()
	maxAttempts := os.maxAttempts
	if maxAttempts <= 0 {
		maxAttempts = MaxRetryAttempts
	}
	if newRetryCount >= maxAttempts {
		if markErr := repo.MarkFailed(ctx, claim, fmt.Sprintf("max retries exceeded: %v", cause)); markErr != nil {
			return apperrors.NewInternalError("failed to mark event as failed", markErr)
		}
		return nil
	}

	if updateErr := repo.IncrementRetry(ctx, claim, cause.Error()); updateErr != nil {
		return apperrors.NewInternalError("failed to update retry count", updateErr)
	}
	log.Printf("⚠️ [Outbox] Event %s failed (Attempt %d/%d). Error: %v", id, newRetryCount, maxAttempts, cause)
	return nil
}

// CleanupProcessed removes old processed events from the outbox.
// This should be called periodically (e.g., daily) to prevent table bloat.
func (os *OutboxService) CleanupProcessed(ctx context.Context, olderThan time.Duration) (int64, error) {
	cutoff := time.Now().Add(-olderThan)
	deleted, err := os.processingRepo().CleanupProcessed(ctx, cutoff)
	if err != nil {
		return deleted, err
	}
	if os.recordEventOutbox != nil && recordEventOutboxEnabled() {
		childDeleted, childErr := os.recordEventOutbox.processingRepo().CleanupProcessed(ctx, cutoff)
		deleted += childDeleted
		if childErr != nil {
			return deleted, childErr
		}
	}
	return deleted, nil
}

// CleanupProcessedBatch owns bounded retention for one independently deployed
// durable queue. The cleanup and deployment fence commit in the same
// transaction, so a stale worker cannot delete completed rows after ownership
// moves. It intentionally does not recurse into a child queue: one worker role
// owns exactly one physical work family.
func (os *OutboxService) CleanupProcessedBatch(ctx context.Context, retention time.Duration, batchSize int) (int64, error) {
	if os == nil || retention < time.Second || batchSize <= 0 {
		return 0, errors.New("durable outbox cleanup requires positive retention and batch size")
	}
	cleaner, ok := os.processingRepo().(processedOutboxBatchCleaner)
	if !ok {
		return 0, errors.New("durable outbox cleanup is unavailable")
	}
	var deleted int64
	err := os.withDurableQueueWrite(ctx, "runtime-record-event-maintenance", func(txCtx context.Context) error {
		var err error
		deleted, err = cleaner.CleanupProcessedBatch(txCtx, retention, batchSize)
		return err
	})
	return deleted, err
}