feat(journal): durable Fedify message queue over pg-boss

Task group 1 of federation-hardening. Fedify was configured with
`InProcessMessageQueue`, so every queued outbound delivery, its pending
retry state, and inbox processing task was lost on a container restart
(routine here: deploys, OOM history) — directly contradicting the
social-federation spec's promise that fan-out survives a deploy.

- `federation-queue.server.ts`: `PgBossMessageQueue` implementing Fedify's
  `MessageQueue` over the pg-boss instance the journal already runs,
  mirroring the `PostgresKvStore` adapter. `nativeRetrial = false` keeps
  Fedify the retry-policy owner; pg-boss supplies durability + delayed
  jobs (delay → whole-second `startAfter`, `retryLimit: 0`).
- Swap it in for `InProcessMessageQueue` in `federation.server.ts`;
  document the two intentional queueing layers (our fan-out jobs feed
  Fedify; Fedify's sends now durable underneath).
- `server.ts`: create the durable queue at startup when federation is on.
- Broaden the structural `BossLike` in `boss.server.ts` with the
  work/offWork/createQueue/getQueue methods the adapter needs.
- Tests: enqueue maps retry/delay correctly, consume roundtrip, restart
  durability (fresh listener drains a prior instance's backlog), abort
  stops the worker, depth reports ready vs delayed.

Verified: journal typecheck + lint clean, 7/7 new unit tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/journal/app/lib/boss.server.ts

@@ -18,18 +18,40 @@
 import { logger } from "./logger.server.ts";
 import type { JobName, JobPayloads } from "../jobs/payloads.ts";
 
-// Structurally typed (we only need `send`) so we don't have to pull
-// pg-boss into the journal app's dep graph just for the typedef.
+// Structurally typed so we don't have to pull pg-boss into the journal
+// app's dep graph just for the typedef. Kept to the subset feature code
+// actually uses: `send` for enqueues, plus `work`/`offWork`/`createQueue`/
+// `getQueue` for the Fedify message-queue adapter (federation-queue.server.ts).
 export interface BossSendOptions {
   retryLimit?: number;
   retryBackoff?: boolean;
   retryDelay?: number;
+  /** Seconds to defer the job (pg-boss `startAfter`). */
+  startAfter?: number;
   /** pg-boss dedup: collapses enqueues sharing a key into one active job. */
   singletonKey?: string;
 }
 
+/** A pg-boss job as our handlers see it (subset of pg-boss's `Job`). */
+export interface BossJob<T = unknown> {
+  id: string;
+  name: string;
+  data: T;
+}
+
+/** Queue depth counters (subset of pg-boss's `QueueResult`). */
+export interface BossQueueDepth {
+  queuedCount: number;
+  readyCount: number;
+  deferredCount: number;
+}
+
 interface BossLike {
   send(queueName: string, data: unknown, options?: BossSendOptions): Promise<string | null>;
+  work(queueName: string, handler: (jobs: BossJob[]) => Promise<unknown>): Promise<string>;
+  offWork(queueName: string): Promise<void>;
+  createQueue(queueName: string, options?: unknown): Promise<void>;
+  getQueue(queueName: string): Promise<BossQueueDepth | null>;
 }
 
 const BOSS_KEY = Symbol.for("trails-cool.journal.pg-boss");

apps/journal/app/lib/federation-queue.server.test.ts

@@ -0,0 +1,129 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { setBoss } from "./boss.server.ts";
+import { PgBossMessageQueue, FEDERATION_QUEUE_NAME } from "./federation-queue.server.ts";
+
+// A durable in-memory stand-in for pg-boss: `sent` survives across adapter
+// instances (like rows in Postgres), which is what lets us model a restart
+// — a fresh PgBossMessageQueue.listen() draining messages a previous
+// instance enqueued. Only the methods the adapter uses are implemented.
+interface SentJob {
+  data: unknown;
+  startAfter?: number;
+  retryLimit?: number;
+  consumed: boolean;
+}
+
+class FakeBoss {
+  sent: SentJob[] = [];
+  offWorkCalls: string[] = [];
+
+  async send(name: string, data: unknown, options?: { startAfter?: number; retryLimit?: number }) {
+    expect(name).toBe(FEDERATION_QUEUE_NAME);
+    this.sent.push({ data, startAfter: options?.startAfter, retryLimit: options?.retryLimit, consumed: false });
+    return "job-id";
+  }
+
+  // Drain every ready (non-delayed, unconsumed) job through the handler,
+  // mirroring a worker that picks up the durable backlog on startup.
+  async work(name: string, handler: (jobs: Array<{ id: string; name: string; data: unknown }>) => Promise<unknown>) {
+    expect(name).toBe(FEDERATION_QUEUE_NAME);
+    for (const job of this.sent) {
+      if (job.consumed || (job.startAfter ?? 0) > 0) continue;
+      job.consumed = true;
+      await handler([{ id: "j", name, data: job.data }]);
+    }
+    return "worker-id";
+  }
+
+  async offWork(name: string) {
+    this.offWorkCalls.push(name);
+  }
+
+  async createQueue() {}
+
+  async getQueue(name: string) {
+    expect(name).toBe(FEDERATION_QUEUE_NAME);
+    const pending = this.sent.filter((j) => !j.consumed);
+    const delayed = pending.filter((j) => (j.startAfter ?? 0) > 0).length;
+    return { queuedCount: pending.length, readyCount: pending.length - delayed, deferredCount: delayed };
+  }
+}
+
+let boss: FakeBoss;
+
+beforeEach(() => {
+  boss = new FakeBoss();
+  setBoss(boss as unknown as Parameters<typeof setBoss>[0]);
+});
+
+afterEach(() => {
+  setBoss(null);
+});
+
+describe("PgBossMessageQueue", () => {
+  it("enqueues with no retry (Fedify owns retries) and no delay by default", async () => {
+    const q = new PgBossMessageQueue();
+    await q.enqueue({ hello: "world" });
+    expect(boss.sent).toHaveLength(1);
+    const [job] = boss.sent;
+    expect(job).toMatchObject({ data: { hello: "world" }, retryLimit: 0 });
+    expect(job?.startAfter).toBeUndefined();
+  });
+
+  it("maps a Temporal delay to whole-second startAfter (rounded up)", async () => {
+    const q = new PgBossMessageQueue();
+    const delay = { total: () => 1.2 } as unknown as Temporal.Duration;
+    await q.enqueue({ x: 1 }, { delay });
+    expect(boss.sent[0]?.startAfter).toBe(2);
+  });
+
+  it("declares nativeRetrial=false so Fedify keeps ownership of retries", () => {
+    expect(new PgBossMessageQueue().nativeRetrial).toBe(false);
+  });
+
+  it("consume roundtrip: a listener receives the enqueued message payload", async () => {
+    const q = new PgBossMessageQueue();
+    await q.enqueue({ type: "Create", id: "a" });
+    const received: unknown[] = [];
+    const ac = new AbortController();
+    // listen() resolves only on abort; register the worker (which drains
+    // the backlog), then stop.
+    const listening = q.listen((m) => void received.push(m), { signal: ac.signal });
+    ac.abort();
+    await listening;
+    expect(received).toEqual([{ type: "Create", id: "a" }]);
+  });
+
+  it("survives a restart: a fresh listener drains messages a prior instance enqueued", async () => {
+    // Prior process: enqueue two, never consumed (crash before delivery).
+    const before = new PgBossMessageQueue();
+    await before.enqueue({ n: 1 });
+    await before.enqueue({ n: 2 });
+
+    // New process: same durable store, brand-new adapter instance.
+    const after = new PgBossMessageQueue();
+    const received: unknown[] = [];
+    const ac = new AbortController();
+    const listening = after.listen((m) => void received.push(m), { signal: ac.signal });
+    ac.abort();
+    await listening;
+
+    expect(received).toEqual([{ n: 1 }, { n: 2 }]);
+  });
+
+  it("listen resolves on abort and stops the worker", async () => {
+    const q = new PgBossMessageQueue();
+    const ac = new AbortController();
+    const listening = q.listen(() => {}, { signal: ac.signal });
+    ac.abort();
+    await expect(listening).resolves.toBeUndefined();
+    expect(boss.offWorkCalls).toContain(FEDERATION_QUEUE_NAME);
+  });
+
+  it("reports queue depth split into ready and delayed", async () => {
+    const q = new PgBossMessageQueue();
+    await q.enqueue({ n: 1 });
+    await q.enqueue({ n: 2 }, { delay: { total: () => 30 } as unknown as Temporal.Duration });
+    expect(await q.getDepth()).toEqual({ queued: 2, ready: 1, delayed: 1 });
+  });
+});

apps/journal/app/lib/federation-queue.server.ts

@@ -0,0 +1,77 @@
+// Postgres-backed Fedify MessageQueue, implemented over the pg-boss
+// instance the journal already runs (packages/jobs). Fedify uses this
+// queue for outbound delivery (with its own retry scheduling) and inbox
+// processing; the previous InProcessMessageQueue lost every queued
+// message on restart (spec: federation-operations "Durable federation
+// queue"). pg-boss gives us durable storage + delayed jobs; Fedify stays
+// the retry-policy owner (see `nativeRetrial` below).
+//
+// This mirrors the PostgresKvStore adapter pattern (federation-kv.server.ts):
+// a thin Fedify-interface shim over infrastructure we already operate,
+// resolved lazily via getBoss() so it works in both the server-bootstrap
+// module and the request-path bundle (see boss.server.ts on the two-copy
+// module situation).
+//
+// Fedify fans a single MessageQueue instance out to three logical roles
+// (inbox / outbox / fanout) and calls listen() once per role with an
+// equivalent type-dispatching handler. Backing all three with one pg-boss
+// queue is correct: any worker can process any message because Fedify
+// routes by message type internally.
+
+import type {
+  MessageQueue,
+  MessageQueueEnqueueOptions,
+  MessageQueueListenOptions,
+  MessageQueueDepth,
+} from "@fedify/fedify";
+import { getBoss } from "./boss.server.ts";
+
+/** pg-boss queue name backing Fedify's inbox/outbox/fanout message queue. */
+export const FEDERATION_QUEUE_NAME = "federation.fedify";
+
+export class PgBossMessageQueue implements MessageQueue {
+  // Fedify owns retry scheduling: on a failed delivery it re-enqueues with
+  // its own backoff policy, so pg-boss must NOT also retry (retryLimit 0
+  // on enqueue) or every failure would be attempted twice.
+  readonly nativeRetrial = false;
+
+  async enqueue(message: unknown, options?: MessageQueueEnqueueOptions): Promise<void> {
+    // Fedify passes delay as a Temporal.Duration; pg-boss startAfter is
+    // whole seconds. Round up so a sub-second backoff still defers.
+    const seconds = options?.delay ? Math.ceil(options.delay.total("seconds")) : 0;
+    await getBoss().send(FEDERATION_QUEUE_NAME, message as object, {
+      retryLimit: 0,
+      ...(seconds > 0 ? { startAfter: seconds } : {}),
+    });
+  }
+
+  async listen(
+    handler: (message: unknown) => Promise<void> | void,
+    options?: MessageQueueListenOptions,
+  ): Promise<void> {
+    const boss = getBoss();
+    await boss.work(FEDERATION_QUEUE_NAME, async (jobs) => {
+      for (const job of jobs) await handler(job.data);
+    });
+    // Fedify's contract: listen() resolves when the signal aborts and never
+    // rejects; with no signal it never resolves (runs for the process
+    // lifetime, consuming the queue).
+    return new Promise<void>((resolve) => {
+      const signal = options?.signal;
+      if (signal == null) return;
+      const stop = () => void boss.offWork(FEDERATION_QUEUE_NAME).finally(() => resolve());
+      if (signal.aborted) stop();
+      else signal.addEventListener("abort", stop, { once: true });
+    });
+  }
+
+  async getDepth(): Promise<MessageQueueDepth> {
+    const queue = await getBoss().getQueue(FEDERATION_QUEUE_NAME);
+    if (queue == null) return { queued: 0, ready: 0, delayed: 0 };
+    return {
+      queued: queue.queuedCount,
+      ready: queue.readyCount,
+      delayed: queue.deferredCount,
+    };
+  }
+}

apps/journal/app/lib/federation.server.ts

@@ -27,7 +27,7 @@
 // exactly the "acknowledge and drop" the spec wants; the same applies
 // to wrapped types we inspect and ignore (e.g. Undo(Like)).
 import { configure, getConsoleSink } from "@logtape/logtape";
-import { createFederation, InProcessMessageQueue, type Federation } from "@fedify/fedify";
+import { createFederation, type Federation } from "@fedify/fedify";
 import { Accept, Follow, Note, Person, PropertyValue, Reject, Undo } from "@fedify/fedify/vocab";
 import { and, eq } from "drizzle-orm";
 import { activities, users } from "@trails-cool/db/schema/journal";
@@ -35,6 +35,7 @@ import { getDb } from "./db.ts";
 import { getOrigin } from "./config.server.ts";
 import { localActorIri } from "./actor-iri.ts";
 import { PostgresKvStore } from "./federation-kv.server.ts";
+import { PgBossMessageQueue } from "./federation-queue.server.ts";
 import { ensureUserKeypair, loadUserKeypair } from "./federation-keys.server.ts";
 import { activityToCreate, activityToNote } from "./federation-objects.server.ts";
 import {
@@ -148,12 +149,17 @@ function buildFederation(): Federation<void> {
     // (e.g. the Accept we push back after a Follow) happen asynchronously
     // with Fedify's own retry policy, instead of inside the inbound HTTP
     // request — Mastodon gives deliveries a 10s read timeout, and a slow
-    // remote must never make us blow it. In-process is acceptable for
-    // the single-process journal: queued work is lost on restart, but
-    // every message type we enqueue is recoverable (remotes retry
-    // Follows; our own pg-boss jobs own activity delivery). Revisit with
-    // a Postgres-backed MessageQueue if that stops being true.
-    queue: new InProcessMessageQueue(),
+    // remote must never make us blow it.
+    //
+    // Two queueing layers coexist by design: our fan-out jobs
+    // (federation-delivery.server.ts enqueues one deliverActivity pg-boss
+    // job per follower) feed Fedify, and Fedify's own send/retry
+    // scheduling now runs on THIS durable pg-boss-backed queue instead of
+    // in-process. Before, in-process meant a deploy or crash dropped every
+    // queued delivery and pending retry (spec: fan-out survives a deploy);
+    // PgBossMessageQueue closes that. Collapsing the two layers into one
+    // is a refactor with no user-visible payoff, so they stay separate.
+    queue: new PgBossMessageQueue(),
     // Started explicitly below — keeps vitest runs (which build this
     // instance via handleFederationRequest) free of dangling timers.
     manuallyStartQueue: true,

apps/journal/server.ts

@@ -165,6 +165,12 @@ server.listen(port, async () => {
   // users get keys before any federation traffic). Each run only
   // touches users whose public_key IS NULL, so repeats are no-ops.
   if (process.env.FEDERATION_ENABLED === "true") {
+    // Create the durable queue backing Fedify's message queue before any
+    // federation traffic can enqueue/consume on it. pg-boss requires queues
+    // to exist explicitly (v10+); createQueue is idempotent.
+    const { FEDERATION_QUEUE_NAME } = await import("./app/lib/federation-queue.server.ts");
+    await boss.createQueue(FEDERATION_QUEUE_NAME);
+
     await enqueue("backfill-user-keypairs", {});
     logger.info("federation keypair backfill enqueued");
   }