338 lines
11 KiB
TypeScript
338 lines
11 KiB
TypeScript
import { and, eq, sql } 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,
|
|
ClaimBusinessEventDeliveriesInput,
|
|
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 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 = result as unknown as Array<typeof businessEventOutbox.$inferSelect>;
|
|
return rows.map((row) => mapDelivery(row));
|
|
}
|
|
|
|
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, 'retry_scheduled', 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' | 'retry_scheduled' | '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(),
|
|
claimedBy: row.claimedBy,
|
|
claimedAt: row.claimedAt?.toISOString() ?? null,
|
|
claimExpiresAt: row.claimExpiresAt?.toISOString() ?? null,
|
|
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
|
|
};
|
|
}
|