Function claimBatch

apps/sim/lib/core/outbox/service.ts:273–305

* Phase 1: claim a batch of due pending events. * * `SELECT ... FOR UPDATE SKIP LOCKED` atomically picks rows that no * other worker is currently looking at. We then flip those rows to * `processing` inside the same tx so the claim survives the lock * release — the status change becomes the out

(batchSize: number)

 * release — the status change becomes the out-of-band mutual exclusion.
 */
async function claimBatch(batchSize: number): Promise<(typeof outboxEvent.$inferSelect)[]> {
  const now = new Date()
  return db.transaction(async (tx) => {
    const rows = await tx
      .select()
      .from(outboxEvent)
      .where(and(eq(outboxEvent.status, 'pending'), lte(outboxEvent.availableAt, now)))
      .orderBy(asc(outboxEvent.createdAt))
      .limit(batchSize)
      .for('update', { skipLocked: true })

    if (rows.length === 0) return []

    await tx
      .update(outboxEvent)
      .set({ status: 'processing', lockedAt: now })
      .where(
        inArray(
          outboxEvent.id,
          rows.map((r) => r.id)
        )
      )

    // Return rows with the claim state we just committed. `lockedAt`
    // on this object is the authoritative lease timestamp used by the
    // terminal-update lease CAS (see `runHandler`).
    return rows.map((row) => ({
      ...row,
      status: 'processing' as const,
      lockedAt: now,
    }))
  })
}

/**
 * Phase 2: invoke the handler for a claimed event, outside any DB