Function processOutboxEventById

apps/sim/lib/core/outbox/service.ts:201–240

(
  eventId: string,
  handlers: OutboxHandlerRegistry
)

 * `FOR UPDATE SKIP LOCKED`, and non-pending rows are left alone.
 */
export async function processOutboxEventById(
  eventId: string,
  handlers: OutboxHandlerRegistry
): Promise<ProcessSingleOutboxResult> {
  const now = new Date()
  const event = await db.transaction(async (tx) => {
    const [row] = await tx
      .select()
      .from(outboxEvent)
      .where(eq(outboxEvent.id, eventId))
      .limit(1)
      .for('update', { skipLocked: true })

    if (!row) return null
    if (row.status !== 'pending') return row.status as ProcessSingleOutboxResult
    if (row.availableAt > now) return 'pending' as const

    await tx
      .update(outboxEvent)
      .set({ status: 'processing', lockedAt: now })
      .where(eq(outboxEvent.id, eventId))

    return {
      ...row,
      status: 'processing' as const,
      lockedAt: now,
    }
  })

  if (!event) {
    const [current] = await db
      .select({ status: outboxEvent.status })
      .from(outboxEvent)
      .where(eq(outboxEvent.id, eventId))
      .limit(1)
    return current ? (current.status as ProcessSingleOutboxResult) : 'not_found'
  }
  if (typeof event === 'string') return event
  return runHandler(event, handlers)
}

/**
 * Reaper: move `processing` rows whose worker died (stale `lockedAt`)