BPOS-MVP/apps/api/src/modules/event/internal/adapter/event-store.ts

/**
 * event 存储 adapter —— Drizzle 只准出现在 adapter 与 app(编码规范 §5.3)。
 *
 * 「先持久后投递」:信封与投递记录在**同一事务**落库(ADR-008)。
 */
import { and, asc, desc, eq, lte, sql } from 'drizzle-orm';
import type { Delivery, Envelope, EventId, RegisteredListener, TaskId } from '@contracts';
import { deliverySchema, envelopeSchema, registeredListenerSchema } from '@contracts';
import type { Db } from '@shared/db.ts';
import type { EventStore } from '../../port.ts';
import type { OutboundRef } from '../domain/ownership.ts';
import { consumptionTable, deliveryTable, eventTable, listenerTable, outboundRefTable } from './event-schema.ts';

type EventRow = typeof eventTable.$inferSelect;
type DeliveryRow = typeof deliveryTable.$inferSelect;
type ListenerRow = typeof listenerTable.$inferSelect;

function toEnvelope(row: EventRow): Envelope {
  return envelopeSchema.parse({
    id: row.id,
    type: row.type,
    correlation: row.correlation,
    payload: row.payload,
    occurredAt: row.occurredAt.toISOString(),
    source: row.source,
    trace: row.trace,
  });
}

function toDelivery(row: DeliveryRow): Delivery {
  return deliverySchema.parse({
    id: row.id,
    eventId: row.eventId,
    target: { kind: row.targetKind, id: row.targetId },
    attempt: row.attempt,
    state: row.state,
    ...(row.lastError === null ? {} : { lastError: row.lastError }),
  });
}

function toListener(row: ListenerRow): RegisteredListener {
  return registeredListenerSchema.parse({
    id: row.id,
    taskId: row.taskId,
    origin: row.origin,
    eventType: row.eventType,
    keyFilter: row.keyFilter,
    ...(row.condition === null ? {} : { condition: row.condition }),
    semantic: row.semantic,
    deadline: row.deadline,
    escalation: row.escalation,
    registeredAt: row.registeredAt.toISOString(),
    expiresAt: row.expiresAt.toISOString(),
    state: row.state,
  });
}

export function createEventStore(db: Db, ids: { next(): string }): EventStore {
  return {
    async persistAndEnqueue(envelope, targets) {
      await db.transaction(async (tx) => {
        await tx.insert(eventTable).values({
          id: envelope.id,
          type: envelope.type,
          correlation: envelope.correlation,
          payload: envelope.payload,
          occurredAt: new Date(envelope.occurredAt),
          source: envelope.source,
          trace: envelope.trace,
          targetKind: envelope.correlation.target?.kind ?? null,
          targetId: envelope.correlation.target?.id ?? null,
        });
        if (targets.length === 0) return;
        await tx.insert(deliveryTable).values(
          targets.map((d) => ({
            id: d.id,
            eventId: d.eventId,
            targetKind: d.target.kind,
            targetId: d.target.id,
            attempt: d.attempt,
            state: d.state,
            lastError: d.lastError ?? null,
          })),
        );
      });
    },

    async get(id: EventId) {
      const rows = await db.select().from(eventTable).where(eq(eventTable.id, id)).limit(1);
      return rows[0] === undefined ? undefined : toEnvelope(rows[0]);
    },

    async listByTask(taskId: TaskId, limit: number) {
      const rows = await db
        .select()
        .from(eventTable)
        .where(and(eq(eventTable.targetKind, 'task'), eq(eventTable.targetId, taskId)))
        .orderBy(asc(eventTable.occurredAt))
        .limit(limit);
      return rows.map(toEnvelope);
    },

    async listenersOf(taskId: TaskId) {
      const rows = await db
        .select()
        .from(listenerTable)
        .where(eq(listenerTable.taskId, taskId))
        .orderBy(desc(listenerTable.registeredAt));
      return rows.map(toListener);
    },

    async armedListeners() {
      const rows = await db.select().from(listenerTable).where(eq(listenerTable.state, 'armed'));
      return rows.map(toListener);
    },

    async saveListener(listener) {
      await db.insert(listenerTable).values({
        id: listener.id,
        taskId: listener.taskId,
        origin: listener.origin,
        eventType: listener.eventType,
        keyFilter: listener.keyFilter,
        condition: listener.condition ?? null,
        semantic: listener.semantic,
        deadline: listener.deadline,
        escalation: listener.escalation,
        registeredAt: new Date(listener.registeredAt),
        expiresAt: new Date(listener.expiresAt),
        state: listener.state,
      });
    },

    async setListenerState(id, state) {
      await db.update(listenerTable).set({ state }).where(eq(listenerTable.id, id));
    },

    async expiredListeners(now: Date) {
      const rows = await db
        .select()
        .from(listenerTable)
        .where(and(eq(listenerTable.state, 'armed'), lte(listenerTable.expiresAt, now)));
      return rows.map(toListener);
    },

    async saveOutbound(ref: OutboundRef) {
      await db.insert(outboundRefTable).values({
        id: ids.next(),
        taskId: ref.taskId,
        threadId: ref.threadId ?? null,
        address: ref.address,
        sentAt: ref.sentAt,
      });
    },

    async recentOutbound(sinceMs: number, now: Date) {
      const since = new Date(now.getTime() - sinceMs);
      const rows = await db
        .select()
        .from(outboundRefTable)
        .where(sql`${outboundRefTable.sentAt} >= ${since}`)
        .orderBy(desc(outboundRefTable.sentAt))
        .limit(500);
      return rows.map((r) => ({
        taskId: r.taskId as TaskId,
        ...(r.threadId === null ? {} : { threadId: r.threadId }),
        address: r.address,
        sentAt: r.sentAt,
      }));
    },

    async claimConsumption(eventId: EventId, consumer: string) {
      const inserted = await db
        .insert(consumptionTable)
        .values({ eventId, consumer })
        .onConflictDoNothing()
        .returning({ eventId: consumptionTable.eventId });
      return inserted.length > 0;
    },

    async getDelivery(id: string) {
      const rows = await db.select().from(deliveryTable).where(eq(deliveryTable.id, id)).limit(1);
      return rows[0] === undefined ? undefined : toDelivery(rows[0]);
    },

    async updateDelivery(delivery) {
      await db
        .update(deliveryTable)
        .set({ attempt: delivery.attempt, state: delivery.state, lastError: delivery.lastError ?? null })
        .where(eq(deliveryTable.id, delivery.id));
    },

    async scheduleRetry(deliveryId: string, delayMs: number) {
      await db
        .update(deliveryTable)
        .set({ nextAttemptAt: sql`now() + make_interval(secs => ${delayMs / 1000})` })
        .where(eq(deliveryTable.id, deliveryId));
    },
  };
}

/**
 * 派发器的取件语句(ADR-008 #2):按目标 ID 取 advisory lock,**同任务按序**。
 * 由 app 层的 worker 调用;放在 adapter 是因为它是 SQL。
 */
export async function claimPendingDeliveries(db: Db, limit: number): Promise<Delivery[]> {
  const rows = await db
    .select()
    .from(deliveryTable)
    .where(and(eq(deliveryTable.state, 'pending'), lte(deliveryTable.nextAttemptAt, sql`now()`)))
    .orderBy(asc(deliveryTable.nextAttemptAt))
    .limit(limit)
    .for('update', { skipLocked: true });
  return rows.map(toDelivery);
}