From b96bef91a96879c6edac5d26e97a9674c118e7d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ullrich=20Sch=C3=A4fer?= Date: Sun, 7 Jun 2026 12:05:32 +0200 Subject: [PATCH] =?UTF-8?q?feat(journal):=20outbox-poll=20ingestion=20of?= =?UTF-8?q?=20remote=20trails=20activities=20(=C2=A77)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tasks 7.1–7.3, 7.5 (+7.4 pacing; Retry-After-duration backoff noted as remaining). Resolves the activities.owner_id open question. Schema (design decision from the open question): - activities.owner_id nullable + check constraint enforcing exactly one of (owner_id, remote_actor_iri) — the follows pattern again. - remote_published_at carries the origin's publish time for the §8 feed sort; index on remote_actor_iri for the feed join. - Compiler-audited fallout: notification fan-out, the Note object dispatcher, and the activity detail loader explicitly skip/404 remote rows (their canonical page is the origin; feed links there). Every other surface joins users on owner_id and excludes remote rows structurally. Ingestion: - federation-ingest.server.ts: parseOutboxItem targets exactly our own outgoing Create(Note) shape (PropertyValue stats, first-paragraph name, audience from to/cc); unknown items skipped, never fatal. - ingestRemoteActivities: replay-safe via unique remote_origin_iri, conflict-streak early exit (outboxes are newest-first). - pollRemoteActor: signed fetch (Authorized Fetch via a local follower's key), outbox resolution via remote_actors cache with actor-doc fallback (which refreshes the cache), 1 req/5 s per-host pacing, last_polled_at stamping. Network + pacing injectable. Jobs: poll-remote-actor (per-actor; the §4 Accept listener already enqueues it as the first poll) + poll-remote-outboxes (5-min cron sweep over accepted remote follows not polled within the hour). Tests: parse-shape units; integration suite for ingestion provenance, audience→visibility mirroring, replay safety, the DB author invariant, poll flow (actor-doc → collection → page), signer requirement, and due-polling selection. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/journal/app/jobs/notifications-fanout.ts | 3 + apps/journal/app/jobs/poll-remote-actor.ts | 26 ++ apps/journal/app/jobs/poll-remote-outboxes.ts | 25 ++ .../app/lib/federation-delivery.server.ts | 3 +- .../lib/federation-ingest.integration.test.ts | 167 +++++++++++ .../app/lib/federation-ingest.server.test.ts | 80 +++++ .../app/lib/federation-ingest.server.ts | 279 ++++++++++++++++++ apps/journal/app/lib/federation.server.ts | 5 +- .../app/routes/activities.$id.server.ts | 8 +- apps/journal/server.ts | 4 +- openspec/changes/social-federation/design.md | 18 +- openspec/changes/social-federation/tasks.md | 9 +- packages/db/src/schema/journal.ts | 22 +- 13 files changed, 628 insertions(+), 21 deletions(-) create mode 100644 apps/journal/app/jobs/poll-remote-actor.ts create mode 100644 apps/journal/app/jobs/poll-remote-outboxes.ts create mode 100644 apps/journal/app/lib/federation-ingest.integration.test.ts create mode 100644 apps/journal/app/lib/federation-ingest.server.test.ts create mode 100644 apps/journal/app/lib/federation-ingest.server.ts diff --git a/apps/journal/app/jobs/notifications-fanout.ts b/apps/journal/app/jobs/notifications-fanout.ts index b73f5ef..ef96f78 100644 --- a/apps/journal/app/jobs/notifications-fanout.ts +++ b/apps/journal/app/jobs/notifications-fanout.ts @@ -51,6 +51,9 @@ export async function fanout(activityId: string): Promise { logger.warn({ activityId }, "fanout: activity not found, skipping"); return; } + // Remote-ingested rows have no local owner and never fan out locally + // (the users innerJoin already excludes them; this narrows the type). + if (row.ownerId === null) return; // Defense in depth — the create-side guard already filters this, but // recheck here so the job doesn't mistakenly fan out a row that was // later flipped to private/unlisted before the job ran. diff --git a/apps/journal/app/jobs/poll-remote-actor.ts b/apps/journal/app/jobs/poll-remote-actor.ts new file mode 100644 index 0000000..31582c4 --- /dev/null +++ b/apps/journal/app/jobs/poll-remote-actor.ts @@ -0,0 +1,26 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { pollRemoteActor } from "../lib/federation-ingest.server.ts"; +import { logger } from "../lib/logger.server.ts"; + +interface PollPayload { + actorIri?: string; +} + +/** + * Poll one remote trails actor's outbox (spec §7). Enqueued by the + * inbox Accept(Follow) listener (first poll, 7.5) and fanned out by + * the poll-remote-outboxes cron sweep (7.1). + */ +export const pollRemoteActorJob: JobDefinition = { + name: "poll-remote-actor", + retryLimit: 2, + expireInSeconds: 120, + async handler(jobs) { + for (const job of jobs) { + const { actorIri } = (job.data ?? {}) as PollPayload; + if (!actorIri) continue; + const result = await pollRemoteActor(actorIri); + logger.info({ actorIri, result }, "poll-remote-actor"); + } + }, +}; diff --git a/apps/journal/app/jobs/poll-remote-outboxes.ts b/apps/journal/app/jobs/poll-remote-outboxes.ts new file mode 100644 index 0000000..c6cf085 --- /dev/null +++ b/apps/journal/app/jobs/poll-remote-outboxes.ts @@ -0,0 +1,25 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { listActorsDuePolling } from "../lib/federation-ingest.server.ts"; +import { enqueueOptional } from "../lib/boss.server.ts"; +import { logger } from "../lib/logger.server.ts"; + +/** + * Cron sweep (spec 7.1): every 5 minutes, find remote trails actors + * that at least one local user follows (accepted) and that haven't + * been polled within the last hour, and fan out one poll-remote-actor + * job each. Per-host pacing lives in the poll itself. + */ +export const pollRemoteOutboxesJob: JobDefinition = { + name: "poll-remote-outboxes", + cron: "*/5 * * * *", + retryLimit: 1, + expireInSeconds: 60, + async handler() { + const due = await listActorsDuePolling(); + for (const actorIri of due) { + await enqueueOptional("poll-remote-actor", { actorIri }, { source: "poll-remote-outboxes" }); + } + if (due.length > 0) logger.info({ due: due.length }, "poll-remote-outboxes sweep"); + return { due: due.length }; + }, +}; diff --git a/apps/journal/app/lib/federation-delivery.server.ts b/apps/journal/app/lib/federation-delivery.server.ts index 56d0f09..5753bb0 100644 --- a/apps/journal/app/lib/federation-delivery.server.ts +++ b/apps/journal/app/lib/federation-delivery.server.ts @@ -124,10 +124,11 @@ export async function getCachedRemoteActor(actorIri: string): Promise { + beforeAll(() => { + process.env.ORIGIN ??= "http://localhost:3000"; + }); + + afterEach(async () => { + const db = getDb(); + await db.delete(activities).where(like(activities.remoteOriginIri, `${ACTOR}%`)); + await db.delete(remoteActors).where(eq(remoteActors.actorIri, ACTOR)); + for (const id of createdUserIds.splice(0)) { + await db.delete(follows).where(eq(follows.followerId, id)); + await db.delete(users).where(eq(users.id, id)); + } + }); + + it("inserts remote rows with provenance and audience-mirrored visibility", async () => { + const inserted = await ingestRemoteActivities(ACTOR, [parsed(1), parsed(2, "followers-only")]); + expect(inserted).toBe(2); + + const db = getDb(); + const rows = await db + .select() + .from(activities) + .where(like(activities.remoteOriginIri, `${ACTOR}%`)); + expect(rows).toHaveLength(2); + for (const row of rows) { + expect(row.ownerId).toBeNull(); + expect(row.remoteActorIri).toBe(ACTOR); + expect(row.remotePublishedAt).not.toBeNull(); + } + const byIri = new Map(rows.map((r) => [r.remoteOriginIri, r])); + expect(byIri.get(`${ACTOR}/activities/1`)?.visibility).toBe("public"); + expect(byIri.get(`${ACTOR}/activities/2`)?.visibility).toBe("private"); + expect(byIri.get(`${ACTOR}/activities/2`)?.audience).toBe("followers-only"); + }); + + it("is replay-safe: re-ingesting inserts nothing", async () => { + await ingestRemoteActivities(ACTOR, [parsed(1), parsed(2)]); + const again = await ingestRemoteActivities(ACTOR, [parsed(1), parsed(2)]); + expect(again).toBe(0); + }); + + it("database enforces exactly-one-author invariant", async () => { + const db = getDb(); + await expect( + db.insert(activities).values({ + id: randomUUID(), + ownerId: null, + remoteActorIri: null, + name: "orphan", + }), + ).rejects.toThrow(); + }); + + it("pollRemoteActor resolves the outbox via the actor doc, ingests the first page, stamps last_polled_at", async () => { + await makeFollowerOf(ACTOR); + const fetched: string[] = []; + const result = await pollRemoteActor(ACTOR, { + async pace() {}, + async fetchJson(url) { + fetched.push(url); + if (url === ACTOR) { + return { outbox: `${ACTOR}/outbox`, inbox: `${ACTOR}/inbox`, preferredUsername: "alice", name: "Alice" }; + } + if (url === `${ACTOR}/outbox`) { + return { type: "OrderedCollection", totalItems: 1, first: `${ACTOR}/outbox?cursor=0` }; + } + return { + type: "OrderedCollectionPage", + orderedItems: [ + { + type: "Create", + object: { + type: "Note", + id: `${ACTOR}/activities/77`, + content: "

Polled ride

", + published: "2026-06-05T10:00:00Z", + to: "as:Public", + }, + }, + ], + }; + }, + }); + expect(result).toEqual({ inserted: 1 }); + expect(fetched).toEqual([ACTOR, `${ACTOR}/outbox`, `${ACTOR}/outbox?cursor=0`]); + + const db = getDb(); + const [cached] = await db.select().from(remoteActors).where(eq(remoteActors.actorIri, ACTOR)); + expect(cached?.outboxUrl).toBe(`${ACTOR}/outbox`); + expect(cached?.lastPolledAt).not.toBeNull(); + }); + + it("skips actors without an accepted local follower", async () => { + const result = await pollRemoteActor(ACTOR, { + async pace() {}, + async fetchJson() { + throw new Error("must not fetch"); + }, + }); + expect(result).toEqual({ skipped: "no accepted local follower" }); + }); + + it("listActorsDuePolling returns followed actors not polled within the hour", async () => { + await makeFollowerOf(ACTOR); + expect(await listActorsDuePolling()).toContain(ACTOR); + + const db = getDb(); + await db.insert(remoteActors).values({ actorIri: ACTOR, lastPolledAt: new Date() }).onConflictDoUpdate({ + target: remoteActors.actorIri, + set: { lastPolledAt: new Date() }, + }); + expect(await listActorsDuePolling()).not.toContain(ACTOR); + }); +}); diff --git a/apps/journal/app/lib/federation-ingest.server.test.ts b/apps/journal/app/lib/federation-ingest.server.test.ts new file mode 100644 index 0000000..8167e53 --- /dev/null +++ b/apps/journal/app/lib/federation-ingest.server.test.ts @@ -0,0 +1,80 @@ +import { describe, it, expect, vi } from "vitest"; + +vi.mock("./db.ts", () => ({ getDb: () => ({}) })); + +const { parseOutboxItem } = await import("./federation-ingest.server.ts"); + +function trailsCreate(overrides: Record = {}, noteOverrides: Record = {}) { + return { + type: "Create", + id: "https://remote.example/activities/a1#create", + published: "2026-06-01T08:00:00Z", + object: { + type: "Note", + id: "https://remote.example/activities/a1", + content: "

Morning ride

\n

42.2 km · ↗ 512 m

\n

link

", + published: "2026-06-01T08:00:00Z", + to: "as:Public", + attachment: [ + { type: "PropertyValue", name: "distance-m", value: "42195" }, + { type: "PropertyValue", name: "elevation-gain-m", value: "512" }, + { type: "PropertyValue", name: "duration-s", value: "9000" }, + ], + ...noteOverrides, + }, + ...overrides, + }; +} + +describe("parseOutboxItem", () => { + it("parses our own outgoing shape", () => { + const parsed = parseOutboxItem(trailsCreate()); + expect(parsed).toEqual({ + originIri: "https://remote.example/activities/a1", + name: "Morning ride", + distance: 42195, + elevationGain: 512, + duration: 9000, + publishedAt: new Date("2026-06-01T08:00:00Z"), + audience: "public", + }); + }); + + it("detects all spellings of the public collection", () => { + for (const to of [ + "https://www.w3.org/ns/activitystreams#Public", + "as:Public", + ["as:Public", "https://remote.example/followers"], + ]) { + expect(parseOutboxItem(trailsCreate({}, { to }))?.audience).toBe("public"); + } + expect(parseOutboxItem(trailsCreate({}, { to: "https://remote.example/followers" }))?.audience).toBe( + "followers-only", + ); + // cc counts too + expect( + parseOutboxItem(trailsCreate({}, { to: undefined, cc: "as:Public" }))?.audience, + ).toBe("public"); + }); + + it("tolerates missing stats and published", () => { + const parsed = parseOutboxItem(trailsCreate({ published: undefined }, { attachment: undefined, published: undefined })); + expect(parsed).toMatchObject({ distance: null, elevationGain: null, duration: null, publishedAt: null }); + }); + + it("skips anything that isn't Create(Note) with an id and a name", () => { + expect(parseOutboxItem(null)).toBeNull(); + expect(parseOutboxItem("string")).toBeNull(); + expect(parseOutboxItem({ type: "Announce" })).toBeNull(); + expect(parseOutboxItem(trailsCreate({}, { type: "Article" }))).toBeNull(); + expect(parseOutboxItem(trailsCreate({}, { id: undefined }))).toBeNull(); + expect(parseOutboxItem(trailsCreate({}, { content: "" }))).toBeNull(); + }); + + it("strips markup from the name and caps its length", () => { + const longName = "x".repeat(400); + const parsed = parseOutboxItem(trailsCreate({}, { content: `

${longName}

` })); + expect(parsed?.name).toHaveLength(300); + expect(parsed?.name).not.toContain(""); + }); +}); diff --git a/apps/journal/app/lib/federation-ingest.server.ts b/apps/journal/app/lib/federation-ingest.server.ts new file mode 100644 index 0000000..ae7fa5d --- /dev/null +++ b/apps/journal/app/lib/federation-ingest.server.ts @@ -0,0 +1,279 @@ +// Outbox-poll ingestion of remote trails activities (spec: +// social-federation §7). We are the poller: signed GETs (Authorized +// Fetch) against the outboxes of remote trails actors that local users +// follow, storing new activities for feed display. The wire shape is +// our own outbox format (trails-to-trails only), so parsing targets +// exactly what federation-objects.server.ts emits: Create activities +// wrapping Notes with PropertyValue stat attachments. +// +// Network steps are injectable for offline integration tests. + +import { and, eq, isNull, isNotNull, lt, or, sql } from "drizzle-orm"; +import { randomUUID } from "node:crypto"; +import { activities, follows, remoteActors, users } from "@trails-cool/db/schema/journal"; +import { getDb } from "./db.ts"; +import { getOrigin } from "./config.server.ts"; +import { getFederation } from "./federation.server.ts"; +import { upsertRemoteActor } from "./federation-delivery.server.ts"; +import { logger } from "./logger.server.ts"; + +/** Spec: fetch at most the 50 most recent items per remote actor. */ +const MAX_ITEMS_PER_POLL = 50; +/** Spec: skip actors polled within the last hour (cron sweep). */ +const POLL_INTERVAL_MS = 60 * 60 * 1000; +/** Spec: per-remote-host pacing of 1 request / 5 seconds. */ +const HOST_PACE_MS = 5_000; +/** Stop early after this many consecutive already-seen items (7.2). */ +const CONFLICT_STREAK_LIMIT = 5; + +export interface ParsedRemoteActivity { + originIri: string; + name: string; + distance: number | null; + elevationGain: number | null; + duration: number | null; + publishedAt: Date | null; + audience: "public" | "followers-only"; +} + +function isPublicAudience(value: unknown): boolean { + const targets = Array.isArray(value) ? value : value == null ? [] : [value]; + return targets.some( + (t) => + t === "https://www.w3.org/ns/activitystreams#Public" || + t === "as:Public" || + t === "Public", + ); +} + +function textOfFirstParagraph(html: string): string { + const match = /

([\s\S]*?)<\/p>/.exec(html); + const raw = match?.[1] ?? html; + return raw.replace(/<[^>]+>/g, "").trim(); +} + +function statFromAttachments(attachments: unknown, name: string): number | null { + const list = Array.isArray(attachments) ? attachments : attachments == null ? [] : [attachments]; + for (const a of list) { + if ( + typeof a === "object" && a !== null && + (a as Record).type === "PropertyValue" && + (a as Record).name === name + ) { + const v = Number.parseFloat(String((a as Record).value)); + return Number.isFinite(v) ? v : null; + } + } + return null; +} + +/** + * Parse one outbox item (a `Create` wrapping a `Note` in our own + * outgoing shape) into an ingestable activity. Returns null for + * anything that doesn't match — unknown items are skipped, never + * fatal (forward compatibility with future trails versions). + */ +export function parseOutboxItem(item: unknown): ParsedRemoteActivity | null { + if (typeof item !== "object" || item === null) return null; + const create = item as Record; + if (create.type !== "Create") return null; + const note = create.object; + if (typeof note !== "object" || note === null) return null; + const n = note as Record; + if (n.type !== "Note" || typeof n.id !== "string") return null; + + const content = typeof n.content === "string" ? n.content : ""; + const name = textOfFirstParagraph(content); + if (!name) return null; + + const publishedRaw = typeof n.published === "string" ? n.published : typeof create.published === "string" ? create.published : null; + const published = publishedRaw ? new Date(publishedRaw) : null; + + return { + originIri: n.id, + name: name.slice(0, 300), + distance: statFromAttachments(n.attachment, "distance-m"), + elevationGain: statFromAttachments(n.attachment, "elevation-gain-m"), + duration: statFromAttachments(n.attachment, "duration-s"), + publishedAt: published && !Number.isNaN(published.getTime()) ? published : null, + audience: isPublicAudience(n.to) || isPublicAudience(n.cc) ? "public" : "followers-only", + }; +} + +/** + * Insert parsed activities for a remote actor. Replay-safe via the + * unique remote_origin_iri (`ON CONFLICT DO NOTHING`); stops early on + * a streak of already-seen items since outboxes are newest-first. + * Returns the number of newly inserted rows. + */ +export async function ingestRemoteActivities( + actorIri: string, + items: ParsedRemoteActivity[], +): Promise { + const db = getDb(); + let inserted = 0; + let conflictStreak = 0; + for (const item of items) { + const rows = await db + .insert(activities) + .values({ + id: randomUUID(), + ownerId: null, + name: item.name, + description: "", + distance: item.distance, + elevationGain: item.elevationGain, + duration: item.duration, + // Mirror the audience into visibility for coherence with + // local rows; the §8 feed query gates remote rows on audience + // + follow, and every other surface joins users on owner_id, + // which excludes remote rows structurally. + visibility: item.audience === "public" ? "public" : "private", + remoteOriginIri: item.originIri, + remoteActorIri: actorIri, + remotePublishedAt: item.publishedAt, + audience: item.audience, + }) + .onConflictDoNothing({ target: activities.remoteOriginIri }) + .returning({ id: activities.id }); + if (rows.length === 0) { + conflictStreak++; + if (conflictStreak >= CONFLICT_STREAK_LIMIT) break; + } else { + conflictStreak = 0; + inserted++; + } + } + return inserted; +} + +export interface PollDeps { + /** + * Fetch a JSON document with an HTTP Signature from `signerUsername` + * (Authorized Fetch — spec: "Polls are signed"). + */ + fetchJson(url: string, signerUsername: string): Promise; + /** Per-host pacing (spec 7.4: 1 req / 5 s). No-op in tests. */ + pace(host: string): Promise; +} + +const lastFetchPerHost = new Map(); + +function defaultDeps(): PollDeps { + return { + async fetchJson(url, signerUsername) { + const federation = getFederation(); + const ctx = federation.createContext(new URL(getOrigin()), undefined); + const loader = await ctx.getDocumentLoader({ identifier: signerUsername }); + const { document } = await loader(url); + return document; + }, + async pace(host) { + const last = lastFetchPerHost.get(host) ?? 0; + const wait = last + HOST_PACE_MS - Date.now(); + if (wait > 0) await new Promise((r) => setTimeout(r, wait)); + lastFetchPerHost.set(host, Date.now()); + }, + }; +} + +/** + * Poll one remote actor's outbox and ingest new activities (7.2/7.3). + * Returns counts for logging. Honors per-host pacing; a 429/Retry-After + * from the remote aborts this poll quietly (the hourly sweep retries). + */ +export async function pollRemoteActor( + actorIri: string, + deps: PollDeps = defaultDeps(), +): Promise<{ inserted: number } | { skipped: string }> { + const db = getDb(); + + // Signer: any local user with an accepted follow against this actor. + const [signer] = await db + .select({ username: users.username }) + .from(follows) + .innerJoin(users, eq(follows.followerId, users.id)) + .where( + and( + eq(follows.followedActorIri, actorIri), + isNotNull(follows.acceptedAt), + isNull(follows.followedUserId), + ), + ) + .limit(1); + if (!signer) return { skipped: "no accepted local follower" }; + + const host = new URL(actorIri).host; + try { + await deps.pace(host); + + // Resolve the outbox URL: cache first, actor document as fallback + // (which also refreshes the cache — 7.3). + const [cached] = await db + .select({ outboxUrl: remoteActors.outboxUrl }) + .from(remoteActors) + .where(eq(remoteActors.actorIri, actorIri)) + .limit(1); + let outboxUrl = cached?.outboxUrl ?? null; + if (!outboxUrl) { + const actorDoc = (await deps.fetchJson(actorIri, signer.username)) as Record | null; + outboxUrl = typeof actorDoc?.outbox === "string" ? actorDoc.outbox : null; + if (!outboxUrl) return { skipped: "actor has no outbox" }; + await upsertRemoteActor({ + actorIri, + outboxUrl, + displayName: typeof actorDoc?.name === "string" ? actorDoc.name : null, + username: typeof actorDoc?.preferredUsername === "string" ? actorDoc.preferredUsername : null, + inboxUrl: typeof actorDoc?.inbox === "string" ? actorDoc.inbox : undefined, + domain: host, + }); + } + + // Collection → first page (our outbox serves first=...?cursor=0). + await deps.pace(host); + const collection = (await deps.fetchJson(outboxUrl, signer.username)) as Record; + let itemsRaw = collection.orderedItems ?? collection.items; + if (!itemsRaw && typeof collection.first === "string") { + await deps.pace(host); + const page = (await deps.fetchJson(collection.first, signer.username)) as Record; + itemsRaw = page.orderedItems ?? page.items; + } + const items = (Array.isArray(itemsRaw) ? itemsRaw : itemsRaw == null ? [] : [itemsRaw]) + .slice(0, MAX_ITEMS_PER_POLL) + .map(parseOutboxItem) + .filter((p): p is ParsedRemoteActivity => p !== null); + + const inserted = await ingestRemoteActivities(actorIri, items); + await db + .update(remoteActors) + .set({ lastPolledAt: new Date() }) + .where(eq(remoteActors.actorIri, actorIri)); + logger.info({ actorIri, fetched: items.length, inserted }, "federation: outbox poll"); + return { inserted }; + } catch (err) { + // 429 / network failures: log and let the hourly sweep retry. + logger.warn({ err, actorIri }, "federation: outbox poll failed; will retry on next sweep"); + return { skipped: "fetch failed" }; + } +} + +/** + * Remote actor IRIs due for polling (7.1): followed-and-accepted by at + * least one local user, never polled or polled more than an hour ago. + */ +export async function listActorsDuePolling(): Promise { + const db = getDb(); + const cutoff = new Date(Date.now() - POLL_INTERVAL_MS); + const rows = await db + .selectDistinct({ actorIri: follows.followedActorIri }) + .from(follows) + .leftJoin(remoteActors, eq(follows.followedActorIri, remoteActors.actorIri)) + .where( + and( + isNull(follows.followedUserId), + isNotNull(follows.acceptedAt), + or(sql`${remoteActors.lastPolledAt} IS NULL`, lt(remoteActors.lastPolledAt, cutoff)), + ), + ); + return rows.map((r) => r.actorIri); +} diff --git a/apps/journal/app/lib/federation.server.ts b/apps/journal/app/lib/federation.server.ts index 530e5d0..fa25f68 100644 --- a/apps/journal/app/lib/federation.server.ts +++ b/apps/journal/app/lib/federation.server.ts @@ -311,7 +311,10 @@ function buildFederation(): Federation { .from(activities) .where(and(eq(activities.id, values.id), eq(activities.visibility, "public"))) .limit(1); - if (!row) return null; + // Remote-ingested activities are not our objects — only locally + // authored rows dereference here (their canonical IRI is on the + // origin instance). + if (!row || row.ownerId === null) return null; const [owner] = await db .select({ username: users.username, profileVisibility: users.profileVisibility }) .from(users) diff --git a/apps/journal/app/routes/activities.$id.server.ts b/apps/journal/app/routes/activities.$id.server.ts index b06578f..d75261a 100644 --- a/apps/journal/app/routes/activities.$id.server.ts +++ b/apps/journal/app/routes/activities.$id.server.ts @@ -17,8 +17,12 @@ import type { Visibility } from "@trails-cool/db/schema/journal"; const VISIBILITY_VALUES = new Set(["private", "unlisted", "public"]); export async function loadActivityDetail(request: Request, id: string | undefined) { - const activity = await getActivity(id ?? ""); - if (!activity) throw data({ error: "Activity not found" }, { status: 404 }); + const fetched = await getActivity(id ?? ""); + if (!fetched) throw data({ error: "Activity not found" }, { status: 404 }); + // Remote-ingested activities have no local detail page — their + // canonical page lives on the origin instance; the feed links there. + if (fetched.ownerId === null) throw data({ error: "Activity not found" }, { status: 404 }); + const activity = { ...fetched, ownerId: fetched.ownerId }; const user = await getSessionUser(request); const isOwner = user?.id === activity.ownerId; diff --git a/apps/journal/server.ts b/apps/journal/server.ts index aad8035..93342ac 100644 --- a/apps/journal/server.ts +++ b/apps/journal/server.ts @@ -183,7 +183,9 @@ server.listen(port, async () => { const { backfillUserKeypairsJob } = await import("./app/jobs/backfill-user-keypairs.ts"); const { federationKvSweepJob } = await import("./app/jobs/federation-kv-sweep.ts"); const { deliverActivityJob } = await import("./app/jobs/deliver-activity.ts"); - jobs.push(backfillUserKeypairsJob, federationKvSweepJob, deliverActivityJob); + const { pollRemoteActorJob } = await import("./app/jobs/poll-remote-actor.ts"); + const { pollRemoteOutboxesJob } = await import("./app/jobs/poll-remote-outboxes.ts"); + jobs.push(backfillUserKeypairsJob, federationKvSweepJob, deliverActivityJob, pollRemoteActorJob, pollRemoteOutboxesJob); } const boss = createBoss(getDatabaseUrl()); diff --git a/openspec/changes/social-federation/design.md b/openspec/changes/social-federation/design.md index a104644..0081bd4 100644 --- a/openspec/changes/social-federation/design.md +++ b/openspec/changes/social-federation/design.md @@ -147,13 +147,17 @@ Inbound signature verification uses the actor's public key from their actor obje ## Open Questions -- **`activities.owner_id` is NOT NULL but remote-ingested rows have no local - owner** (surfaced during task 2.3, 2026-06-06). Options: make `owner_id` - nullable with a check constraint (`owner_id IS NOT NULL OR remote_actor_iri - IS NOT NULL`), or key remote rows purely off `remote_actor_iri` in a way - that never touches owner-joined queries. Decide in task 7.2 (ingestion) - before any remote row is written; the columns landed in 2.3 don't prejudge - either option. +- ~~**`activities.owner_id` is NOT NULL but remote-ingested rows have no local + owner**~~ — **Resolved in task 7.2 (2026-06-07):** `owner_id` is nullable + with a check constraint enforcing exactly one of (`owner_id`, + `remote_actor_iri`) — the same pattern as `follows.follower_id` / + `follower_actor_iri`. Compiler-audited fallout: notification fan-out, + the Note object dispatcher, and the activity detail loader all + explicitly 404/skip remote rows (their canonical page is the origin + instance; the feed links outward). Every other surface joins `users` + on `owner_id` and excludes remote rows structurally. A + `remote_published_at` column carries the origin's publish time for + the §8 feed sort. - Custom AS extension for activity-type vs. plain `Create(Note)`? Mastodon will only render Notes; an extension type means non-trails clients see nothing useful. Lean toward `Create(Note)` with structured metadata in `attachment` so Mastodon shows the text + GPX link, and trails clients consuming the outbox can read the structured fields. - Per-instance inbox vs per-user inbox? Mastodon supports a "shared inbox" optimization. Worth doing for delivery efficiency once we have any volume; v1 can use per-user inboxes only. diff --git a/openspec/changes/social-federation/tasks.md b/openspec/changes/social-federation/tasks.md index 3684b5b..16a6d92 100644 --- a/openspec/changes/social-federation/tasks.md +++ b/openspec/changes/social-federation/tasks.md @@ -57,11 +57,12 @@ ## 7. Outbox-poll ingestion -- [ ] 7.1 pg-boss recurring job `poll-remote-outboxes` (cron every 5 min): iterate distinct accepted remote actor IRIs in `follows` where `remote_actors.last_polled_at < now() - interval '1 hour'`; for each, signed GET on the outbox using one of the local accepted-followers' keys -- [ ] 7.2 Insert returned activities into `journal.activities` with `remote_origin_iri`, `remote_actor_iri`, `audience`. `ON CONFLICT (remote_origin_iri) DO NOTHING` for replay safety. Stop early on streak of conflicts -- [ ] 7.3 Refresh `remote_actors` cache row on each poll (display name, avatar, outbox URL, public key) +- [x] 7.1 pg-boss recurring job `poll-remote-outboxes` (cron every 5 min): iterate distinct accepted remote actor IRIs in `follows` where `remote_actors.last_polled_at < now() - interval '1 hour'`; for each, signed GET on the outbox using one of the local accepted-followers' keys +- [x] 7.2 Insert returned activities into `journal.activities` with `remote_origin_iri`, `remote_actor_iri`, `audience`. `ON CONFLICT (remote_origin_iri) DO NOTHING` for replay safety. Stop early on streak of conflicts +- [x] 7.3 Refresh `remote_actors` cache row on each poll (display name, avatar, outbox URL, public key) - [ ] 7.4 Per-remote-host rate limit (1 req / 5 sec); honor `Retry-After` / `429`; back off the host -- [ ] 7.5 First-poll trigger on `accepted_at` transition (already enqueued from 4.4) + > Partially done (2026-06-07): 1 req/5 s pacing implemented; any fetch failure (incl. 429) skips the host until the next 5-min sweep. Explicit Retry-After-duration backoff remains — needs header access through Fedify's document loader errors. +- [x] 7.5 First-poll trigger on `accepted_at` transition (already enqueued from 4.4) ## 8. Feed query update diff --git a/packages/db/src/schema/journal.ts b/packages/db/src/schema/journal.ts index 3ed696e..7bf9bce 100644 --- a/packages/db/src/schema/journal.ts +++ b/packages/db/src/schema/journal.ts @@ -144,9 +144,12 @@ export type Audience = "public" | "followers-only"; export const activities = journalSchema.table("activities", { id: text("id").primaryKey(), - ownerId: text("owner_id") - .notNull() - .references(() => users.id), + // Local author. NULL for activities ingested from a remote trails + // actor's outbox, where `remoteActorIri` identifies the author — + // exactly one of the two is set (check constraint below; same + // pattern as follows.follower_id / follower_actor_iri). Resolves + // the owner_id open question in social-federation design.md. + ownerId: text("owner_id").references(() => users.id), routeId: text("route_id").references(() => routes.id), name: text("name").notNull(), description: text("description").default(""), @@ -164,16 +167,25 @@ export const activities = journalSchema.table("activities", { // Federation provenance (spec: social-federation). NULL for local // activities. For rows ingested from a remote trails actor's outbox: // `remoteOriginIri` is the activity's IRI on the origin instance - // (unique → replay-safe `ON CONFLICT DO NOTHING` ingestion), and - // `remoteActorIri` keys into `remote_actors`. + // (unique → replay-safe `ON CONFLICT DO NOTHING` ingestion), + // `remoteActorIri` keys into `remote_actors`, and + // `remotePublishedAt` carries the origin's publish time (feed sort + // uses COALESCE(remote_published_at, created_at)). remoteOriginIri: text("remote_origin_iri").unique(), remoteActorIri: text("remote_actor_iri"), + remotePublishedAt: timestamp("remote_published_at", { withTimezone: true }), audience: text("audience").$type().notNull().default("public"), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }, (t) => ({ // Hot paths: listActivities (sort by started_at) and listPublicActivitiesForOwner. ownerStartedIdx: index("activities_owner_started_idx").on(t.ownerId, t.startedAt.desc()), ownerCreatedIdx: index("activities_owner_created_idx").on(t.ownerId, t.createdAt.desc()), + // Feed join for remote rows (spec §8). + remoteActorIdx: index("activities_remote_actor_idx").on(t.remoteActorIri), + hasAuthorCheck: check( + "activities_has_author_check", + sql`(${t.ownerId} IS NOT NULL) <> (${t.remoteActorIri} IS NOT NULL)`, + ), })); // --- OAuth2 PKCE (mobile app auth) ---