trails-cool/trails
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #465 from trails-cool/federation/inbox-queue

Browse source 
feat(journal): async federation inbox via Fedify message queue
This commit is contained in:
Ullrich Schäfer 2026-06-06 22:05:32 +02:00 • committed by GitHub
commit 0cb2a901fe
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 19 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

20
apps/journal/app/lib/federation.server.ts
View file

@ -26,7 +26,7 @@
// exactly the "acknowledge and drop" the spec wants; the same applies // exactly the "acknowledge and drop" the spec wants; the same applies
// to wrapped types we inspect and ignore (e.g. Undo(Like)). // to wrapped types we inspect and ignore (e.g. Undo(Like)).
import { configure, getConsoleSink } from "@logtape/logtape"; import { configure, getConsoleSink } from "@logtape/logtape";
import { createFederation, type Federation } from "@fedify/fedify"; import { createFederation, InProcessMessageQueue, type Federation } from "@fedify/fedify";
import { Accept, Follow, Person, PropertyValue, Reject, Undo } from "@fedify/fedify/vocab"; import { Accept, Follow, Person, PropertyValue, Reject, Undo } from "@fedify/fedify/vocab";
import { eq } from "drizzle-orm"; import { eq } from "drizzle-orm";
import { users } from "@trails-cool/db/schema/journal"; import { users } from "@trails-cool/db/schema/journal";
@ -129,10 +129,28 @@ function buildFederation(): Federation<void> {
configureFederationLogging(); configureFederationLogging();
const federation = createFederation<void>({ const federation = createFederation<void>({
kv: new PostgresKvStore(), kv: new PostgresKvStore(),
// Message queue so inbox listener execution and outbound deliveries
// (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(),
// Started explicitly below — keeps vitest runs (which build this
// instance via handleFederationRequest) free of dangling timers.
manuallyStartQueue: true,
// Canonical origin so generated IRIs are correct behind the Caddy // Canonical origin so generated IRIs are correct behind the Caddy
// proxy (the Node server itself only sees plain HTTP). // proxy (the Node server itself only sees plain HTTP).
origin: getOrigin(), origin: getOrigin(),
}); });
if (!process.env.VITEST) {
// Fire-and-forget: runs the queue consumer loop for the process
// lifetime. Dev and prod both reach this on first federation use.
void federation.startQueue(undefined);
}
federation federation
.setActorDispatcher("/users/{identifier}", async (ctx, identifier) => { .setActorDispatcher("/users/{identifier}", async (ctx, identifier) => {