Merge pull request #486 from trails-cool/federation/outbox-polling

feat(journal): outbox-poll ingestion of remote trails activities (§7)

apps/journal/app/jobs/notifications-fanout.ts

@@ -51,6 +51,9 @@ export async function fanout(activityId: string): Promise<void> {
     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.

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");
+    }
+  },
+};

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 };
+  },
+};

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

@@ -124,10 +124,11 @@ export async function getCachedRemoteActor(actorIri: string): Promise<CachedRemo
   return row ?? null;
 }
 
-/** Upsert the fields delivery learns about a remote actor. */
+/** Upsert the fields delivery/polling learn about a remote actor. */
 export async function upsertRemoteActor(fields: {
   actorIri: string;
   inboxUrl?: string | null;
+  outboxUrl?: string | null;
   displayName?: string | null;
   username?: string | null;
   domain?: string | null;

apps/journal/app/lib/federation-ingest.integration.test.ts

@@ -0,0 +1,167 @@
+import { describe, it, expect, beforeAll, afterEach } from "vitest";
+import { eq, like } from "drizzle-orm";
+import { randomUUID } from "node:crypto";
+import { getDb } from "./db.ts";
+import { users, follows, activities, remoteActors } from "@trails-cool/db/schema/journal";
+import {
+  ingestRemoteActivities,
+  pollRemoteActor,
+  listActorsDuePolling,
+  type ParsedRemoteActivity,
+} from "./federation-ingest.server.ts";
+
+// Opt-in: real Postgres; network injected. Same convention as the
+// other *.integration.test.ts files.
+const runIntegration = process.env.FEDERATION_INTEGRATION === "1";
+
+const ACTOR = `https://poll-target.example/users/alice-${Date.now()}`;
+const createdUserIds: string[] = [];
+
+async function makeFollowerOf(actorIri: string) {
+  const db = getDb();
+  const id = randomUUID();
+  await db.insert(users).values({
+    id,
+    email: `poll-${id}@example.test`,
+    username: `poll-${id.slice(0, 8)}`,
+    domain: "test.local",
+    profileVisibility: "public",
+  });
+  createdUserIds.push(id);
+  await db.insert(follows).values({
+    id: randomUUID(),
+    followerId: id,
+    followedActorIri: actorIri,
+    followedUserId: null,
+    acceptedAt: new Date(),
+  });
+  return id;
+}
+
+function parsed(n: number, audience: "public" | "followers-only" = "public"): ParsedRemoteActivity {
+  return {
+    originIri: `${ACTOR}/activities/${n}`,
+    name: `Remote activity ${n}`,
+    distance: 1000 * n,
+    elevationGain: null,
+    duration: null,
+    publishedAt: new Date(Date.now() - n * 60_000),
+    audience,
+  };
+}
+
+describe.runIf(runIntegration)("outbox-poll ingestion (integration)", () => {
+  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: "<p>Polled ride</p>",
+                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);
+  });
+});

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<string, unknown> = {}, noteOverrides: Record<string, unknown> = {}) {
+  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: "<p>Morning ride</p>\n<p>42.2 km · ↗ 512 m</p>\n<p><a href=\"https://remote.example/activities/a1\">link</a></p>",
+      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: `<p><b>${longName}</b></p>` }));
+    expect(parsed?.name).toHaveLength(300);
+    expect(parsed?.name).not.toContain("<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 = /<p>([\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<string, unknown>).type === "PropertyValue" &&
+      (a as Record<string, unknown>).name === name
+    ) {
+      const v = Number.parseFloat(String((a as Record<string, unknown>).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<string, unknown>;
+  if (create.type !== "Create") return null;
+  const note = create.object;
+  if (typeof note !== "object" || note === null) return null;
+  const n = note as Record<string, unknown>;
+  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<number> {
+  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<unknown>;
+  /** Per-host pacing (spec 7.4: 1 req / 5 s). No-op in tests. */
+  pace(host: string): Promise<void>;
+}
+
+const lastFetchPerHost = new Map<string, number>();
+
+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<string, unknown> | 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<string, unknown>;
+    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<string, unknown>;
+      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<string[]> {
+  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);
+}

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

@@ -311,7 +311,10 @@ function buildFederation(): Federation<void> {
       .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)

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<Visibility>(["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;

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());

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.

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
 

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<Audience>().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) ---