b96bef91a9
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) <noreply@anthropic.com>
25 lines
993 B
TypeScript
25 lines
993 B
TypeScript
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 };
|
|
},
|
|
};
|