diff --git a/apps/journal/app/jobs/federation-kv-sweep.ts b/apps/journal/app/jobs/federation-kv-sweep.ts new file mode 100644 index 0000000..b21a7bb --- /dev/null +++ b/apps/journal/app/jobs/federation-kv-sweep.ts @@ -0,0 +1,20 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { PostgresKvStore } from "../lib/federation-kv.server.ts"; +import { logger } from "../lib/logger.server.ts"; + +/** + * Daily cleanup of expired federation_kv rows (Fedify replay-protection + * nonces and caches carry TTLs; reads already filter expired rows, this + * keeps the table from growing unbounded). + */ +export const federationKvSweepJob: JobDefinition = { + name: "federation-kv-sweep", + cron: "15 4 * * *", // daily at 04:15 UTC (offset from the other sweeps) + retryLimit: 1, + expireInSeconds: 60, + async handler() { + const purged = await new PostgresKvStore().sweepExpired(); + logger.info({ purged }, "federation-kv-sweep"); + return { purged }; + }, +}; diff --git a/apps/journal/app/jobs/notifications-fanout.ts b/apps/journal/app/jobs/notifications-fanout.ts index 8d9581e..b73f5ef 100644 --- a/apps/journal/app/jobs/notifications-fanout.ts +++ b/apps/journal/app/jobs/notifications-fanout.ts @@ -59,7 +59,9 @@ export async function fanout(activityId: string): Promise { return; } - // Find every accepted follower of the owner. + // Find every accepted *local* follower of the owner. Remote + // followers (follower_id NULL, follower_actor_iri set) get the + // activity via federation push delivery, not via notifications. const recipients = await db .select({ followerId: follows.followerId }) .from(follows) @@ -67,6 +69,7 @@ export async function fanout(activityId: string): Promise { and( eq(follows.followedUserId, row.ownerId), isNotNull(follows.acceptedAt), + isNotNull(follows.followerId), ), ); @@ -75,7 +78,7 @@ export async function fanout(activityId: string): Promise { // Don't notify the owner about their own activity if they happen // to follow themselves (shouldn't happen — followUser refuses // self-follow — but defense in depth). - if (r.followerId === row.ownerId) continue; + if (r.followerId === row.ownerId || r.followerId === null) continue; const created = await createNotification({ type: "activity_published", recipientUserId: r.followerId, diff --git a/apps/journal/app/lib/federation-inbox.integration.test.ts b/apps/journal/app/lib/federation-inbox.integration.test.ts new file mode 100644 index 0000000..313c0b4 --- /dev/null +++ b/apps/journal/app/lib/federation-inbox.integration.test.ts @@ -0,0 +1,155 @@ +import { describe, it, expect, beforeAll, afterEach } from "vitest"; +import { eq } from "drizzle-orm"; +import { randomUUID } from "node:crypto"; +import { getDb } from "./db.ts"; +import { users, follows } from "@trails-cool/db/schema/journal"; +import { localActorIri } from "./actor-iri.ts"; +import { + recordRemoteFollow, + removeRemoteFollow, + settleOutgoingFollow, + rejectOutgoingFollow, +} from "./federation-inbox.server.ts"; + +// Opt-in: these talk to real Postgres. Gated by an env flag so laptop +// runs without Postgres aren't blocked. Same convention as +// follow.integration.test.ts. +const runIntegration = process.env.FEDERATION_INTEGRATION === "1"; + +const REMOTE_ACTOR = "https://other-trails.example/users/alice"; + +const createdUserIds: string[] = []; + +async function makeUser(opts: { username: string; profileVisibility?: "public" | "private" }) { + const db = getDb(); + const id = randomUUID(); + await db.insert(users).values({ + id, + email: `${opts.username}@example.test`, + username: opts.username, + domain: "test.local", + profileVisibility: opts.profileVisibility ?? "public", + }); + createdUserIds.push(id); + return id; +} + +describe.runIf(runIntegration)("federation inbox (integration)", () => { + beforeAll(() => { + process.env.ORIGIN ??= "http://localhost:3000"; + }); + + afterEach(async () => { + const db = getDb(); + for (const id of createdUserIds.splice(0)) { + await db.delete(follows).where(eq(follows.followedUserId, id)); + await db.delete(follows).where(eq(follows.followerId, id)); + await db.delete(users).where(eq(users.id, id)); + } + }); + + it("inbound Follow on a public user records an accepted remote follow", async () => { + const username = `fed-pub-${Date.now()}`; + const userId = await makeUser({ username }); + + const { outcome } = await recordRemoteFollow(REMOTE_ACTOR, username); + expect(outcome).toBe("accepted"); + + const db = getDb(); + const rows = await db.select().from(follows).where(eq(follows.followedUserId, userId)); + expect(rows).toHaveLength(1); + expect(rows[0]!.followerActorIri).toBe(REMOTE_ACTOR); + expect(rows[0]!.followerId).toBeNull(); + expect(rows[0]!.followedActorIri).toBe(localActorIri(username)); + expect(rows[0]!.acceptedAt).not.toBeNull(); + }); + + it("inbound Follow is idempotent — replays don't double-insert", async () => { + const username = `fed-replay-${Date.now()}`; + const userId = await makeUser({ username }); + + await recordRemoteFollow(REMOTE_ACTOR, username); + await recordRemoteFollow(REMOTE_ACTOR, username); + + const db = getDb(); + const rows = await db.select().from(follows).where(eq(follows.followedUserId, userId)); + expect(rows).toHaveLength(1); + }); + + it("inbound Follow on a private user is refused with no row", async () => { + const username = `fed-priv-${Date.now()}`; + const userId = await makeUser({ username, profileVisibility: "private" }); + + const { outcome } = await recordRemoteFollow(REMOTE_ACTOR, username); + expect(outcome).toBe("refused"); + + const db = getDb(); + const rows = await db.select().from(follows).where(eq(follows.followedUserId, userId)); + expect(rows).toHaveLength(0); + }); + + it("Undo(Follow) removes the remote follow row", async () => { + const username = `fed-undo-${Date.now()}`; + const userId = await makeUser({ username }); + await recordRemoteFollow(REMOTE_ACTOR, username); + + await removeRemoteFollow(REMOTE_ACTOR, username); + + const db = getDb(); + const rows = await db.select().from(follows).where(eq(follows.followedUserId, userId)); + expect(rows).toHaveLength(0); + }); + + it("Accept(Follow) settles a Pending outgoing follow exactly once", async () => { + const username = `fed-accept-${Date.now()}`; + const userId = await makeUser({ username }); + const db = getDb(); + await db.insert(follows).values({ + id: randomUUID(), + followerId: userId, + followedActorIri: REMOTE_ACTOR, + acceptedAt: null, + }); + + const first = await settleOutgoingFollow(userId, REMOTE_ACTOR); + expect(first.settled).toBe(true); + // A replayed Accept is a no-op (row no longer Pending). + const second = await settleOutgoingFollow(userId, REMOTE_ACTOR); + expect(second.settled).toBe(false); + + const rows = await db.select().from(follows).where(eq(follows.followerId, userId)); + expect(rows[0]!.acceptedAt).not.toBeNull(); + }); + + it("Reject(Follow) deletes the Pending outgoing follow", async () => { + const username = `fed-reject-${Date.now()}`; + const userId = await makeUser({ username }); + const db = getDb(); + await db.insert(follows).values({ + id: randomUUID(), + followerId: userId, + followedActorIri: REMOTE_ACTOR, + acceptedAt: null, + }); + + await rejectOutgoingFollow(userId, REMOTE_ACTOR); + + const rows = await db.select().from(follows).where(eq(follows.followerId, userId)); + expect(rows).toHaveLength(0); + }); + + it("database enforces exactly-one-follower invariant", async () => { + const username = `fed-check-${Date.now()}`; + const userId = await makeUser({ username }); + const db = getDb(); + await expect( + db.insert(follows).values({ + id: randomUUID(), + // Neither followerId nor followerActorIri — must violate the + // follows_has_follower_check constraint. + followedActorIri: localActorIri(username), + followedUserId: userId, + }), + ).rejects.toThrow(); + }); +}); diff --git a/apps/journal/app/lib/federation-inbox.server.ts b/apps/journal/app/lib/federation-inbox.server.ts new file mode 100644 index 0000000..270314a --- /dev/null +++ b/apps/journal/app/lib/federation-inbox.server.ts @@ -0,0 +1,119 @@ +// Domain logic behind the federation inbox (spec: social-federation, +// "Narrow inbox — follow-graph activities only"). Pure DB operations, +// kept separate from the Fedify listener wiring in federation.server.ts +// so they're directly unit/integration-testable without HTTP signatures. + +import { randomUUID } from "node:crypto"; +import { and, eq, isNull } from "drizzle-orm"; +import { users, follows } from "@trails-cool/db/schema/journal"; +import { getDb } from "./db.ts"; +import { localActorIri } from "./actor-iri.ts"; +import { logger } from "./logger.server.ts"; + +/** + * Inbound `Follow` from a remote actor targeting a local user. + * Auto-accepts for public profiles (records the follow row); private + * profiles refuse without leaking existence — the caller maps `refused` + * to the same 404 the actor endpoint serves. + * + * Idempotent: a duplicate Follow from the same actor upserts onto the + * `(follower_actor_iri, followed_user_id)` partial unique index, so + * replays and Mastodon's retry storms cannot double-insert. + */ +export async function recordRemoteFollow( + remoteActorIri: string, + localUsername: string, +): Promise<{ outcome: "accepted" | "refused" }> { + const db = getDb(); + const [user] = await db + .select({ id: users.id, profileVisibility: users.profileVisibility }) + .from(users) + .where(eq(users.username, localUsername)) + .limit(1); + if (!user || user.profileVisibility !== "public") { + return { outcome: "refused" }; + } + await db + .insert(follows) + .values({ + id: randomUUID(), + followerActorIri: remoteActorIri, + followedActorIri: localActorIri(localUsername), + followedUserId: user.id, + // Auto-accept: public profiles accept every follow (locked + // accounts are a later change). + acceptedAt: new Date(), + }) + .onConflictDoNothing(); + logger.info({ remoteActorIri, localUsername }, "federation: inbound follow accepted"); + return { outcome: "accepted" }; +} + +/** Inbound `Undo(Follow)`: remove the remote actor's follow row. */ +export async function removeRemoteFollow( + remoteActorIri: string, + localUsername: string, +): Promise { + const db = getDb(); + const deleted = await db + .delete(follows) + .where( + and( + eq(follows.followerActorIri, remoteActorIri), + eq(follows.followedActorIri, localActorIri(localUsername)), + ), + ) + .returning({ id: follows.id }); + if (deleted.length > 0) { + logger.info({ remoteActorIri, localUsername }, "federation: inbound follow undone"); + } +} + +/** + * Inbound `Accept(Follow)`: a remote instance accepted our local user's + * outgoing follow — settle the Pending row. Returns the settled row's + * remote actor IRI so the caller can enqueue the first outbox poll + * (spec: "First poll triggered immediately on accepted follow"). + */ +export async function settleOutgoingFollow( + localUserId: string, + remoteActorIri: string, +): Promise<{ settled: boolean }> { + const db = getDb(); + const updated = await db + .update(follows) + .set({ acceptedAt: new Date() }) + .where( + and( + eq(follows.followerId, localUserId), + eq(follows.followedActorIri, remoteActorIri), + isNull(follows.acceptedAt), + ), + ) + .returning({ id: follows.id }); + if (updated.length > 0) { + logger.info({ localUserId, remoteActorIri }, "federation: outgoing follow accepted"); + } + return { settled: updated.length > 0 }; +} + +/** Inbound `Reject(Follow)`: the remote refused — drop our Pending row. */ +export async function rejectOutgoingFollow( + localUserId: string, + remoteActorIri: string, +): Promise { + const db = getDb(); + const deleted = await db + .delete(follows) + .where( + and( + eq(follows.followerId, localUserId), + eq(follows.followedActorIri, remoteActorIri), + isNull(follows.acceptedAt), + ), + ) + .returning({ id: follows.id }); + if (deleted.length > 0) { + logger.info({ localUserId, remoteActorIri }, "federation: outgoing follow rejected by remote"); + } +} diff --git a/apps/journal/app/lib/federation-kv.integration.test.ts b/apps/journal/app/lib/federation-kv.integration.test.ts new file mode 100644 index 0000000..69807f1 --- /dev/null +++ b/apps/journal/app/lib/federation-kv.integration.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect, afterAll } from "vitest"; +import { like } from "drizzle-orm"; +import { getDb } from "./db.ts"; +import { federationKv } from "@trails-cool/db/schema/journal"; +import { PostgresKvStore } from "./federation-kv.server.ts"; + +// Opt-in: talks to real Postgres (same convention as the other +// *.integration.test.ts files). +const runIntegration = process.env.FEDERATION_INTEGRATION === "1"; + +// Namespace keys per test run so concurrent runs can't collide. +const NS = `test-${Date.now()}`; + +describe.runIf(runIntegration)("PostgresKvStore (integration)", () => { + const store = new PostgresKvStore(); + + afterAll(async () => { + const db = getDb(); + await db.delete(federationKv).where(like(federationKv.key, `["${NS}"%`)); + }); + + it("roundtrips JSON values", async () => { + await store.set([NS, "a"], { hello: "world", n: 42 }); + expect(await store.get([NS, "a"])).toEqual({ hello: "world", n: 42 }); + }); + + it("returns undefined for missing keys", async () => { + expect(await store.get([NS, "missing"])).toBeUndefined(); + }); + + it("overwrites on set (upsert)", async () => { + await store.set([NS, "b"], 1); + await store.set([NS, "b"], 2); + expect(await store.get([NS, "b"])).toBe(2); + }); + + it("expired values read as missing and are swept", async () => { + // Duck-typed Temporal.Duration — the store only calls + // .total("milliseconds"), and pulling in the polyfill just for the + // test isn't worth a dependency. + const oneMs = { total: () => 1 } as unknown as Temporal.Duration; + await store.set([NS, "ttl"], "ephemeral", { ttl: oneMs }); + await new Promise((r) => setTimeout(r, 10)); + expect(await store.get([NS, "ttl"])).toBeUndefined(); + const purged = await store.sweepExpired(); + expect(purged).toBeGreaterThanOrEqual(1); + }); + + it("deletes keys", async () => { + await store.set([NS, "c"], "x"); + await store.delete([NS, "c"]); + expect(await store.get([NS, "c"])).toBeUndefined(); + }); + + it("lists by prefix without matching sibling prefixes", async () => { + await store.set([NS, "list", "one"], 1); + await store.set([NS, "list", "two"], 2); + await store.set([NS, "listish"], 3); // shares textual prefix, different key path + + const entries = []; + for await (const e of store.list([NS, "list"])) entries.push(e); + const keys = entries.map((e) => e.key.join("/")).sort(); + expect(keys).toEqual([`${NS}/list/one`, `${NS}/list/two`]); + }); +}); diff --git a/apps/journal/app/lib/federation-kv.server.ts b/apps/journal/app/lib/federation-kv.server.ts new file mode 100644 index 0000000..fdc9b87 --- /dev/null +++ b/apps/journal/app/lib/federation-kv.server.ts @@ -0,0 +1,81 @@ +// Postgres-backed Fedify KvStore. Fedify uses its KvStore for inbox +// replay protection (activity-id idempotence), remote document caching, +// and key caching — all state that must survive process restarts and be +// shared across instances of the journal server, which rules out +// MemoryKvStore outside the dev loop (spec: social-federation 4.7). +// +// Keys are KvKey (readonly string[]) serialized as their JSON encoding; +// prefix listing relies on JSON arrays sharing a stable textual prefix +// up to the closing bracket. + +import type { KvKey, KvStore, KvStoreListEntry, KvStoreSetOptions } from "@fedify/fedify"; +import { eq, sql, like, and, or, isNull, gt, lt } from "drizzle-orm"; +import { federationKv } from "@trails-cool/db/schema/journal"; +import { getDb } from "./db.ts"; + +function serializeKey(key: KvKey): string { + return JSON.stringify(key); +} + +/** Predicate: row is not expired (expires_at NULL or in the future). */ +function notExpired() { + return or(isNull(federationKv.expiresAt), gt(federationKv.expiresAt, new Date())); +} + +export class PostgresKvStore implements KvStore { + async get(key: KvKey): Promise { + const db = getDb(); + const [row] = await db + .select({ value: federationKv.value }) + .from(federationKv) + .where(and(eq(federationKv.key, serializeKey(key)), notExpired())) + .limit(1); + return row === undefined ? undefined : (row.value as T); + } + + async set(key: KvKey, value: unknown, options?: KvStoreSetOptions): Promise { + const db = getDb(); + const expiresAt = options?.ttl + ? new Date(Date.now() + options.ttl.total("milliseconds")) + : null; + await db + .insert(federationKv) + .values({ key: serializeKey(key), value, expiresAt }) + .onConflictDoUpdate({ + target: federationKv.key, + set: { value, expiresAt }, + }); + } + + async delete(key: KvKey): Promise { + const db = getDb(); + await db.delete(federationKv).where(eq(federationKv.key, serializeKey(key))); + } + + async *list(prefix?: KvKey): AsyncIterable { + const db = getDb(); + const wherePrefix = prefix + ? // '["a","b"]' minus the trailing ']' is a textual prefix of every + // key extending ["a","b"] — and of nothing else, since the next + // character is either ']' (exact) or ',' (extension). + like(federationKv.key, `${serializeKey(prefix).slice(0, -1)}%`) + : undefined; + const rows = await db + .select({ key: federationKv.key, value: federationKv.value }) + .from(federationKv) + .where(wherePrefix ? and(wherePrefix, notExpired()) : notExpired()); + for (const row of rows) { + yield { key: JSON.parse(row.key) as KvKey, value: row.value }; + } + } + + /** Remove expired rows. Called by the federation-kv-sweep job. */ + async sweepExpired(): Promise { + const db = getDb(); + const deleted = await db + .delete(federationKv) + .where(and(sql`${federationKv.expiresAt} IS NOT NULL`, lt(federationKv.expiresAt, new Date()))) + .returning({ key: federationKv.key }); + return deleted.length; + } +} diff --git a/apps/journal/app/lib/federation.server.test.ts b/apps/journal/app/lib/federation.server.test.ts index 880247e..fdad691 100644 --- a/apps/journal/app/lib/federation.server.test.ts +++ b/apps/journal/app/lib/federation.server.test.ts @@ -19,7 +19,7 @@ vi.mock("./db.ts", () => ({ }), })); -const { handleFederationRequest, wantsActivityJson } = await import( +const { handleFederationRequest, wantsActivityJson, federationSourceHost } = await import( "./federation.server.ts" ); @@ -104,6 +104,33 @@ describe("actor object", () => { }); }); +describe("federationSourceHost", () => { + it("extracts the host from the HTTP Signature keyId", () => { + const req = new Request("http://localhost:3000/users/bruno/inbox", { + method: "POST", + headers: { + signature: + 'keyId="https://mastodon.social/users/alice#main-key",algorithm="rsa-sha256",headers="(request-target) host date digest",signature="abc=="', + }, + }); + expect(federationSourceHost(req)).toBe("mastodon.social"); + }); + + it("buckets unsigned or malformed requests as unknown", () => { + expect( + federationSourceHost(new Request("http://localhost:3000/users/bruno/inbox", { method: "POST" })), + ).toBe("unknown"); + expect( + federationSourceHost( + new Request("http://localhost:3000/users/bruno/inbox", { + method: "POST", + headers: { signature: 'keyId="not a url",signature="x"' }, + }), + ), + ).toBe("unknown"); + }); +}); + describe("wantsActivityJson", () => { it("matches AP client Accept headers, not browsers", () => { expect(wantsActivityJson(actorRequest("bruno"))).toBe(true); diff --git a/apps/journal/app/lib/federation.server.ts b/apps/journal/app/lib/federation.server.ts index 1bda9bd..7ba410f 100644 --- a/apps/journal/app/lib/federation.server.ts +++ b/apps/journal/app/lib/federation.server.ts @@ -1,9 +1,5 @@ // ActivityPub federation via Fedify. See openspec/changes/social-federation. // -// Spike scope (task 1.1): a minimal Federation instance serving WebFinger -// discovery + Person actor objects for public users. No keypairs, no inbox -// processing, no delivery yet — those land in later tasks. -// // Version pinning (task 1.2): @fedify/fedify is pinned to exactly 2.1.16. // The whole 2.2.x line is uninstallable from npm — each release depends on // @fedify/webfinger@2.2.x, which was never published (latest is 2.1.16). @@ -11,23 +7,47 @@ // // Mounting model: Fedify owns URL dispatch for its endpoints via // `federation.fetch(request)`. We delegate to it from React Router routes: -// - /.well-known/webfinger → resource route (no component, raw Response) -// - /users/:username → route middleware short-circuits to Fedify -// when the request asks for an ActivityPub representation; browsers -// fall through to the HTML profile loader (content negotiation per -// the design decision "actor IRI is the profile URL"). -import { createFederation, MemoryKvStore, type Federation } from "@fedify/fedify"; -import { Person } from "@fedify/fedify/vocab"; +// - /.well-known/webfinger → resource route (raw Response) +// - /.well-known/nodeinfo, +// /nodeinfo/2.1 → resource route (software discovery for +// the trails-to-trails outbound check — NodeInfo is the fediverse +// standard for this; the actor-level `software` AS extension the +// tasks file originally sketched isn't expressible with Fedify's +// typed vocab, and NodeInfo is what Mastodon et al. actually read) +// - /users/:username → route middleware short-circuits to +// Fedify when the request asks for an ActivityPub representation; +// browsers fall through to the HTML profile loader +// - /users/:username/inbox → resource route (POST), rate-limited +// per source instance before Fedify verifies the HTTP Signature +// +// Inbox scope (spec: "Narrow inbox — follow-graph activities only"): +// Follow, Undo(Follow), Accept(Follow), Reject(Follow). Fedify returns +// 202 for activity types without a registered listener, which is +// exactly the "acknowledge and drop" the spec wants; the same applies +// to wrapped types we inspect and ignore (e.g. Undo(Like)). +import { createFederation, type Federation } from "@fedify/fedify"; +import { Accept, Follow, Person, Reject, Undo } from "@fedify/fedify/vocab"; import { eq } from "drizzle-orm"; import { users } from "@trails-cool/db/schema/journal"; 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 { ensureUserKeypair, loadUserKeypair } from "./federation-keys.server.ts"; +import { + recordRemoteFollow, + removeRemoteFollow, + settleOutgoingFollow, + rejectOutgoingFollow, +} from "./federation-inbox.server.ts"; +import { enqueueOptional } from "./boss.server.ts"; +import { logger } from "./logger.server.ts"; /** * Feature flag (task 1.3). All federation surfaces — WebFinger, actor - * objects, inbox, outbox — are disabled (404) unless FEDERATION_ENABLED - * is exactly "true". Default off so schema/code can deploy before any - * federation traffic is possible. + * objects, inbox, outbox, NodeInfo — are disabled (404) unless + * FEDERATION_ENABLED is exactly "true". Default off so schema/code can + * deploy before any federation traffic is possible. */ export function federationEnabled(): boolean { return process.env.FEDERATION_ENABLED === "true"; @@ -42,41 +62,159 @@ export function wantsActivityJson(request: Request): boolean { ); } +type UserRow = typeof users.$inferSelect; + +async function findUserByUsername(username: string): Promise { + const db = getDb(); + const [user] = await db + .select() + .from(users) + .where(eq(users.username, username)) + .limit(1); + return user ?? null; +} + +/** + * Resolve a *local, public* user from an actor IRI; null otherwise. + * Matches against our canonical actor IRI shape (`{origin}/users/{name}`, + * the same shape localActorIri produces). + */ +async function findLocalPublicUserByIri(iri: URL | null): Promise { + if (iri === null) return null; + const prefix = `${getOrigin()}/users/`; + if (!iri.href.startsWith(prefix)) return null; + const username = iri.href.slice(prefix.length); + if (!username || username.includes("/")) return null; + const user = await findUserByUsername(username); + return user && user.profileVisibility === "public" ? user : null; +} + let _federation: Federation | null = null; function buildFederation(): Federation { const federation = createFederation({ - // MemoryKvStore is fine for the spike (it only caches remote documents - // and nonces). Replace with a Postgres-backed store before real - // federation traffic (revisit in task 4.x). - kv: new MemoryKvStore(), + kv: new PostgresKvStore(), // Canonical origin so generated IRIs are correct behind the Caddy // proxy (the Node server itself only sees plain HTTP). origin: getOrigin(), }); - federation.setActorDispatcher("/users/{identifier}", async (ctx, identifier) => { - const db = getDb(); - const [user] = await db - .select() - .from(users) - .where(eq(users.username, identifier)) - .limit(1); - // Private profiles do not federate at all: returning null makes Fedify - // respond 404, and WebFinger lookups for this user 404 with it — no - // leak of user existence (spec: "Private user is invisible to - // federation"). - if (!user || user.profileVisibility !== "public") return null; - return new Person({ - id: ctx.getActorUri(identifier), - preferredUsername: identifier, - name: user.displayName ?? user.username, - summary: user.bio || undefined, - // Human-facing profile URL — same as the actor IRI by design, but - // declared explicitly so AP clients link browsers to the HTML view. - url: new URL(`/users/${identifier}`, getOrigin()), + federation + .setActorDispatcher("/users/{identifier}", async (ctx, identifier) => { + const user = await findUserByUsername(identifier); + // Private profiles do not federate at all: returning null makes + // Fedify respond 404, and WebFinger lookups 404 with it — no leak + // of user existence (spec: "Private user is invisible to + // federation"). + if (!user || user.profileVisibility !== "public") return null; + const keys = await ctx.getActorKeyPairs(identifier); + return new Person({ + id: ctx.getActorUri(identifier), + preferredUsername: identifier, + name: user.displayName ?? user.username, + summary: user.bio || undefined, + // Human-facing profile URL — same as the actor IRI by design, + // declared explicitly so AP clients link browsers to HTML. + url: new URL(localActorIri(identifier)), + inbox: ctx.getInboxUri(identifier), + // Public keys for inbound HTTP-Signature verification by + // remotes (Mastodon reads publicKey; newer stacks read the + // multikey assertionMethods). + publicKey: keys[0]?.cryptographicKey, + assertionMethods: keys.map((k) => k.multikey), + }); + }) + .setKeyPairsDispatcher(async (_ctxData, identifier) => { + let user = await findUserByUsername(identifier); + if (!user || user.profileVisibility !== "public") return []; + // The backfill job normally guarantees keys; generate lazily as a + // last line of defense (idempotent, guarded on NULL public_key). + if (!user.publicKey) { + await ensureUserKeypair(user.id); + user = await findUserByUsername(identifier); + if (!user) return []; + } + const pair = await loadUserKeypair(user); + return pair ? [pair] : []; }); - }); + + federation + .setInboxListeners("/users/{identifier}/inbox") + .on(Follow, async (ctx, follow) => { + // Spec 4.2: Follow from any AP-compatible remote — auto-accept + // when the local target is public; otherwise drop (the actor + // already 404s for private users). + if (follow.id == null || follow.actorId == null || follow.objectId == null) return; + const parsed = ctx.parseUri(follow.objectId); + if (parsed?.type !== "actor") return; + const { outcome } = await recordRemoteFollow(follow.actorId.href, parsed.identifier); + if (outcome !== "accepted") return; + const follower = await follow.getActor(ctx); + if (follower == null) return; + await ctx.sendActivity( + { identifier: parsed.identifier }, + follower, + new Accept({ + id: new URL(`${localActorIri(parsed.identifier)}#accepts/${crypto.randomUUID()}`), + actor: ctx.getActorUri(parsed.identifier), + object: follow, + }), + ); + }) + .on(Undo, async (ctx, undo) => { + // Spec 4.3: Undo(Follow) removes the follow row. Other Undos are + // acknowledged and dropped. + const object = await undo.getObject(ctx); + if (!(object instanceof Follow)) return; + if (undo.actorId == null || object.objectId == null) return; + const parsed = ctx.parseUri(object.objectId); + if (parsed?.type !== "actor") return; + await removeRemoteFollow(undo.actorId.href, parsed.identifier); + }) + .on(Accept, async (ctx, accept) => { + // Spec 4.4: a remote accepted our outgoing Follow — settle the + // Pending row and trigger the first outbox poll for that actor. + const object = await accept.getObject(ctx); + if (!(object instanceof Follow)) return; + if (accept.actorId == null) return; + const localUser = await findLocalPublicUserByIri(object.actorId); + if (!localUser) return; + const { settled } = await settleOutgoingFollow(localUser.id, accept.actorId.href); + if (settled) { + // Queue worker lands in the outbox-poll change (task 7.x); + // enqueueOptional logs-and-continues until then. + await enqueueOptional("poll-remote-actor", { actorIri: accept.actorId.href }, { + reason: "first poll after accepted follow", + }); + } + }) + .on(Reject, async (ctx, reject) => { + // Spec 4.5: remote refused our Follow — drop the Pending row. + const object = await reject.getObject(ctx); + if (!(object instanceof Follow)) return; + if (reject.actorId == null) return; + const localUser = await findLocalPublicUserByIri(object.actorId); + if (!localUser) return; + await rejectOutgoingFollow(localUser.id, reject.actorId.href); + }) + .onError(async (_ctx, error) => { + logger.warn({ err: error }, "federation: inbox listener error"); + }); + + // Software discovery (task 3.4, adapted): NodeInfo is the standard + // the fediverse uses to identify server software; the trails-to-trails + // outbound check (task 6.x) reads this from remote instances. + federation.setNodeInfoDispatcher("/nodeinfo/2.1", () => ({ + software: { + name: "trails-cool", + version: process.env.SENTRY_RELEASE ?? "0.0.0-dev", + homepage: new URL("https://trails.cool/"), + }, + protocols: ["activitypub"], + // Deliberately zeroed: publishing per-instance usage counts is a + // privacy decision we haven't made (privacy manifest, task 10.1). + usage: { users: {}, localPosts: 0, localComments: 0 }, + })); return federation; } @@ -97,3 +235,22 @@ export function handleFederationRequest(request: Request): Promise { } return getFederation().fetch(request, { contextData: undefined }); } + +/** + * Extract the source instance host from an inbound federation request + * for per-instance rate limiting (spec 4.8). The HTTP Signature keyId + * is a URL on the sender's instance; its host identifies the instance + * before any verification work. Unsigned junk shares one tight bucket. + */ +export function federationSourceHost(request: Request): string { + const signature = request.headers.get("signature") ?? ""; + const match = /keyId="([^"]+)"/.exec(signature); + if (match?.[1]) { + try { + return new URL(match[1]).host; + } catch { + // unparseable keyId → shared bucket below + } + } + return "unknown"; +} diff --git a/apps/journal/app/lib/follow.server.ts b/apps/journal/app/lib/follow.server.ts index 7508a63..f1254ae 100644 --- a/apps/journal/app/lib/follow.server.ts +++ b/apps/journal/app/lib/follow.server.ts @@ -238,7 +238,10 @@ export async function approveFollowRequest(ownerId: string, followId: string): P .select({ username: users.username, displayName: users.displayName }) .from(users) .where(eq(users.id, ownerId)); - if (target) { + // followerId is NULL for inbound federated follows — those are + // auto-accepted and never appear in the Requests tab, but guard + // anyway: remote followers can't receive local notifications. + if (target && followerId !== null) { await createNotification({ type: "follow_request_approved", recipientUserId: followerId, diff --git a/apps/journal/app/routes.ts b/apps/journal/app/routes.ts index 4fabf60..db8d720 100644 --- a/apps/journal/app/routes.ts +++ b/apps/journal/app/routes.ts @@ -4,6 +4,9 @@ export default [ index("routes/home.tsx"), route(".well-known/trails-cool", "routes/api.well-known.trails-cool.ts"), route(".well-known/webfinger", "routes/api.well-known.webfinger.ts"), + route(".well-known/nodeinfo", "routes/api.well-known.nodeinfo.ts"), + route("nodeinfo/2.1", "routes/api.nodeinfo.ts"), + route("users/:username/inbox", "routes/users.$username.inbox.ts"), route("oauth/authorize", "routes/oauth.authorize.tsx"), route("oauth/token", "routes/oauth.token.ts"), route("auth/register", "routes/auth.register.tsx"), diff --git a/apps/journal/app/routes/api.nodeinfo.ts b/apps/journal/app/routes/api.nodeinfo.ts new file mode 100644 index 0000000..ba56b76 --- /dev/null +++ b/apps/journal/app/routes/api.nodeinfo.ts @@ -0,0 +1,15 @@ +import type { Route } from "./+types/api.nodeinfo"; +import { handleFederationRequest } from "~/lib/federation.server"; + +/** + * GET /nodeinfo/2.1 + * + * Standard fediverse software discovery (NodeInfo). The + * trails-to-trails outbound check reads this from remote instances to + * decide whether a follow target runs trails.cool. Discovery document + * lives at /.well-known/nodeinfo (routes/api.well-known.nodeinfo.ts). + * 404s while FEDERATION_ENABLED is off. + */ +export function loader({ request }: Route.LoaderArgs) { + return handleFederationRequest(request); +} diff --git a/apps/journal/app/routes/api.well-known.nodeinfo.ts b/apps/journal/app/routes/api.well-known.nodeinfo.ts new file mode 100644 index 0000000..0a56353 --- /dev/null +++ b/apps/journal/app/routes/api.well-known.nodeinfo.ts @@ -0,0 +1,11 @@ +import type { Route } from "./+types/api.well-known.nodeinfo"; +import { handleFederationRequest } from "~/lib/federation.server"; + +/** + * GET /.well-known/nodeinfo — NodeInfo discovery document pointing at + * /nodeinfo/2.1 (served by routes/api.nodeinfo.ts). Handled by Fedify; + * 404s while FEDERATION_ENABLED is off. + */ +export function loader({ request }: Route.LoaderArgs) { + return handleFederationRequest(request); +} diff --git a/apps/journal/app/routes/users.$username.inbox.ts b/apps/journal/app/routes/users.$username.inbox.ts new file mode 100644 index 0000000..e4fe0d5 --- /dev/null +++ b/apps/journal/app/routes/users.$username.inbox.ts @@ -0,0 +1,35 @@ +import type { Route } from "./+types/users.$username.inbox"; +import { handleFederationRequest, federationSourceHost } from "~/lib/federation.server"; +import { consumeRateLimit } from "~/lib/rate-limit.server"; + +/** + * POST /users/:username/inbox — ActivityPub inbox (spec: + * social-federation, "Narrow inbox"). Fedify verifies the HTTP + * Signature and dispatches to the registered listeners; everything + * else about the request is its problem. We rate-limit per source + * instance *before* signature verification so a hostile instance + * can't burn CPU on key fetches + crypto (spec 4.8: 60 req / 5 min + * per host). + */ +export function action({ request }: Route.ActionArgs) { + const { allowed, resetMs } = consumeRateLimit({ + scope: "federation-inbox", + key: federationSourceHost(request), + limit: 60, + windowMs: 5 * 60 * 1000, + }); + if (!allowed) { + return Promise.resolve( + new Response("Too Many Requests", { + status: 429, + headers: { "Retry-After": String(Math.ceil(resetMs / 1000)) }, + }), + ); + } + return handleFederationRequest(request); +} + +/** Inboxes are POST-only; GET has no representation here. */ +export function loader() { + return new Response("Method Not Allowed", { status: 405, headers: { Allow: "POST" } }); +} diff --git a/apps/journal/server.ts b/apps/journal/server.ts index 9aaa341..0897533 100644 --- a/apps/journal/server.ts +++ b/apps/journal/server.ts @@ -178,10 +178,11 @@ server.listen(port, async () => { const { sendWelcomeEmailJob } = await import("./app/jobs/send-welcome-email.ts"); const { consumedJtiSweepJob } = await import("./app/jobs/consumed-jti-sweep.ts"); jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob, importBatchesSweepJob, sendWelcomeEmailJob, consumedJtiSweepJob); - // Federation keypair backfill — registered only when federation is on. + // Federation jobs — registered only when federation is on. if (process.env.FEDERATION_ENABLED === "true") { const { backfillUserKeypairsJob } = await import("./app/jobs/backfill-user-keypairs.ts"); - jobs.push(backfillUserKeypairsJob); + const { federationKvSweepJob } = await import("./app/jobs/federation-kv-sweep.ts"); + jobs.push(backfillUserKeypairsJob, federationKvSweepJob); } const boss = createBoss(getDatabaseUrl()); diff --git a/openspec/changes/social-federation/design.md b/openspec/changes/social-federation/design.md index 0ed78c6..b5efc86 100644 --- a/openspec/changes/social-federation/design.md +++ b/openspec/changes/social-federation/design.md @@ -106,6 +106,25 @@ Inbound signature verification uses the actor's public key from their actor obje 6. Flip `FEDERATION_ENABLED` on, soften the home blurb, document the federation runbook in `docs/deployment.md`. 7. Rollback: feature flag off (instant), or revert PR (recoverable). Existing `follows` rows stay intact; remote rows would need cleanup. +## Implementation Decisions (made during apply) + +- **Inbound remote followers live in `follows` with a nullable `follower_id`** + (decided in task 4.2, 2026-06-06). The original claim that `follows` was + federation-ready only covered outbound; a remote follower has no local + `users` row. Added `follower_actor_iri` with a check constraint enforcing + exactly one of (`follower_id`, `follower_actor_iri`), and a partial unique + index deduping `(follower_actor_iri, followed_user_id)`. Existing queries + filter on `follower_id`/`followed_user_id` and are unaffected; follower + counts now naturally include remote followers; notification fan-out + explicitly filters `follower_id IS NOT NULL`. +- **Software identity ships via NodeInfo, not a custom AS actor field** + (task 3.4). Fedify's typed vocab can't emit arbitrary actor properties, and + NodeInfo (`software.name: trails-cool`) is the standard the fediverse reads. + The trails-to-trails outbound check (task 6.x) should read remote NodeInfo, + with `/.well-known/trails-cool` as the secondary signal. +- **Fedify's KvStore is Postgres-backed** (`journal.federation_kv`) so inbox + replay protection and document caches survive restarts; swept daily. + ## Open Questions - **`activities.owner_id` is NOT NULL but remote-ingested rows have no local diff --git a/openspec/changes/social-federation/proposal.md b/openspec/changes/social-federation/proposal.md index abc4049..73f5420 100644 --- a/openspec/changes/social-federation/proposal.md +++ b/openspec/changes/social-federation/proposal.md @@ -45,7 +45,7 @@ Outbound federation is intentionally narrower: trails users can follow other **t - **UI**: Pending state on Follow button, "outgoing follows" section listing Pending requests with cancel option, federation-aware empty states on `/feed`. - **Federation surface**: real federation traffic crosses the public internet for the first time — push to followers, inbox processing, outbox polling. Rate-limited per-host on outbound, per-actor on inbound. - **Dependencies**: Fedify (or chosen equivalent). No other heavy runtime additions. -- **Schema**: additive — `users.public_key`, `users.private_key_encrypted`, `remote_actors` table, `activities.remote_origin_iri/remote_actor_iri/audience`, no changes to `follows` (already federation-ready). +- **Schema**: additive — `users.public_key`, `users.private_key_encrypted`, `remote_actors` table, `activities.remote_origin_iri/remote_actor_iri/audience`, plus (discovered during implementation) `follows.follower_actor_iri` with `follower_id` relaxed to nullable: the original "no changes to `follows`" claim only held for *outbound* follows — an inbound remote follower has no local `users` row to reference, so the follower side needed its own IRI column (exactly-one-of enforced by check constraint). Also `federation_kv` (Fedify KvStore backing table for replay protection + caches). - **Privacy manifest**: federation entry covering inbox/outbox traffic, remote actor cache, and what a remote instance learns when it fetches one of our actor objects. - **Operational**: deploy raises real public-facing concerns — abuse-prone inbox endpoint, outbound rate limits, key rotation. Pre-launch checklist documented in deployment docs. - **Out of scope** (tracked as later changes): diff --git a/openspec/changes/social-federation/tasks.md b/openspec/changes/social-federation/tasks.md index ca37cee..1ec4675 100644 --- a/openspec/changes/social-federation/tasks.md +++ b/openspec/changes/social-federation/tasks.md @@ -14,21 +14,23 @@ ## 3. Identity surface -- [ ] 3.1 `localActorIri(username)` already exists from `social-feed`; ensure it's reused everywhere (no string concat on URLs) -- [ ] 3.2 Implement `/.well-known/webfinger?resource=acct:user@domain` returning JRD with `rel="self"` actor link (gated on `profile_visibility = 'public'`) -- [ ] 3.3 Content-negotiated handler at `/users/:username`: HTML for browsers, `application/activity+json` for AP clients (returning the `Person` actor object). Both gated on `profile_visibility = 'public'` -- [ ] 3.4 Actor object includes `software: trails.cool` (custom AS extension) so other instances can recognize us for the trails-to-trails outbound check +- [x] 3.1 `localActorIri(username)` already exists from `social-feed`; ensure it's reused everywhere (no string concat on URLs) +- [x] 3.2 Implement `/.well-known/webfinger?resource=acct:user@domain` returning JRD with `rel="self"` actor link (gated on `profile_visibility = 'public'`) +- [x] 3.3 Content-negotiated handler at `/users/:username`: HTML for browsers, `application/activity+json` for AP clients (returning the `Person` actor object). Both gated on `profile_visibility = 'public'` +- [x] 3.4 Actor object includes `software: trails.cool` (custom AS extension) so other instances can recognize us for the trails-to-trails outbound check + > Adapted during implementation: shipped as a standard **NodeInfo** endpoint (`/.well-known/nodeinfo` + `/nodeinfo/2.1`, `software.name: trails-cool`) instead of a custom AS field — Fedify's typed vocab can't emit arbitrary actor properties, and NodeInfo is what the fediverse actually reads for software identity. `/.well-known/trails-cool` remains as the secondary discovery surface. ## 4. Inbox -- [ ] 4.1 Inbox endpoint at `/users/:username/inbox`. Verifies HTTP Signature on every request before any further processing -- [ ] 4.2 Handle `Follow`: gate on local user's `profile_visibility`, write follow row, push `Accept(Follow)` back, return 202 -- [ ] 4.3 Handle `Undo(Follow)`: delete the matching row, return 202 -- [ ] 4.4 Handle `Accept(Follow)`: settle our outgoing Pending row's `accepted_at`, enqueue a one-off outbox-poll for the just-accepted actor, return 202 -- [ ] 4.5 Handle `Reject(Follow)`: delete our outgoing follow row, surface in user's outgoing-follows list, return 202 -- [ ] 4.6 Handle every other activity type: return 202 and drop silently (logged at debug) -- [ ] 4.7 Replay protection: dedupe on activity IRI for follow-graph activities (small hot table or Redis set) -- [ ] 4.8 Per-source-instance rate limit on inbox: 60 requests / 5 min from any single host +- [x] 4.1 Inbox endpoint at `/users/:username/inbox`. Verifies HTTP Signature on every request before any further processing +- [x] 4.2 Handle `Follow`: gate on local user's `profile_visibility`, write follow row, push `Accept(Follow)` back, return 202 +- [x] 4.3 Handle `Undo(Follow)`: delete the matching row, return 202 +- [x] 4.4 Handle `Accept(Follow)`: settle our outgoing Pending row's `accepted_at`, enqueue a one-off outbox-poll for the just-accepted actor, return 202 +- [x] 4.5 Handle `Reject(Follow)`: delete our outgoing follow row, surface in user's outgoing-follows list, return 202 + > Protocol part done; the UI notice lands with the outgoing-follows list (task 6.6). +- [x] 4.6 Handle every other activity type: return 202 and drop silently (logged at debug) +- [x] 4.7 Replay protection: dedupe on activity IRI for follow-graph activities (small hot table or Redis set) +- [x] 4.8 Per-source-instance rate limit on inbox: 60 requests / 5 min from any single host ## 5. Outbox + push delivery diff --git a/packages/db/src/schema/journal.ts b/packages/db/src/schema/journal.ts index 1dec133..546de41 100644 --- a/packages/db/src/schema/journal.ts +++ b/packages/db/src/schema/journal.ts @@ -10,6 +10,7 @@ import { customType, uniqueIndex, index, + check, } from "drizzle-orm/pg-core"; const bytea = customType<{ data: Buffer }>({ @@ -330,18 +331,44 @@ export const importBatches = journalSchema.table("import_batches", { export const follows = journalSchema.table("follows", { id: text("id").primaryKey(), - followerId: text("follower_id") - .notNull() - .references(() => users.id, { onDelete: "cascade" }), + // Local follower. NULL for inbound federated follows (a remote actor + // following a local user), where `followerActorIri` identifies the + // follower instead. Exactly one of the two is set (check constraint). + followerId: text("follower_id").references(() => users.id, { onDelete: "cascade" }), + // Remote follower's actor IRI (spec: social-federation, inbound + // Follow). NULL for local-origin follows. + followerActorIri: text("follower_actor_iri"), followedActorIri: text("followed_actor_iri").notNull(), followedUserId: text("followed_user_id").references(() => users.id, { onDelete: "cascade" }), acceptedAt: timestamp("accepted_at", { withTimezone: true }), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }, (t) => ({ followerActorUnique: uniqueIndex("follows_follower_actor_unique").on(t.followerId, t.followedActorIri), + // Dedupe inbound remote follows: one row per (remote follower, local + // followed user). Partial — local-origin rows have a NULL IRI here. + remoteFollowerUnique: uniqueIndex("follows_remote_follower_unique") + .on(t.followerActorIri, t.followedUserId) + .where(sql`${t.followerActorIri} IS NOT NULL`), followerCreatedIdx: index("follows_follower_created_idx").on(t.followerId, t.createdAt.desc()), followedActorIdx: index("follows_followed_actor_idx").on(t.followedActorIri), followedUserIdx: index("follows_followed_user_idx").on(t.followedUserId), + hasFollowerCheck: check( + "follows_has_follower_check", + sql`(${t.followerId} IS NOT NULL) <> (${t.followerActorIri} IS NOT NULL)`, + ), +})); + +// Fedify KvStore backing table (inbox replay protection, remote +// document / key caches). Keys are JSON-serialized KvKey arrays; +// values arbitrary JSON. Expired rows are filtered at read time and +// swept by the `federation-kv-sweep` job. +export const federationKv = journalSchema.table("federation_kv", { + key: text("key").primaryKey(), + value: jsonb("value").notNull(), + expiresAt: timestamp("expires_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), +}, (t) => ({ + expiresAtIdx: index("federation_kv_expires_at_idx").on(t.expiresAt), })); // Cache of remote ActivityPub actors we interact with (spec: