task ep.1.4

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

@@ -0,0 +1,296 @@
+import { and, eq } from 'drizzle-orm';
+import {
+  businessEventOutbox,
+  projectionConsumerCheckpoints,
+  projectionDeadLetters,
+  projectionHealthSnapshots
+} from '@/db/schema';
+import { db } from '@/lib/db';
+import type { BusinessEvent } from '../../business-events/types.ts';
+import type {
+  BeginProjectionCheckpointInput,
+  BusinessEventDeliveryRecord,
+  CompleteProjectionCheckpointInput,
+  CreateProjectionDeadLetterInput,
+  FailProjectionCheckpointInput,
+  ProjectionCheckpointRecord,
+  ProjectionRuntimeStore,
+  UnsupportedProjectionCheckpointInput,
+  ProjectionHealthSnapshotInput,
+  ProjectionFailure
+} from '../types.ts';
+
+export class DrizzleProjectionRuntimeStore implements ProjectionRuntimeStore {
+  async persistEvent(event: BusinessEvent): Promise<BusinessEventDeliveryRecord> {
+    const existing = await this.getDelivery(event.eventId);
+    if (existing) {
+      return existing;
+    }
+
+    const [created] = await db
+      .insert(businessEventOutbox)
+      .values({
+        id: event.eventId,
+        organizationId: event.organizationId,
+        eventType: event.eventType,
+        schemaVersion: event.schemaVersion,
+        entityType: event.entityType,
+        entityId: event.entityId,
+        correlationId: event.correlationId,
+        causationId: event.causationId ?? null,
+        eventEnvelope: event,
+        deliveryStatus: 'pending',
+        occurredAt: new Date(event.occurredAt)
+      })
+      .returning();
+
+    return mapDelivery(created);
+  }
+
+  async markDeliveryProcessing(eventId: string, now: Date): Promise<void> {
+    const existing = await this.getDelivery(eventId);
+    await db
+      .update(businessEventOutbox)
+      .set({
+        deliveryStatus: 'processing',
+        attemptCount: (existing?.attemptCount ?? 0) + 1,
+        processingStartedAt: now,
+        updatedAt: now
+      })
+      .where(eq(businessEventOutbox.id, eventId));
+  }
+
+  async markDeliveryCompleted(eventId: string, now: Date): Promise<void> {
+    await updateDeliveryStatus(eventId, 'completed', now, { dispatchedAt: now });
+  }
+
+  async markDeliveryFailed(eventId: string, error: ProjectionFailure, now: Date): Promise<void> {
+    await updateDeliveryStatus(eventId, 'failed', now, {
+      failedAt: now,
+      lastErrorCode: error.code,
+      lastErrorMessage: error.message
+    });
+  }
+
+  async markDeliveryDeadLetter(eventId: string, error: ProjectionFailure, now: Date): Promise<void> {
+    await updateDeliveryStatus(eventId, 'dead_letter', now, {
+      deadLetteredAt: now,
+      lastErrorCode: error.code,
+      lastErrorMessage: error.message
+    });
+  }
+
+  async getCheckpoint(
+    consumerName: string,
+    eventId: string
+  ): Promise<ProjectionCheckpointRecord | null> {
+    const [row] = await db
+      .select()
+      .from(projectionConsumerCheckpoints)
+      .where(
+        and(
+          eq(projectionConsumerCheckpoints.consumerName, consumerName),
+          eq(projectionConsumerCheckpoints.eventId, eventId)
+        )
+      )
+      .limit(1);
+
+    return row ? mapCheckpoint(row) : null;
+  }
+
+  async beginCheckpoint(input: BeginProjectionCheckpointInput): Promise<ProjectionCheckpointRecord> {
+    const existing = await this.getCheckpoint(input.consumerName, input.eventId);
+
+    if (existing) {
+      const [updated] = await db
+        .update(projectionConsumerCheckpoints)
+        .set({
+          processingStatus: 'processing',
+          attemptCount: existing.attemptCount + 1,
+          lastAttemptedAt: input.now,
+          updatedAt: input.now
+        })
+        .where(
+          and(
+            eq(projectionConsumerCheckpoints.consumerName, input.consumerName),
+            eq(projectionConsumerCheckpoints.eventId, input.eventId)
+          )
+        )
+        .returning();
+      return mapCheckpoint(updated);
+    }
+
+    const [created] = await db
+      .insert(projectionConsumerCheckpoints)
+      .values({
+        id: crypto.randomUUID(),
+        organizationId: input.organizationId,
+        consumerName: input.consumerName,
+        projectionName: input.projectionName,
+        eventId: input.eventId,
+        eventType: input.eventType,
+        schemaVersion: input.schemaVersion,
+        processingStatus: 'processing',
+        attemptCount: 1,
+        firstAttemptedAt: input.now,
+        lastAttemptedAt: input.now
+      })
+      .returning();
+
+    return mapCheckpoint(created);
+  }
+
+  async completeCheckpoint(input: CompleteProjectionCheckpointInput): Promise<void> {
+    await db
+      .update(projectionConsumerCheckpoints)
+      .set({
+        processingStatus: 'completed',
+        completedAt: input.now,
+        nextRetryAt: null,
+        errorCode: null,
+        errorMessage: null,
+        processingDurationMs: Math.round(input.durationMs),
+        updatedAt: input.now
+      })
+      .where(
+        and(
+          eq(projectionConsumerCheckpoints.consumerName, input.consumerName),
+          eq(projectionConsumerCheckpoints.eventId, input.eventId)
+        )
+      );
+  }
+
+  async failCheckpoint(input: FailProjectionCheckpointInput): Promise<void> {
+    await db
+      .update(projectionConsumerCheckpoints)
+      .set({
+        processingStatus: input.deadLetter ? 'dead_letter' : 'retry_scheduled',
+        nextRetryAt: input.nextRetryAt,
+        errorCode: input.failure.code,
+        errorMessage: input.failure.message,
+        processingDurationMs: Math.round(input.durationMs),
+        updatedAt: input.now
+      })
+      .where(
+        and(
+          eq(projectionConsumerCheckpoints.consumerName, input.consumerName),
+          eq(projectionConsumerCheckpoints.eventId, input.eventId)
+        )
+      );
+  }
+
+  async skipUnsupportedVersion(input: UnsupportedProjectionCheckpointInput): Promise<void> {
+    const existing = await this.getCheckpoint(input.consumerName, input.eventId);
+    if (existing) {
+      return;
+    }
+
+    await db.insert(projectionConsumerCheckpoints).values({
+      id: crypto.randomUUID(),
+      organizationId: input.organizationId,
+      consumerName: input.consumerName,
+      projectionName: input.projectionName,
+      eventId: input.eventId,
+      eventType: input.eventType,
+      schemaVersion: input.schemaVersion,
+      processingStatus: 'skipped_unsupported_version',
+      attemptCount: 0,
+      errorCode: 'unsupported_event_version',
+      errorMessage: `Unsupported schema version ${input.schemaVersion}`
+    });
+  }
+
+  async createDeadLetter(input: CreateProjectionDeadLetterInput): Promise<void> {
+    await db.insert(projectionDeadLetters).values({
+      id: crypto.randomUUID(),
+      organizationId: input.organizationId,
+      consumerName: input.consumerName,
+      projectionName: input.projectionName,
+      eventId: input.eventId,
+      eventType: input.eventType,
+      schemaVersion: input.schemaVersion,
+      attemptCount: input.attemptCount,
+      errorCode: input.failure.code,
+      errorMessage: input.failure.message,
+      failureMetadata: {
+        retryable: input.failure.retryable
+      }
+    });
+  }
+
+  async recordHealthSnapshot(input: ProjectionHealthSnapshotInput): Promise<void> {
+    await db.insert(projectionHealthSnapshots).values({
+      id: crypto.randomUUID(),
+      organizationId: input.organizationId,
+      projectionName: input.projectionName,
+      consumerName: input.consumerName,
+      status: input.status,
+      lastProcessedEventId: input.lastProcessedEventId ?? null,
+      lastSuccessfulProcessingAt: input.lastSuccessfulProcessingAt
+        ? new Date(input.lastSuccessfulProcessingAt)
+        : null,
+      pendingCount: input.pendingCount ?? 0,
+      retryCount: input.retryCount ?? 0,
+      deadLetterCount: input.deadLetterCount ?? 0,
+      unsupportedVersionCount: input.unsupportedVersionCount ?? 0,
+      processingLagMs: input.processingLagMs ?? 0,
+      averageProcessingDurationMs: input.averageProcessingDurationMs ?? 0,
+      lastRebuildAt: input.lastRebuildAt ? new Date(input.lastRebuildAt) : null,
+      rebuildSource: input.rebuildSource ?? null
+    });
+  }
+
+  private async getDelivery(eventId: string): Promise<BusinessEventDeliveryRecord | null> {
+    const [row] = await db
+      .select()
+      .from(businessEventOutbox)
+      .where(eq(businessEventOutbox.id, eventId))
+      .limit(1);
+
+    return row ? mapDelivery(row) : null;
+  }
+}
+
+async function updateDeliveryStatus(
+  eventId: string,
+  deliveryStatus: 'completed' | 'failed' | 'dead_letter',
+  now: Date,
+  values: Partial<typeof businessEventOutbox.$inferInsert>
+): Promise<void> {
+  await db
+    .update(businessEventOutbox)
+    .set({ ...values, deliveryStatus, updatedAt: now })
+    .where(eq(businessEventOutbox.id, eventId));
+}
+
+function mapDelivery(row: typeof businessEventOutbox.$inferSelect): BusinessEventDeliveryRecord {
+  return {
+    eventId: row.id,
+    organizationId: row.organizationId,
+    eventType: row.eventType,
+    schemaVersion: row.schemaVersion,
+    status: row.deliveryStatus as BusinessEventDeliveryRecord['status'],
+    attemptCount: row.attemptCount,
+    availableAt: row.availableAt.toISOString(),
+    event: row.eventEnvelope as BusinessEvent
+  };
+}
+
+function mapCheckpoint(
+  row: typeof projectionConsumerCheckpoints.$inferSelect
+): ProjectionCheckpointRecord {
+  return {
+    id: row.id,
+    organizationId: row.organizationId,
+    consumerName: row.consumerName,
+    projectionName: row.projectionName,
+    eventId: row.eventId,
+    eventType: row.eventType,
+    schemaVersion: row.schemaVersion,
+    status: row.processingStatus as ProjectionCheckpointRecord['status'],
+    attemptCount: row.attemptCount,
+    nextRetryAt: row.nextRetryAt?.toISOString() ?? null,
+    errorCode: row.errorCode,
+    errorMessage: row.errorMessage
+  };
+}