feat(events): process transactional invalidations

lib/outbox/processor.test.ts

@@ -0,0 +1,62 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { processOutboxEvent, type OutboxTransport } from "./processor";
+
+function record(overrides: Partial<Parameters<typeof processOutboxEvent>[0]> = {}) {
+  return {
+    topic: "page:72df08ab-50dd-4cbd-9a69-70949d34cf9f",
+    aggregateId: "72df08ab-50dd-4cbd-9a69-70949d34cf9f",
+    eventType: "page.updated",
+    payload: {
+      id: "72df08ab-50dd-4cbd-9a69-70949d34cf9f",
+      version: 8,
+      privateContent: "must not leave PostgreSQL",
+    },
+    ...overrides,
+  };
+}
+
+describe("transactional outbox processor", () => {
+  it("invalidates before publishing a privacy-safe event", async () => {
+    const calls: string[] = [];
+    const transport: OutboxTransport = {
+      invalidate: vi.fn(async () => {
+        calls.push("invalidate");
+      }),
+      publish: vi.fn(async () => {
+        calls.push("publish");
+      }),
+    };
+
+    const event = await processOutboxEvent(record(), transport, 1234);
+
+    expect(calls).toEqual(["invalidate", "publish"]);
+    expect(transport.invalidate).toHaveBeenCalledWith(
+      ["buzz:page:72df08ab-50dd-4cbd-9a69-70949d34cf9f"],
+      1234,
+    );
+    expect(transport.publish).toHaveBeenCalledWith(
+      "page:72df08ab-50dd-4cbd-9a69-70949d34cf9f",
+      {
+        type: "page.updated",
+        id: "72df08ab-50dd-4cbd-9a69-70949d34cf9f",
+        version: 8,
+      },
+    );
+    expect(event).not.toHaveProperty("privateContent");
+  });
+
+  it("rejects a mismatched topic or malformed version", async () => {
+    const transport: OutboxTransport = {
+      invalidate: vi.fn(),
+      publish: vi.fn(),
+    };
+    await expect(
+      processOutboxEvent(record({ topic: "admin" }), transport),
+    ).rejects.toThrow("invalid public envelope");
+    await expect(
+      processOutboxEvent(record({ payload: { id: record().aggregateId, version: "8" } }), transport),
+    ).rejects.toThrow("invalid public envelope");
+    expect(transport.invalidate).not.toHaveBeenCalled();
+  });
+});

lib/outbox/processor.ts

@@ -0,0 +1,53 @@
+import type Redis from "ioredis";
+
+import { cacheTagsForOutboxEvent } from "@/lib/cache/tags";
+import {
+  expectedTopic,
+  toInvalidationEvent,
+  type InvalidationEvent,
+  type OutboxEventLike,
+} from "@/lib/events/invalidation";
+import {
+  redisCacheTagStateKey,
+  redisEventChannel,
+} from "@/lib/redis/client";
+
+export interface OutboxTransport {
+  invalidate(tags: string[], timestamp: number): Promise<void>;
+  publish(topic: string, event: InvalidationEvent): Promise<void>;
+}
+
+export function createRedisOutboxTransport(client: Redis): OutboxTransport {
+  return {
+    async invalidate(tags, timestamp) {
+      if (tags.length === 0) return;
+      const pipeline = client.pipeline();
+      for (const tag of tags) {
+        pipeline.hset(
+          redisCacheTagStateKey(),
+          tag,
+          JSON.stringify({ expired: timestamp }),
+        );
+      }
+      const results = await pipeline.exec();
+      const failure = results?.find(([cause]) => cause);
+      if (failure?.[0]) throw failure[0];
+    },
+    async publish(topic, event) {
+      await client.publish(redisEventChannel(topic), JSON.stringify(event));
+    },
+  };
+}
+
+export async function processOutboxEvent(
+  record: OutboxEventLike,
+  transport: OutboxTransport,
+  timestamp = Date.now(),
+): Promise<InvalidationEvent> {
+  const event = toInvalidationEvent(record);
+  if (!event) throw new Error("Outbox event has an invalid public envelope or topic.");
+  const tags = cacheTagsForOutboxEvent(record);
+  await transport.invalidate(tags, timestamp);
+  await transport.publish(expectedTopic(event), event);
+  return event;
+}

lib/outbox/repository.test.ts

@@ -0,0 +1,11 @@
+import { describe, expect, it } from "vitest";
+
+import { retryDelayMs } from "./repository";
+
+describe("outbox retry policy", () => {
+  it("uses capped exponential backoff", () => {
+    expect(retryDelayMs(1)).toBe(500);
+    expect(retryDelayMs(4)).toBe(4_000);
+    expect(retryDelayMs(30)).toBe(60_000);
+  });
+});

lib/outbox/repository.ts

@@ -0,0 +1,89 @@
+import { and, asc, eq, inArray, lte, sql } from "drizzle-orm";
+
+import { getDb } from "@/db/client";
+import { outboxEvents } from "@/db/schema";
+
+export type ClaimedOutboxEvent = Pick<
+  typeof outboxEvents.$inferSelect,
+  "id" | "topic" | "aggregateId" | "eventType" | "payload" | "attempts"
+>;
+
+const CLAIMABLE_STATUSES = ["pending", "processing"] as const;
+
+export async function claimOutboxEvents(
+  batchSize = 20,
+  leaseMs = 30_000,
+  now = new Date(),
+): Promise<ClaimedOutboxEvent[]> {
+  const size = Math.max(1, Math.min(100, Math.floor(batchSize)));
+  const leaseUntil = new Date(now.getTime() + Math.max(5_000, leaseMs));
+
+  return getDb().transaction(async (tx) => {
+    const candidates = await tx
+      .select({ id: outboxEvents.id })
+      .from(outboxEvents)
+      .where(
+        and(
+          inArray(outboxEvents.status, [...CLAIMABLE_STATUSES]),
+          lte(outboxEvents.availableAt, now),
+        ),
+      )
+      .orderBy(asc(outboxEvents.id))
+      .limit(size)
+      .for("update", { skipLocked: true });
+
+    const ids = candidates.map(({ id }) => id);
+    if (ids.length === 0) return [];
+    return tx
+      .update(outboxEvents)
+      .set({
+        status: "processing",
+        attempts: sql<number>`${outboxEvents.attempts} + 1`,
+        availableAt: leaseUntil,
+      })
+      .where(inArray(outboxEvents.id, ids))
+      .returning({
+        id: outboxEvents.id,
+        topic: outboxEvents.topic,
+        aggregateId: outboxEvents.aggregateId,
+        eventType: outboxEvents.eventType,
+        payload: outboxEvents.payload,
+        attempts: outboxEvents.attempts,
+      });
+  });
+}
+
+export async function markOutboxProcessed(
+  id: number,
+  now = new Date(),
+): Promise<void> {
+  await getDb()
+    .update(outboxEvents)
+    .set({
+      status: "processed",
+      processedAt: now,
+      lastError: null,
+    })
+    .where(and(eq(outboxEvents.id, id), eq(outboxEvents.status, "processing")));
+}
+
+export function retryDelayMs(attempts: number): number {
+  return Math.min(60_000, 500 * 2 ** Math.max(0, Math.min(attempts - 1, 16)));
+}
+
+export async function releaseOutboxEvent(
+  id: number,
+  attempts: number,
+  cause: unknown,
+  now = new Date(),
+): Promise<void> {
+  const message = cause instanceof Error ? cause.message : String(cause);
+  await getDb()
+    .update(outboxEvents)
+    .set({
+      status: "pending",
+      availableAt: new Date(now.getTime() + retryDelayMs(attempts)),
+      lastError: message.slice(0, 2_000),
+    })
+    .where(and(eq(outboxEvents.id, id), eq(outboxEvents.status, "processing")));
+}