diff --git a/.changeset/feed-targets-shape.md b/.changeset/feed-targets-shape.md new file mode 100644 index 0000000..357f1bf --- /dev/null +++ b/.changeset/feed-targets-shape.md @@ -0,0 +1,5 @@ +--- +"@atmo-dev/contrail-lexicons": patch +--- + +Handle the new `FeedConfig.targets` shape (`string | { collection, maxItems? }`) when generating the feed lexicon and computing pull NSIDs, and fall back to the default `"follow"` short name when `FeedConfig.follow` is unset. diff --git a/.changeset/follow-feed-overhaul.md b/.changeset/follow-feed-overhaul.md new file mode 100644 index 0000000..295c44f --- /dev/null +++ b/.changeset/follow-feed-overhaul.md @@ -0,0 +1,23 @@ +--- +"@atmo-dev/contrail": minor +--- + +Follow-feed overhaul. Several related changes that together fix correctness and storage problems with how follow-driven feeds are bootstrapped, ingested, and recovered. + +**Backfill correctness — `time_us` now reflects record `createdAt`.** Backfilled records previously had `time_us` set to ingest time, which silently broke any time-ordered query and made `feed_items` snapshots taken right after a backfill useless. The canonical time is parsed from the record's `createdAt` (clamped to now to defuse user-supplied future timestamps) and used as `time_us`. Per-collection override via the new `CollectionConfig.timeField` (set to `false` to keep ingest time, e.g. for collections without a time field). + +**`feed_backfills.completed` no longer falsely marks success.** The wrapper used to mark `completed = 1` even when the underlying follow walk timed out or returned zero, locking users into a permanently empty feed. The flag now flips only after `backfills.completed = 1` is observed for the follow collection. New `retries`, `last_error`, and `started_at` columns mirror the existing `backfills` schema and let stuck rows be re-armed after `BACKFILL_STALE_MS`. + +**Feed bootstrap moved out of the request path.** `getFeed` no longer blocks on a synchronous PDS walk. Instead it claims the `feed_backfills` row and schedules `runFeedBackfill` via `c.executionCtx.waitUntil` (Cloudflare Workers) or fire-and-forget on Node/Bun. First request returns whatever `feed_items` already has; subsequent requests reflect the full backfill once it lands. Live fanout (which adds a new follow's last 100 posts on the spot) makes the empty first response uncommon in practice for already-active users. + +**Per-target item caps.** `FeedConfig.targets` now accepts `string | { collection, maxItems? }`, and pruning partitions by `(actor, collection)` so a high-volume target (e.g. RSVPs) can't squeeze a low-volume one (e.g. events) out of the cap. `pruneFeedItems` accepts either a global cap (legacy) or `Map`; jetstream/persistent ingest cycles now compute the per-collection map via `buildFeedTargetCaps`. + +**Subject filter for follow ingest + backfill.** New `CollectionConfig.subjectField` — when set, ingest drops records whose subject DID isn't already in `identities`. For a typical bsky user with 2k follows but only 10 pointing at known DIDs, this trims storage by ~200x. Applied identically in live jetstream filtering and per-page during backfill. + +**`app.bsky.*` defaults to `discover: false`.** Any collection whose NSID lives under `app.bsky.*` and doesn't explicitly set `discover` is treated as dependent — preventing a footgun where forgetting `discover: false` on `app.bsky.graph.follow` would persist every follow on the network. + +**Auto-add follow collection.** `FeedConfig.follow` is now optional and defaults to `"follow"` (auto-added with NSID `app.bsky.graph.follow`, `discover: false`, and `subjectField: "subject"`) when no feed declares it. `feeds: { home: { targets: ["post"] } }` now produces correct behavior with no explicit follow plumbing. + +**Constellation reverse-lookup (opt-out, default on).** When a DID first appears in `identities` via a discoverable event, contrail queries [Constellation](https://constellation.microcosm.blue/) for follow records pointing at that DID and ingests synthesized rows for any follower already in `identities`. Lets newcomers immediately surface in existing users' feeds without per-follower PDS walks. Disable with `constellation: false` or `constellation: { enabled: false }`. Sends `User-Agent: contrail/` per Constellation's request that callers identify themselves. + +**Wire-level `collection` param accepts NSIDs.** `getFeed` now matches the generated lexicon enum: the `collection` parameter is interpreted as a full NSID and translated to the short name internally. Short names are still tolerated for backwards compatibility. diff --git a/packages/contrail/src/core/backfill.ts b/packages/contrail/src/core/backfill.ts index 3d23dc1..559131a 100644 --- a/packages/contrail/src/core/backfill.ts +++ b/packages/contrail/src/core/backfill.ts @@ -3,10 +3,42 @@ import { isDid, isNsid } from "@atcute/lexicons/syntax"; import type { Client } from "@atcute/client"; import type { ContrailConfig, Database, IngestEvent } from "./types"; -import { getDiscoverableNsids, getDependentNsids, DEFAULT_RELAYS } from "./types"; +import { + getDiscoverableNsids, + getDependentNsids, + DEFAULT_RELAYS, + shortNameForNsid, +} from "./types"; import { applyEvents, getLastCursor, saveCursor } from "./db"; import { getClient, getPDS } from "./client"; +const DEFAULT_TIME_FIELD = "createdAt"; + +/** Parse the record's canonical time (e.g. createdAt) and return microseconds. + * Falls back to `nowUs` when missing/invalid. Clamps to nowUs to avoid + * user-controlled future timestamps pinning records at the top of feeds. */ +function recordTimeUs( + record: unknown, + collection: string, + config: ContrailConfig | undefined, + nowUs: number +): number { + if (!config) return nowUs; + const short = shortNameForNsid(config, collection); + const colCfg = short ? config.collections[short] : undefined; + const field = colCfg?.timeField ?? DEFAULT_TIME_FIELD; + if (field === false) return nowUs; + const raw = + record && typeof record === "object" + ? (record as Record)[field] + : undefined; + if (typeof raw !== "string") return nowUs; + const ms = Date.parse(raw); + if (!Number.isFinite(ms) || ms <= 0) return nowUs; + const us = ms * 1000; + return us > nowUs ? nowUs : us; +} + const PAGE_SIZE = 100; const BATCH_SIZE = 100; const MAX_RETRIES = 5; @@ -39,6 +71,49 @@ async function withRetry( throw lastError; } +/** Drop events whose `subjectField` value is a DID we have no identity for. + * One bulk SELECT per call, suitable for use after each backfill page. */ +async function filterEventsBySubject( + db: Database, + events: IngestEvent[], + subjectField: string +): Promise { + const subjects = new Set(); + const eventSubjects = new Map(); + for (const e of events) { + if (!e.record) continue; + let subj: unknown; + try { + subj = JSON.parse(e.record)?.[subjectField]; + } catch { + continue; + } + if (typeof subj === "string" && isDid(subj)) { + subjects.add(subj); + eventSubjects.set(e.uri, subj); + } + } + if (subjects.size === 0) return []; + + const known = new Set(); + const list = [...subjects]; + const CHUNK = 100; + for (let i = 0; i < list.length; i += CHUNK) { + const chunk = list.slice(i, i + CHUNK); + const placeholders = chunk.map(() => "?").join(","); + const rows = await db + .prepare(`SELECT did FROM identities WHERE did IN (${placeholders})`) + .bind(...chunk) + .all<{ did: string }>(); + for (const r of rows.results ?? []) known.add(r.did); + } + + return events.filter((e) => { + const subj = eventSubjects.get(e.uri); + return subj !== undefined && known.has(subj); + }); +} + async function markFailed( db: Database, did: string, @@ -124,6 +199,15 @@ export async function backfillUser( let totalInserted = 0; let done = false; + // Lookup subject filter once: if this collection declares a subjectField, we + // drop records whose subject DID isn't already in our identities table. + const collectionShort = config + ? shortNameForNsid(config, collection) + : undefined; + const subjectField = collectionShort + ? config?.collections[collectionShort]?.subjectField + : undefined; + try { while (Date.now() < deadline) { const response = await withRetry( @@ -156,7 +240,8 @@ export async function backfillUser( } const now = Date.now(); - const events: IngestEvent[] = response.data.records.map((r) => ({ + const nowUs = now * 1000; + let events: IngestEvent[] = response.data.records.map((r) => ({ uri: r.uri, did, collection, @@ -164,14 +249,20 @@ export async function backfillUser( operation: "create" as const, cid: r.cid, record: JSON.stringify(r.value), - time_us: now * 1000, - indexed_at: now * 1000, + time_us: recordTimeUs(r.value, collection, config, nowUs), + indexed_at: nowUs, })); - await applyEvents(db, events, config, { - skipReplayDetection: options?.skipReplayDetection, - skipFeedFanout: true, - }); + if (subjectField) { + events = await filterEventsBySubject(db, events, subjectField); + } + + if (events.length > 0) { + await applyEvents(db, events, config, { + skipReplayDetection: options?.skipReplayDetection, + skipFeedFanout: true, + }); + } totalInserted += events.length; currentCursor = response.data.cursor ?? undefined; diff --git a/packages/contrail/src/core/constellation.ts b/packages/contrail/src/core/constellation.ts new file mode 100644 index 0000000..5812685 --- /dev/null +++ b/packages/contrail/src/core/constellation.ts @@ -0,0 +1,181 @@ +import { isDid } from "@atcute/lexicons/syntax"; +import type { ContrailConfig, Database, Logger } from "./types"; +import { + DEFAULT_CONSTELLATION_URL, + DEFAULT_FOLLOW_NSID, + recordsTableName, + shortNameForNsid, +} from "./types"; + +const PAGE_LIMIT = 100; +const DID_FILTER_CHUNK = 50; + +interface BacklinksPage { + links?: Array<{ + did?: string; + rkey?: string; + /** Some Constellation versions return the full URI rather than did/rkey split. */ + uri?: string; + }>; + cursor?: string; +} + +function getLogger(config: ContrailConfig): Logger { + return config.logger ?? console; +} + +/** Resolve effective Constellation config; null when disabled. */ +function getConstellationSettings( + config: ContrailConfig +): { url: string; userAgent: string } | null { + const c = config.constellation; + if (c === false) return null; + if (c?.enabled === false) return null; + return { + url: c?.url ?? DEFAULT_CONSTELLATION_URL, + userAgent: c?.userAgent ?? `contrail/${config.namespace}`, + }; +} + +/** Find the configured short name for `app.bsky.graph.follow`, if any. */ +function getFollowShort(config: ContrailConfig): string | null { + const short = shortNameForNsid(config, DEFAULT_FOLLOW_NSID); + return short ?? null; +} + +interface BacklinkRow { + did: string; + rkey: string; + uri: string; +} + +function parseBacklink(entry: NonNullable[number]): BacklinkRow | null { + if (entry.did && entry.rkey && isDid(entry.did)) { + return { + did: entry.did, + rkey: entry.rkey, + uri: entry.uri ?? `at://${entry.did}/${DEFAULT_FOLLOW_NSID}/${entry.rkey}`, + }; + } + if (entry.uri) { + const m = /^at:\/\/(did:[^/]+)\/[^/]+\/([^/]+)$/.exec(entry.uri); + if (m && isDid(m[1])) { + return { did: m[1], rkey: m[2], uri: entry.uri }; + } + } + return null; +} + +/** Filter a candidate-follower DID list to those already in our identities table. */ +async function filterKnownDids( + db: Database, + candidates: string[] +): Promise> { + const known = new Set(); + for (let i = 0; i < candidates.length; i += DID_FILTER_CHUNK) { + const chunk = candidates.slice(i, i + DID_FILTER_CHUNK); + const placeholders = chunk.map(() => "?").join(","); + const rows = await db + .prepare(`SELECT did FROM identities WHERE did IN (${placeholders})`) + .bind(...chunk) + .all<{ did: string }>(); + for (const r of rows.results ?? []) known.add(r.did); + } + return known; +} + +/** Fetch one page of getBacklinks. Returns null on non-2xx (caller decides whether to bail). */ +async function fetchBacklinksPage( + url: string, + userAgent: string, + subject: string, + cursor?: string +): Promise { + const u = new URL("/xrpc/blue.microcosm.links.getBacklinks", url); + u.searchParams.set("subject", subject); + u.searchParams.set("source", `${DEFAULT_FOLLOW_NSID}:.subject`); + u.searchParams.set("limit", String(PAGE_LIMIT)); + if (cursor) u.searchParams.set("cursor", cursor); + try { + const res = await fetch(u.toString(), { + headers: { "user-agent": userAgent, accept: "application/json" }, + }); + if (!res.ok) return null; + return (await res.json()) as BacklinksPage; + } catch { + return null; + } +} + +/** For a newly-known subject DID, find existing followers via Constellation + * and ingest synthesized follow records into the configured follow table. + * Best-effort: failures are logged but not retried (caller can re-trigger). */ +export async function backfillFollowersFromConstellation( + db: Database, + config: ContrailConfig, + subjectDid: string +): Promise { + const settings = getConstellationSettings(config); + if (!settings) return 0; + if (!isDid(subjectDid)) return 0; + const followShort = getFollowShort(config); + if (!followShort) return 0; + const log = getLogger(config); + + const followTable = recordsTableName(followShort); + const recordJson = JSON.stringify({ + $type: DEFAULT_FOLLOW_NSID, + subject: subjectDid, + createdAt: new Date().toISOString(), + }); + const nowUs = Date.now() * 1000; + + let cursor: string | undefined; + let inserted = 0; + let pages = 0; + + while (true) { + const page = await fetchBacklinksPage( + settings.url, + settings.userAgent, + subjectDid, + cursor + ); + if (!page) break; + pages++; + + const rows: BacklinkRow[] = (page.links ?? []) + .map(parseBacklink) + .filter((r): r is BacklinkRow => r !== null && r.did !== subjectDid); + + if (rows.length > 0) { + const known = await filterKnownDids( + db, + rows.map((r) => r.did) + ); + const survivors = rows.filter((r) => known.has(r.did)); + + for (const r of survivors) { + const result = await db + .prepare( + `INSERT INTO ${followTable} (uri, did, rkey, cid, record, time_us, indexed_at) + VALUES (?, ?, ?, NULL, ?, ?, ?) + ON CONFLICT(uri) DO NOTHING` + ) + .bind(r.uri, r.did, r.rkey, recordJson, nowUs, nowUs) + .run(); + inserted += (result as { changes?: number })?.changes ?? 0; + } + } + + cursor = page.cursor ?? undefined; + if (!cursor) break; + } + + if (inserted > 0) { + log.log( + `[constellation] subject=${subjectDid} pages=${pages} inserted=${inserted}` + ); + } + return inserted; +} diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts index a45ab9b..51414b8 100644 --- a/packages/contrail/src/core/db/records.ts +++ b/packages/contrail/src/core/db/records.ts @@ -18,6 +18,9 @@ import { spacesRecordsTableName, shortNameForNsid, nsidForShortName, + normalizeFeedTarget, + feedTargetMaxItems, + DEFAULT_FOLLOW_SHORT, } from "../types"; import { getSearchableFields, ftsTableName, buildFtsContent } from "../search"; import { ftsQueryClause, getDialect } from "../dialect"; @@ -210,10 +213,13 @@ function buildFeedStatements( if (!eventShort) return []; for (const [, feedConfig] of Object.entries(config.feeds)) { - const followTable = recordsTableName(feedConfig.follow); + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; + const followTable = recordsTableName(followShort); + const targets = feedConfig.targets.map(normalizeFeedTarget); + const targetShorts = targets.map((t) => t.collection); // Target collection: fan out to followers - if (feedConfig.targets.includes(eventShort)) { + if (targetShorts.includes(eventShort)) { if (event.operation === "create" || event.operation === "update") { stmts.push( db @@ -235,14 +241,15 @@ function buildFeedStatements( } // Follow collection: handle follow/unfollow - if (eventShort === feedConfig.follow) { + if (eventShort === followShort) { if (event.operation === "create") { const record = event.record ? JSON.parse(event.record) : null; const subject = record?.subject; if (subject) { - for (const targetShort of feedConfig.targets) { - const targetTable = recordsTableName(targetShort); - const targetNsid = nsidForShortName(config, targetShort) ?? targetShort; + for (const target of targets) { + const targetTable = recordsTableName(target.collection); + const targetNsid = nsidForShortName(config, target.collection) ?? target.collection; + const cap = feedTargetMaxItems(feedConfig, target); stmts.push( db .prepare( @@ -252,7 +259,7 @@ function buildFeedStatements( FROM ${targetTable} r WHERE r.did = ? ORDER BY r.time_us DESC - LIMIT 100` + LIMIT ${cap}` ) ) .bind(event.did, targetNsid, subject) @@ -265,8 +272,8 @@ function buildFeedStatements( const parsed = JSON.parse(existingRecord); const subject = parsed?.subject; if (subject) { - for (const targetShort of feedConfig.targets) { - const targetTable = recordsTableName(targetShort); + for (const target of targets) { + const targetTable = recordsTableName(target.collection); stmts.push( db .prepare( @@ -288,22 +295,47 @@ function buildFeedStatements( // --- Feed pruning --- +/** Prune feed_items per (actor, collection) to the given cap. + * + * - If `caps` is a number: legacy behavior — global per-actor cap across all collections. + * - If `caps` is a Map: each collection is pruned independently per actor, + * so high-volume collections (e.g. RSVPs) can't squeeze out lower-volume ones (e.g. events). + * Collections not present in the map are left alone. + */ export async function pruneFeedItems( db: Database, - maxItems: number + caps: number | Map ): Promise { - const result = await db - .prepare( - `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( - SELECT actor, uri FROM ( - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn - FROM feed_items - ) sub WHERE rn <= ? - )` - ) - .bind(maxItems) - .run(); - return (result as any)?.changes ?? 0; + if (typeof caps === "number") { + const result = await db + .prepare( + `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( + SELECT actor, uri FROM ( + SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn + FROM feed_items + ) sub WHERE rn <= ? + )` + ) + .bind(caps) + .run(); + return (result as any)?.changes ?? 0; + } + let total = 0; + for (const [collection, cap] of caps) { + const result = await db + .prepare( + `DELETE FROM feed_items WHERE collection = ? AND (actor, uri) NOT IN ( + SELECT actor, uri FROM ( + SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn + FROM feed_items WHERE collection = ? + ) sub WHERE rn <= ? + )` + ) + .bind(collection, collection, cap) + .run(); + total += (result as any)?.changes ?? 0; + } + return total; } // --- Cursor --- diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts index 3f3c676..5a56cbd 100644 --- a/packages/contrail/src/core/db/schema.ts +++ b/packages/contrail/src/core/db/schema.ts @@ -214,11 +214,16 @@ function buildFeedTables(config: ContrailConfig, dialect: SqlDialect): string[] actor TEXT NOT NULL, feed TEXT NOT NULL, completed INTEGER NOT NULL DEFAULT 0, + retries INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + started_at ${dialect.bigintType}, PRIMARY KEY (actor, feed) )`, ]; - const followCollections = new Set(Object.values(config.feeds).map((f) => f.follow)); + const followCollections = new Set( + Object.values(config.feeds).map((f) => f.follow ?? "follow") + ); for (const col of followCollections) { const table = recordsTableName(col); const safe = sanitizeName(col); @@ -250,6 +255,9 @@ const MIGRATIONS = [ "ALTER TABLE backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", "ALTER TABLE backfills ADD COLUMN last_error TEXT", "ALTER TABLE spaces_invites ADD COLUMN kind TEXT NOT NULL DEFAULT 'join'", + "ALTER TABLE feed_backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE feed_backfills ADD COLUMN last_error TEXT", + "ALTER TABLE feed_backfills ADD COLUMN started_at BIGINT", ]; async function runMigrations(db: Database): Promise { diff --git a/packages/contrail/src/core/jetstream.ts b/packages/contrail/src/core/jetstream.ts index d42a5e8..566c406 100644 --- a/packages/contrail/src/core/jetstream.ts +++ b/packages/contrail/src/core/jetstream.ts @@ -1,8 +1,14 @@ import { JetstreamSubscription } from "@atcute/jetstream"; import type { ContrailConfig, IngestEvent, Database, Logger } from "./types"; -import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS } from "./types"; +import { + getCollectionNsids, + getDependentNsids, + shortNameForNsid, + buildFeedTargetCaps, +} from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; import { refreshStaleIdentities } from "./identity"; +import { backfillFollowersFromConstellation } from "./constellation"; const BATCH_SIZE = 50; const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; // 1 hour @@ -27,7 +33,11 @@ export async function ingestEvents( cursor: number | null, safetyTimeoutMs: number = 25_000, knownDids?: Set -): Promise<{ events: IngestEvent[]; lastCursor: number | null }> { +): Promise<{ + events: IngestEvent[]; + lastCursor: number | null; + newlyKnownDids: string[]; +}> { const log = getLogger(config); const startTimeUs = Date.now() * 1000; const deadline = Date.now() + safetyTimeoutMs; @@ -45,6 +55,7 @@ export async function ingestEvents( let connectCount = 0; const seenUris = new Map(); // uri -> time_us of first occurrence const duplicateUris: string[] = []; + const newlyKnownDids = new Set(); const subscription = new JetstreamSubscription({ url: urls, @@ -81,6 +92,22 @@ export async function ingestEvents( if (filteredDidSamples.size < 10) filteredDidSamples.add(event.did); continue; } + // Subject filter: for collections with subjectField (e.g. follows + // pointing at a `subject` DID), drop records whose subject isn't a + // DID we care about. Trims network-wide social graph to the + // subjects our discoverable users overlap with. + const short = shortNameForNsid(config, commit.collection); + const subjectField = short + ? config.collections[short]?.subjectField + : undefined; + if (subjectField && commit.operation !== "delete") { + const subj = (commit.record as Record | undefined)?.[ + subjectField + ]; + if (typeof subj === "string" && !knownDids.has(subj)) { + continue; + } + } } const prev = seenUris.get(uri); @@ -115,7 +142,10 @@ export async function ingestEvents( ); if (knownDids && !dependentCollections.has(commit.collection)) { - knownDids.add(event.did); + if (!knownDids.has(event.did)) { + knownDids.add(event.did); + newlyKnownDids.add(event.did); + } } } @@ -172,7 +202,7 @@ export async function ingestEvents( ); } - return { events: collected, lastCursor }; + return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids] }; } // Run a full ingest cycle: init schema, load cursor, ingest, apply, save cursor @@ -220,7 +250,7 @@ export async function runIngestCycle( } } - const { events, lastCursor } = await ingestEvents( + const { events, lastCursor, newlyKnownDids } = await ingestEvents( config, cursor, timeoutMs, @@ -266,13 +296,26 @@ export async function runIngestCycle( log.log(`[ingest] no cursor returned from subscription; not saving`); } - // Prune feed items hourly + // Newly-discovered DIDs: ask Constellation for back-edges so they + // immediately appear in existing followers' feeds (best-effort, opt-out). + if (config.feeds && newlyKnownDids.length > 0) { + for (const subj of newlyKnownDids) { + try { + await backfillFollowersFromConstellation(db, config, subj); + } catch (err) { + log.warn(`[constellation] subject=${subj} failed: ${err}`); + } + } + } + + // Prune feed items hourly, per-target so high-volume targets don't + // squeeze out lower-volume ones. if (config.feeds && Date.now() - s.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { - const maxItems = Math.max( - ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS) - ); - const pruned = await pruneFeedItems(db, maxItems); - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + const caps = buildFeedTargetCaps(config); + if (caps.size > 0) { + const pruned = await pruneFeedItems(db, caps); + if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + } s.lastFeedPruneMs = Date.now(); } diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts index c1e1c18..e53740c 100644 --- a/packages/contrail/src/core/persistent.ts +++ b/packages/contrail/src/core/persistent.ts @@ -1,8 +1,15 @@ import type { JetstreamSubscription } from "@atcute/jetstream"; import type { ContrailConfig, IngestEvent, Database, Logger, ResolvedContrailConfig } from "./types"; -import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS, resolveConfig } from "./types"; +import { + getCollectionNsids, + getDependentNsids, + buildFeedTargetCaps, + resolveConfig, + shortNameForNsid, +} from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; import { refreshStaleIdentities } from "./identity"; +import { backfillFollowersFromConstellation } from "./constellation"; import { createIngestState } from "./jetstream"; import type { IngestState } from "./jetstream"; @@ -75,6 +82,7 @@ export async function runPersistent( collections, dependentCollections, knownDids, + newlyKnownDids: new Set(), state, log, createSubscription: options?.createSubscription, @@ -101,6 +109,9 @@ interface StreamOptions { collections: string[]; dependentCollections: Set; knownDids?: Set; + /** DIDs that crossed from unknown→known during this stream's lifetime. + * Drained on each flush so Constellation reverse-lookups can run for them. */ + newlyKnownDids?: Set; state: IngestState; log: Logger; createSubscription?: (cursor: number | null) => any; @@ -152,12 +163,25 @@ async function streamAndFlush( } } + // Drain newly-known DIDs and ask Constellation for back-edges. + if (config.feeds && opts.newlyKnownDids && opts.newlyKnownDids.size > 0) { + const drained = [...opts.newlyKnownDids]; + opts.newlyKnownDids.clear(); + for (const subj of drained) { + try { + await backfillFollowersFromConstellation(db, config, subj); + } catch (err) { + log.warn(`[constellation] subject=${subj} failed: ${err}`); + } + } + } + if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { - const maxItems = Math.max( - ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS) - ); - const pruned = await pruneFeedItems(db, maxItems); - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + const caps = buildFeedTargetCaps(config); + if (caps.size > 0) { + const pruned = await pruneFeedItems(db, caps); + if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); + } state.lastFeedPruneMs = Date.now(); } @@ -208,6 +232,17 @@ async function streamAndFlush( if (dependentCollections.has(commit.collection) && knownDids) { if (!knownDids.has(event.did)) continue; + // Subject filter: skip records whose subject DID isn't known. + const short = shortNameForNsid(config, commit.collection); + const subjectField = short + ? config.collections[short]?.subjectField + : undefined; + if (subjectField && commit.operation !== "delete") { + const subj = (commit.record as Record | undefined)?.[ + subjectField + ]; + if (typeof subj === "string" && !knownDids.has(subj)) continue; + } } const now = Date.now(); @@ -226,7 +261,10 @@ async function streamAndFlush( }); if (knownDids && !dependentCollections.has(commit.collection)) { - knownDids.add(event.did); + if (!knownDids.has(event.did)) { + knownDids.add(event.did); + opts.newlyKnownDids?.add(event.did); + } } } diff --git a/packages/contrail/src/core/router/feed.ts b/packages/contrail/src/core/router/feed.ts index 0430292..0d6a21d 100644 --- a/packages/contrail/src/core/router/feed.ts +++ b/packages/contrail/src/core/router/feed.ts @@ -1,78 +1,223 @@ -import type { Hono } from "hono"; -import type { ContrailConfig, Database, FeedConfig } from "../types"; +import type { Context, Hono } from "hono"; +import type { + ContrailConfig, + Database, + FeedConfig, + FeedTargetConfig, +} from "../types"; import { getDialect } from "../dialect"; -import { DEFAULT_FEED_MAX_ITEMS, recordsTableName } from "../types"; +import { + DEFAULT_FOLLOW_SHORT, + feedTargetMaxItems, + normalizeFeedTarget, + recordsTableName, + shortNameForNsid, +} from "../types"; import { resolveActor } from "../identity"; import { backfillUser } from "../backfill"; import { runPipeline } from "./collection"; -async function maybeBackfillFeed( +const BACKFILL_TIMEOUT_MS = 30_000; +const BACKFILL_REQUEST_TIMEOUT_MS = 10_000; +const BACKFILL_MAX_RETRIES = 3; +/** Re-arm a stuck in-progress row after this long (covers process crashes mid-backfill). */ +const BACKFILL_STALE_MS = 5 * 60 * 1000; + +interface FeedBackfillStatus { + completed: number; + retries: number; + last_error: string | null; + started_at: number | null; +} + +/** Schedule async work, preferring waitUntil on Cloudflare Workers so the + * runtime keeps the request alive until the promise settles. Falls back to + * fire-and-forget with a logged catch. */ +function scheduleBackground( + c: Context, + config: ContrailConfig, + task: () => Promise +): void { + const log = config.logger ?? console; + const promise = task().catch((err) => + log.error(`[feed] background task failed: ${err}`) + ); + try { + c.executionCtx.waitUntil(promise); + } catch { + // No executionCtx (Node/Bun); promise runs detached. + } +} + +/** Run the bootstrap copy + per-target prune. Returns rows inserted. */ +async function bootstrapFeedItems( db: Database, config: ContrailConfig, actor: string, - feedName: string, feedConfig: FeedConfig -): Promise { - const status = await db - .prepare("SELECT completed FROM feed_backfills WHERE actor = ? AND feed = ?") - .bind(actor, feedName) - .first<{ completed: number }>(); - - if (status?.completed) return; - - // Ensure the user's follow records are backfilled first - await backfillUser(db, actor, feedConfig.follow, Date.now() + 3_000, config, { - maxRetries: 0, - requestTimeout: 3_000, - }); - - // Mark as in-progress (idempotent) - await db - .prepare( - "INSERT INTO feed_backfills (actor, feed, completed) VALUES (?, ?, 0) ON CONFLICT DO NOTHING" - ) - .bind(actor, feedName) - .run(); +): Promise { + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; + const followTable = recordsTableName(followShort); + const targets = feedConfig.targets.map(normalizeFeedTarget); + let totalInserted = 0; - const maxItems = feedConfig.maxItems ?? DEFAULT_FEED_MAX_ITEMS; + for (const target of targets) { + const targetTable = recordsTableName(target.collection); + const targetCfg = config.collections[target.collection]; + if (!targetCfg) continue; + const cap = feedTargetMaxItems(feedConfig, target); - // Populate feed from existing records by followed users - const followTable = recordsTableName(feedConfig.follow); - for (const targetCol of feedConfig.targets) { - const targetTable = recordsTableName(targetCol); - await db + const insert = await db .prepare( getDialect(db).insertOrIgnore( `INSERT INTO feed_items (actor, uri, collection, time_us) - SELECT ?, r.uri, ?, r.time_us - FROM ${targetTable} r - WHERE r.did IN ( - SELECT ${getDialect(db).jsonExtract('f.record', 'subject')} + SELECT ?, r.uri, ?, r.time_us + FROM ${targetTable} r + WHERE r.did IN ( + SELECT ${getDialect(db).jsonExtract("f.record", "subject")} FROM ${followTable} f WHERE f.did = ? ) - ORDER BY r.time_us DESC - LIMIT ?` + ORDER BY r.time_us DESC + LIMIT ${cap}` + ) + ) + .bind(actor, targetCfg.collection, actor) + .run(); + totalInserted += (insert as { changes?: number })?.changes ?? 0; + + // Per-target prune so high-volume targets don't squeeze out lower-volume ones. + await db + .prepare( + `DELETE FROM feed_items WHERE actor = ? AND collection = ? AND uri NOT IN ( + SELECT uri FROM feed_items WHERE actor = ? AND collection = ? + ORDER BY time_us DESC LIMIT ? + )` + ) + .bind(actor, targetCfg.collection, actor, targetCfg.collection, cap) + .run(); + } + + return totalInserted; +} + +/** Run a full backfill cycle: walk follow records → bootstrap feed_items → mark complete. + * Updates feed_backfills row with retries/last_error on failure. */ +async function runFeedBackfill( + db: Database, + config: ContrailConfig, + actor: string, + feedName: string, + feedConfig: FeedConfig +): Promise { + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; + const followCfg = config.collections[followShort]; + if (!followCfg) return; + + try { + const inserted = await backfillUser( + db, + actor, + followCfg.collection, + Date.now() + BACKFILL_TIMEOUT_MS, + config, + { + skipReplayDetection: true, + maxRetries: BACKFILL_MAX_RETRIES, + requestTimeout: BACKFILL_REQUEST_TIMEOUT_MS, + } + ); + + // Bootstrap from whatever follow records we now have (may be from this + // backfill, from earlier live ingest, or both). + await bootstrapFeedItems(db, config, actor, feedConfig); + + // Only mark complete if the underlying follow backfill actually finished + // (backfills.completed = 1). Avoids the old bug where timeouts/empty + // walks would lock the user out of any retry. + const followStatus = await db + .prepare( + "SELECT completed FROM backfills WHERE did = ? AND collection = ?" + ) + .bind(actor, followCfg.collection) + .first<{ completed: number }>(); + + if (followStatus?.completed) { + await db + .prepare( + "UPDATE feed_backfills SET completed = 1, last_error = NULL WHERE actor = ? AND feed = ?" + ) + .bind(actor, feedName) + .run(); + } else { + // Walk didn't complete (timeout/error), but record the partial progress + // so the next request retries. + await db + .prepare( + "UPDATE feed_backfills SET retries = retries + 1, started_at = NULL, last_error = ? WHERE actor = ? AND feed = ?" ) + .bind( + `follow backfill incomplete (inserted=${inserted})`, + actor, + feedName + ) + .run(); + } + } catch (err) { + await db + .prepare( + "UPDATE feed_backfills SET retries = retries + 1, started_at = NULL, last_error = ? WHERE actor = ? AND feed = ?" ) - .bind(actor, targetCol, actor, maxItems) + .bind(String(err), actor, feedName) .run(); } +} - // Prune oldest items beyond the cap - await db +/** Decide whether to (re)kick off a background backfill, and do so if needed. + * Always returns immediately so the request path stays cheap. */ +async function maybeBackfillFeed( + c: Context, + db: Database, + config: ContrailConfig, + actor: string, + feedName: string, + feedConfig: FeedConfig +): Promise { + const status = await db .prepare( - `DELETE FROM feed_items WHERE actor = ? AND uri NOT IN ( - SELECT uri FROM feed_items WHERE actor = ? ORDER BY time_us DESC LIMIT ? - )` + "SELECT completed, retries, last_error, started_at FROM feed_backfills WHERE actor = ? AND feed = ?" ) - .bind(actor, actor, maxItems) - .run(); - - await db - .prepare("UPDATE feed_backfills SET completed = 1 WHERE actor = ? AND feed = ?") .bind(actor, feedName) - .run(); + .first(); + + if (status?.completed) return; + + const now = Date.now(); + + // Skip if a backfill is already in flight (started_at recently set) — avoids + // duplicate work from concurrent requests for the same actor. + if (status?.started_at && now - status.started_at < BACKFILL_STALE_MS) return; + + // Either no row, or stale started_at. Claim it. + if (!status) { + await db + .prepare( + "INSERT INTO feed_backfills (actor, feed, completed, started_at) VALUES (?, ?, 0, ?) ON CONFLICT DO NOTHING" + ) + .bind(actor, feedName, now) + .run(); + } else { + await db + .prepare( + "UPDATE feed_backfills SET started_at = ? WHERE actor = ? AND feed = ?" + ) + .bind(now, actor, feedName) + .run(); + } + + scheduleBackground(c, config, () => + runFeedBackfill(db, config, actor, feedName, feedConfig) + ); } export function registerFeedRoutes( @@ -101,11 +246,29 @@ export function registerFeedRoutes( const did = await resolveActor(db, actor); if (!did) return c.json({ error: "Could not resolve actor" }, 400); - await maybeBackfillFeed(db, config, did, feedName, feedConfig); + await maybeBackfillFeed(c, db, config, did, feedName, feedConfig); - const collection = params.get("collection") || feedConfig.targets[0]; - if (!feedConfig.targets.includes(collection)) { - return c.json({ error: "Collection not in feed targets" }, 400); + const targets = feedConfig.targets.map(normalizeFeedTarget); + if (targets.length === 0) { + return c.json({ error: "Feed has no targets configured" }, 500); + } + // Wire-level `collection` is an NSID (matches the generated lex enum and + // what's stored in feed_items.collection). Internally runPipeline expects + // the short name, so translate. + const requestedRaw = params.get("collection"); + let requestedShort: string; + if (!requestedRaw) { + requestedShort = targets[0].collection; + } else if (targets.some((t) => t.collection === requestedRaw)) { + // Tolerate callers passing the short name directly. + requestedShort = requestedRaw; + } else { + const asShort = shortNameForNsid(config, requestedRaw); + if (asShort && targets.some((t) => t.collection === asShort)) { + requestedShort = asShort; + } else { + return c.json({ error: "Collection not in feed targets" }, 400); + } } // Strip feed-specific params so runPipeline doesn't misinterpret them @@ -122,7 +285,7 @@ export function registerFeedRoutes( }; try { - const result = await runPipeline(db, config, collection, pipelineParams, source); + const result = await runPipeline(db, config, requestedShort, pipelineParams, source); return c.json(result); } catch (e: any) { if (e.message === "Could not resolve actor") { @@ -132,3 +295,5 @@ export function registerFeedRoutes( } }); } + +export type { FeedTargetConfig }; diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts index 53239d0..48cea2a 100644 --- a/packages/contrail/src/core/types.ts +++ b/packages/contrail/src/core/types.ts @@ -60,16 +60,63 @@ export type PipelineQueryHandler = ( config: ContrailConfig ) => Promise; +export interface FeedTargetConfig { + /** Short name of the target collection. */ + collection: string; + /** Per-target item cap. Falls back to FeedConfig.maxItems if unset. */ + maxItems?: number; +} + export interface FeedConfig { - /** Short name of the follow collection. */ - follow: string; - /** Short names of target collections to fan out to. */ - targets: string[]; - /** Max feed items per user (default: 200). Oldest items are pruned after backfill. */ + /** Short name of the follow collection. Defaults to "follow" + * (auto-added with NSID `app.bsky.graph.follow`, `discover: false`). */ + follow?: string; + /** Target collections to fan out to. Each entry is either a short name + * or `{ collection, maxItems? }` for per-target caps. */ + targets: (string | FeedTargetConfig)[]; + /** Default per-target item cap when a target doesn't specify its own + * (default: 200). Oldest items per (actor, collection) are pruned. */ maxItems?: number; } export const DEFAULT_FEED_MAX_ITEMS = 200; +export const DEFAULT_FOLLOW_NSID = "app.bsky.graph.follow"; +export const DEFAULT_FOLLOW_SHORT = "follow"; + +/** Normalize a feed target entry to FeedTargetConfig. */ +export function normalizeFeedTarget( + t: string | FeedTargetConfig +): FeedTargetConfig { + return typeof t === "string" ? { collection: t } : t; +} + +/** Resolve a feed's per-target item cap, falling back to FeedConfig.maxItems then global default. */ +export function feedTargetMaxItems( + feed: FeedConfig, + target: FeedTargetConfig +): number { + return target.maxItems ?? feed.maxItems ?? DEFAULT_FEED_MAX_ITEMS; +} + +/** Build a Map across all configured feeds, taking the + * largest cap if the same target collection appears in multiple feeds. */ +export function buildFeedTargetCaps( + config: ContrailConfig +): Map { + const caps = new Map(); + if (!config.feeds) return caps; + for (const feed of Object.values(config.feeds)) { + for (const t of feed.targets) { + const target = normalizeFeedTarget(t); + const colCfg = config.collections[target.collection]; + if (!colCfg) continue; + const cap = feedTargetMaxItems(feed, target); + const existing = caps.get(colCfg.collection) ?? 0; + if (cap > existing) caps.set(colCfg.collection, cap); + } + } + return caps; +} export type CollectionMethod = "listRecords" | "getRecord"; export const DEFAULT_COLLECTION_METHODS: CollectionMethod[] = [ @@ -96,6 +143,16 @@ export interface CollectionConfig { /** When spaces are enabled globally, emit a parallel spaces_records_ table * so this collection can also live inside spaces. Defaults to true. */ allowInSpaces?: boolean; + /** JSON field on the record used as the canonical event time, parsed and + * written into `time_us` during backfill (and clamped to now). Default + * `"createdAt"`. Set to `false` to disable parsing and keep ingest time. */ + timeField?: string | false; + /** JSON field on the record holding a DID that this record points at + * (e.g. `"subject"` for follows). When set on a `discover: false` + * collection, ingest also drops records whose subject DID is not in + * knownDids — useful for trimming network-wide social graphs to the + * subjects we care about. */ + subjectField?: string; } export interface ProfileConfig { @@ -168,8 +225,26 @@ export interface ContrailConfig { labels?: import("./labels/types").LabelsConfig; /** Customize the auto-generated `.authFull` lexicon. */ permissionSet?: PermissionSetConfig; + /** Constellation-backed reverse-follower lookup (default: enabled). + * When a DID is first seen producing a discoverable record, contrail + * queries Constellation for follow records pointing at that DID and + * ingests synthesized rows for any follower already in our identities + * table. Lets newcomers immediately appear in existing users' feeds. */ + constellation?: ConstellationConfig | false; +} + +export interface ConstellationConfig { + /** Override the default Constellation instance URL. */ + url?: string; + /** Sent as the User-Agent header per Constellation's request that + * callers identify themselves. Defaults to `contrail/`. */ + userAgent?: string; + /** Set false to disable lookups while keeping the table around. */ + enabled?: boolean; } +export const DEFAULT_CONSTELLATION_URL = "https://constellation.microcosm.blue"; + /** Single entry in an atproto permission-set's `permissions` array. * See https://atproto.com/guides/permission-sets for the full schema. */ export type PermissionEntry = @@ -216,7 +291,20 @@ export function resolveConfig(config: ContrailConfig): ResolvedContrailConfig { const profiles = (config.profiles ?? DEFAULT_PROFILES).map( normalizeProfileConfig ); - const collections = { ...config.collections }; + const collections: Record = {}; + + // Default `discover: false` for any collection whose NSID lives under the + // `app.bsky.*` namespace, since these are external/network-wide records that + // would otherwise blow up storage if left discoverable. + for (const [short, c] of Object.entries(config.collections)) { + collections[short] = + c.discover === undefined && + typeof c.collection === "string" && + c.collection.startsWith("app.bsky.") + ? { ...c, discover: false } + : c; + } + for (const p of profiles) { const short = p.shortName!; if (!collections[short]) { @@ -224,10 +312,26 @@ export function resolveConfig(config: ContrailConfig): ResolvedContrailConfig { } } - // Auto-add follow collections from feed configs as dependent collections if they're - // not already listed. Feed config already uses short names so nothing to resolve — - // but if the user forgot to declare the follow collection, we can't auto-add it without - // knowing its NSID. In that case we warn later via validateConfig. + // Auto-add a follow collection for any feed that doesn't declare one. + // Default short name `follow` → `app.bsky.graph.follow`, with a `subject` + // filter so we only persist follows pointing at known DIDs. + const feeds = config.feeds; + if (feeds) { + const usedFollowShorts = new Set(); + for (const [, feed] of Object.entries(feeds)) { + const shortName = feed.follow ?? DEFAULT_FOLLOW_SHORT; + usedFollowShorts.add(shortName); + } + for (const short of usedFollowShorts) { + if (!collections[short]) { + collections[short] = { + collection: DEFAULT_FOLLOW_NSID, + discover: false, + subjectField: "subject", + }; + } + } + } const base = { ...config, @@ -277,7 +381,11 @@ function _resolveQueryableMaps(config: ContrailConfig): ResolvedMaps { export function getFeedFollowShortNames(config: ContrailConfig): string[] { if (!config.feeds) return []; - return [...new Set(Object.values(config.feeds).map((f) => f.follow))]; + return [ + ...new Set( + Object.values(config.feeds).map((f) => f.follow ?? DEFAULT_FOLLOW_SHORT) + ), + ]; } /** Alias for getFeedFollowShortNames. */ @@ -375,15 +483,17 @@ export function validateConfig(config: ContrailConfig): void { if (config.feeds) { for (const [feedName, feed] of Object.entries(config.feeds)) { - if (!config.collections[feed.follow]) { + const followShort = feed.follow ?? DEFAULT_FOLLOW_SHORT; + if (!config.collections[followShort]) { throw new Error( - `Feed "${feedName}" references unknown follow collection "${feed.follow}"` + `Feed "${feedName}" references unknown follow collection "${followShort}"` ); } - for (const target of feed.targets) { - if (!config.collections[target]) { + for (const t of feed.targets) { + const targetShort = normalizeFeedTarget(t).collection; + if (!config.collections[targetShort]) { throw new Error( - `Feed "${feedName}" references unknown target collection "${target}"` + `Feed "${feedName}" references unknown target collection "${targetShort}"` ); } } diff --git a/packages/lexicons/src/generate.ts b/packages/lexicons/src/generate.ts index 9ca0dd2..93151f8 100644 --- a/packages/lexicons/src/generate.ts +++ b/packages/lexicons/src/generate.ts @@ -401,9 +401,16 @@ export function generateLexicons(options: GenerateOptions): Record f.targets))]; + // feedConfig.targets are short names (or `{ collection, maxItems? }`); + // expose NSIDs in the lexicon since the `collection` param filters by + // the record's NSID at the wire level. + const allTargets = [ + ...new Set( + Object.values(config.feeds).flatMap((f) => + f.targets.map((t) => (typeof t === "string" ? t : t.collection)) + ) + ), + ]; const allTargetNsids = allTargets .map((t) => config.collections[t]?.collection) .filter((n): n is string => !!n); @@ -1033,11 +1040,12 @@ export function generateLexicons(options: GenerateOptions): Record config.collections[f.follow]?.collection) + .map((f) => config.collections[f.follow ?? "follow"]?.collection) .filter((nsid): nsid is string => typeof nsid === "string") : []; const pullNsids = new Set([...collectionNsids, ...profileNsids, ...feedFollowNsids]);