Function runHandler

apps/sim/lib/core/outbox/service.ts:319–394

* Phase 2: invoke the handler for a claimed event, outside any DB * transaction, then transition the row to its terminal or retry state. * * Every terminal UPDATE is guarded by a lease CAS (`WHERE status = * 'processing' AND locked_at = event.lockedAt`). This defends against * the "slow handler

(
  event: typeof outboxEvent.$inferSelect,
  handlers: OutboxHandlerRegistry
)

 * rowCount is 0 — and we log+skip instead of clobbering the new lease.
 */
async function runHandler(
  event: typeof outboxEvent.$inferSelect,
  handlers: OutboxHandlerRegistry
): Promise<'completed' | 'pending' | 'dead_letter' | 'lease_lost'> {
  const handler = handlers[event.eventType]

  if (!handler) {
    logger.error('No handler registered for outbox event type', {
      eventId: event.id,
      eventType: event.eventType,
    })
    await updateIfLeaseHeld(event, {
      status: 'dead_letter',
      lastError: `No handler registered for event type '${event.eventType}'`,
      processedAt: new Date(),
      lockedAt: null,
    })
    return 'dead_letter'
  }

  try {
    await runHandlerWithTimeout(handler, event)
    const updated = await updateIfLeaseHeld(event, {
      status: 'completed',
      processedAt: new Date(),
      lockedAt: null,
    })
    if (!updated) {
      logger.warn('Outbox event completion skipped — lease lost (reaped + reclaimed)', {
        eventId: event.id,
        eventType: event.eventType,
      })
      return 'lease_lost'
    }
    logger.info('Outbox event processed', {
      eventId: event.id,
      eventType: event.eventType,
      attempts: event.attempts + 1,
    })
    return 'completed'
  } catch (error) {
    if (error instanceof OutboxHandlerTimeoutError) {
      return recordTimedOutAttempt(event, error.message)
    }

    const nextAttempts = event.attempts + 1
    const isDead = nextAttempts >= event.maxAttempts
    const errMsg = toError(error).message

    if (isDead) {
      const updated = await updateIfLeaseHeld(event, {
        attempts: nextAttempts,
        status: 'dead_letter',
        lastError: errMsg,
        processedAt: new Date(),
        lockedAt: null,
      })
      if (!updated) {