task ep.1.4.1

src/features/foundation/projections/server/store.ts

@@ -1,4 +1,4 @@
-import { and, eq } from 'drizzle-orm';
+import { and, eq, sql } from 'drizzle-orm';
 import {
   businessEventOutbox,
   projectionConsumerCheckpoints,
@@ -10,6 +10,7 @@ import type { BusinessEvent } from '../../business-events/types.ts';
 import type {
   BeginProjectionCheckpointInput,
   BusinessEventDeliveryRecord,
+  ClaimBusinessEventDeliveriesInput,
   CompleteProjectionCheckpointInput,
   CreateProjectionDeadLetterInput,
   FailProjectionCheckpointInput,
@@ -47,6 +48,43 @@ export class DrizzleProjectionRuntimeStore implements ProjectionRuntimeStore {
     return mapDelivery(created);
   }
 
+  async claimAvailableEvents(
+    input: ClaimBusinessEventDeliveriesInput
+  ): Promise<BusinessEventDeliveryRecord[]> {
+    const result = await db.execute(sql`
+      WITH claimable AS (
+        SELECT id
+        FROM business_event_outbox
+        WHERE
+          (
+            delivery_status IN ('pending', 'retry_scheduled')
+            AND available_at <= ${input.now}
+          )
+          OR (
+            delivery_status IN ('claimed', 'processing')
+            AND claim_expires_at IS NOT NULL
+            AND claim_expires_at <= ${input.now}
+          )
+        ORDER BY available_at ASC, created_at ASC, id ASC
+        LIMIT ${input.batchSize}
+        FOR UPDATE SKIP LOCKED
+      )
+      UPDATE business_event_outbox
+      SET
+        delivery_status = 'claimed',
+        claimed_by = ${input.workerId},
+        claimed_at = ${input.now},
+        claim_expires_at = ${input.leaseExpiresAt},
+        updated_at = ${input.now}
+      FROM claimable
+      WHERE business_event_outbox.id = claimable.id
+      RETURNING business_event_outbox.*
+    `);
+
+    const rows = Array.isArray(result) ? result : result.rows;
+    return rows.map((row) => mapDelivery(row as typeof businessEventOutbox.$inferSelect));
+  }
+
   async markDeliveryProcessing(eventId: string, now: Date): Promise<void> {
     const existing = await this.getDelivery(eventId);
     await db
@@ -65,7 +103,7 @@ export class DrizzleProjectionRuntimeStore implements ProjectionRuntimeStore {
   }
 
   async markDeliveryFailed(eventId: string, error: ProjectionFailure, now: Date): Promise<void> {
-    await updateDeliveryStatus(eventId, 'failed', now, {
+    await updateDeliveryStatus(eventId, 'retry_scheduled', now, {
       failedAt: now,
       lastErrorCode: error.code,
       lastErrorMessage: error.message
@@ -253,7 +291,7 @@ export class DrizzleProjectionRuntimeStore implements ProjectionRuntimeStore {
 
 async function updateDeliveryStatus(
   eventId: string,
-  deliveryStatus: 'completed' | 'failed' | 'dead_letter',
+  deliveryStatus: 'completed' | 'retry_scheduled' | 'dead_letter',
   now: Date,
   values: Partial<typeof businessEventOutbox.$inferInsert>
 ): Promise<void> {
@@ -272,6 +310,9 @@ function mapDelivery(row: typeof businessEventOutbox.$inferSelect): BusinessEven
     status: row.deliveryStatus as BusinessEventDeliveryRecord['status'],
     attemptCount: row.attemptCount,
     availableAt: row.availableAt.toISOString(),
+    claimedBy: row.claimedBy,
+    claimedAt: row.claimedAt?.toISOString() ?? null,
+    claimExpiresAt: row.claimExpiresAt?.toISOString() ?? null,
     event: row.eventEnvelope as BusinessEvent
   };
 }