From 9cca8cb6ee59b2bc2879338f3d1fde643492c400 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sun, 3 May 2026 19:29:41 +0200 Subject: [PATCH 1/8] quick follow feed fix --- .changeset/feed-follow-pull-nsids.md | 5 ++ packages/lexicons/src/generate.ts | 10 ++- packages/lexicons/tests/generate.test.ts | 78 ++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 .changeset/feed-follow-pull-nsids.md diff --git a/.changeset/feed-follow-pull-nsids.md b/.changeset/feed-follow-pull-nsids.md new file mode 100644 index 0000000..f64f452 --- /dev/null +++ b/.changeset/feed-follow-pull-nsids.md @@ -0,0 +1,5 @@ +--- +"@atmo-dev/contrail-lexicons": patch +--- + +fix: resolve `feeds[*].follow` short names to NSIDs when emitting `lex.config.js`. previously the generator pushed the raw short name (e.g. `"follow"`) into `pull.sources[0].nsids`, causing `lex-cli pull` to fail with `ValitaError: must be valid nsid`. now matches the existing `collections` / `profiles` resolution path; feeds pointing at unknown collections are skipped instead of leaking `undefined`. diff --git a/packages/lexicons/src/generate.ts b/packages/lexicons/src/generate.ts index ed8bb63..9ca0dd2 100644 --- a/packages/lexicons/src/generate.ts +++ b/packages/lexicons/src/generate.ts @@ -1031,7 +1031,15 @@ export function generateLexicons(options: GenerateOptions): Record (typeof p === "string" ? p : p.collection) ); - const feedFollowNsids = config.feeds ? Object.values(config.feeds).map((f) => f.follow) : []; + // f.follow is a short name (a key in config.collections), not an NSID — resolve + // it before pushing into the pull list, otherwise lex-cli pull rejects it as + // "must be valid nsid". Filter out any feed pointing at a non-existent + // collection so we never emit an undefined. + const feedFollowNsids = config.feeds + ? Object.values(config.feeds) + .map((f) => config.collections[f.follow]?.collection) + .filter((nsid): nsid is string => typeof nsid === "string") + : []; const pullNsids = new Set([...collectionNsids, ...profileNsids, ...feedFollowNsids]); for (const ref of allRefs) { if (!ref.startsWith("com.atproto.")) pullNsids.add(ref); diff --git a/packages/lexicons/tests/generate.test.ts b/packages/lexicons/tests/generate.test.ts index 44f45c5..6a8e884 100644 --- a/packages/lexicons/tests/generate.test.ts +++ b/packages/lexicons/tests/generate.test.ts @@ -373,3 +373,81 @@ describe("manifest emission (lexicons/generated/index.ts)", () => { } }); }); + +describe("runtime files: lex.config.js pull NSIDs", () => { + let workdir: string; + + beforeAll(() => { + workdir = mkdtempSync(join(tmpdir(), "contrail-feedpull-")); + }); + afterAll(() => { + rmSync(workdir, { recursive: true, force: true }); + }); + + function readPullNsids(): string[] { + const lexConfig = readFileSync(join(workdir, "lex.config.js"), "utf-8"); + const match = lexConfig.match(/nsids:\s*(\[[\s\S]*?\])/); + if (!match) throw new Error("could not locate pull nsids array in lex.config.js"); + return JSON.parse(match[1]); + } + + it("resolves feed.follow short names to NSIDs (regression: lex-cli pull rejects bare short names)", () => { + const config: ContrailConfig = { + namespace: "test.app", + collections: { + follow: { collection: "app.bsky.graph.follow" }, + event: { collection: "community.lexicon.calendar.event" }, + }, + feeds: { + network: { follow: "follow", targets: ["event"] }, + }, + }; + + generateLexicons({ + config, + rootDir: workdir, + lexiconDirs: [], + writeRuntimeFiles: true, + quiet: true, + }); + + const nsids = readPullNsids(); + + // Every entry must be a valid NSID — at least one dot, and never equal a feed short name. + const feedShortNames = Object.keys(config.collections); + for (const nsid of nsids) { + expect(nsid).toMatch(/\./); + expect(feedShortNames).not.toContain(nsid); + } + + // The follow collection's NSID is included via the feeds path. + expect(nsids).toContain("app.bsky.graph.follow"); + }); + + it("skips feeds whose follow short name is missing from collections (no undefineds in pull list)", () => { + const config = { + namespace: "test.app", + collections: { + event: { collection: "community.lexicon.calendar.event" }, + }, + feeds: { + broken: { follow: "doesNotExist", targets: ["event"] }, + }, + } as unknown as ContrailConfig; + + generateLexicons({ + config, + rootDir: workdir, + lexiconDirs: [], + writeRuntimeFiles: true, + quiet: true, + }); + + const nsids = readPullNsids(); + for (const nsid of nsids) { + expect(typeof nsid).toBe("string"); + expect(nsid).toMatch(/\./); + } + expect(nsids).not.toContain("doesNotExist"); + }); +}); -- 2.51.2 From 606284bb634d84c2273761dd656f562ebe04fe8c Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 3 May 2026 17:30:22 +0000 Subject: [PATCH 2/8] Version Packages --- .changeset/feed-follow-pull-nsids.md | 5 ----- packages/lexicons/CHANGELOG.md | 6 ++++++ packages/lexicons/package.json | 2 +- 3 files changed, 7 insertions(+), 6 deletions(-) delete mode 100644 .changeset/feed-follow-pull-nsids.md diff --git a/.changeset/feed-follow-pull-nsids.md b/.changeset/feed-follow-pull-nsids.md deleted file mode 100644 index f64f452..0000000 --- a/.changeset/feed-follow-pull-nsids.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@atmo-dev/contrail-lexicons": patch ---- - -fix: resolve `feeds[*].follow` short names to NSIDs when emitting `lex.config.js`. previously the generator pushed the raw short name (e.g. `"follow"`) into `pull.sources[0].nsids`, causing `lex-cli pull` to fail with `ValitaError: must be valid nsid`. now matches the existing `collections` / `profiles` resolution path; feeds pointing at unknown collections are skipped instead of leaking `undefined`. diff --git a/packages/lexicons/CHANGELOG.md b/packages/lexicons/CHANGELOG.md index 69e51d1..4962bbc 100644 --- a/packages/lexicons/CHANGELOG.md +++ b/packages/lexicons/CHANGELOG.md @@ -1,5 +1,11 @@ # @atmo-dev/contrail-lexicons +## 0.4.3 + +### Patch Changes + +- 9cca8cb: fix: resolve `feeds[*].follow` short names to NSIDs when emitting `lex.config.js`. previously the generator pushed the raw short name (e.g. `"follow"`) into `pull.sources[0].nsids`, causing `lex-cli pull` to fail with `ValitaError: must be valid nsid`. now matches the existing `collections` / `profiles` resolution path; feeds pointing at unknown collections are skipped instead of leaking `undefined`. + ## 0.4.2 ### Patch Changes diff --git a/packages/lexicons/package.json b/packages/lexicons/package.json index 109bf42..bc8e089 100644 --- a/packages/lexicons/package.json +++ b/packages/lexicons/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail-lexicons", - "version": "0.4.2", + "version": "0.4.3", "description": "Generate atproto lexicon JSON (and optionally TypeScript types via @atcute/lex-cli) from a Contrail config.", "type": "module", "files": [ -- 2.51.2 From 1a6d8cf32d59296d9516b9b2a7836de0cc9d7f5f Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sun, 3 May 2026 21:31:09 +0200 Subject: [PATCH 3/8] follow feed fixes --- .changeset/feed-targets-shape.md | 5 + .changeset/follow-feed-overhaul.md | 23 ++ packages/contrail/src/core/backfill.ts | 107 +++++++- packages/contrail/src/core/constellation.ts | 181 +++++++++++++ packages/contrail/src/core/db/records.ts | 76 ++++-- packages/contrail/src/core/db/schema.ts | 10 +- packages/contrail/src/core/jetstream.ts | 65 ++++- packages/contrail/src/core/persistent.ts | 52 +++- packages/contrail/src/core/router/feed.ts | 275 ++++++++++++++++---- packages/contrail/src/core/types.ts | 142 ++++++++-- packages/lexicons/src/generate.ts | 20 +- 11 files changed, 830 insertions(+), 126 deletions(-) create mode 100644 .changeset/feed-targets-shape.md create mode 100644 .changeset/follow-feed-overhaul.md create mode 100644 packages/contrail/src/core/constellation.ts 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]); -- 2.51.2 From b24cc67df77ece2074789323b218cc00a8242437 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 3 May 2026 19:32:12 +0000 Subject: [PATCH 4/8] Version Packages --- .changeset/feed-targets-shape.md | 5 ----- .changeset/follow-feed-overhaul.md | 23 ----------------------- packages/contrail/CHANGELOG.md | 24 ++++++++++++++++++++++++ packages/contrail/package.json | 2 +- packages/lexicons/CHANGELOG.md | 8 ++++++++ packages/lexicons/package.json | 2 +- 6 files changed, 34 insertions(+), 30 deletions(-) delete mode 100644 .changeset/feed-targets-shape.md delete mode 100644 .changeset/follow-feed-overhaul.md diff --git a/.changeset/feed-targets-shape.md b/.changeset/feed-targets-shape.md deleted file mode 100644 index 357f1bf..0000000 --- a/.changeset/feed-targets-shape.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@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 deleted file mode 100644 index 295c44f..0000000 --- a/.changeset/follow-feed-overhaul.md +++ /dev/null @@ -1,23 +0,0 @@ ---- -"@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/CHANGELOG.md b/packages/contrail/CHANGELOG.md index 3390b9c..01982cf 100644 --- a/packages/contrail/CHANGELOG.md +++ b/packages/contrail/CHANGELOG.md @@ -1,5 +1,29 @@ # @atmo-dev/contrail +## 0.5.0 + +### Minor Changes + +- 1a6d8cf: 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. + ## 0.4.2 ### Patch Changes diff --git a/packages/contrail/package.json b/packages/contrail/package.json index a64484d..49123de 100644 --- a/packages/contrail/package.json +++ b/packages/contrail/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail", - "version": "0.4.2", + "version": "0.5.0", "description": "Index AT Protocol records with typed XRPC endpoints. Cloudflare Workers + D1, SvelteKit, Node.js.", "type": "module", "sideEffects": false, diff --git a/packages/lexicons/CHANGELOG.md b/packages/lexicons/CHANGELOG.md index 4962bbc..d7ae8fa 100644 --- a/packages/lexicons/CHANGELOG.md +++ b/packages/lexicons/CHANGELOG.md @@ -1,5 +1,13 @@ # @atmo-dev/contrail-lexicons +## 0.4.4 + +### Patch Changes + +- 1a6d8cf: 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. +- Updated dependencies [1a6d8cf] + - @atmo-dev/contrail@0.5.0 + ## 0.4.3 ### Patch Changes diff --git a/packages/lexicons/package.json b/packages/lexicons/package.json index bc8e089..3053c3d 100644 --- a/packages/lexicons/package.json +++ b/packages/lexicons/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail-lexicons", - "version": "0.4.3", + "version": "0.4.4", "description": "Generate atproto lexicon JSON (and optionally TypeScript types via @atcute/lex-cli) from a Contrail config.", "type": "module", "files": [ -- 2.51.2 From 070442aaab83e1ea4d11ccb425f6602a0318966d Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sun, 3 May 2026 21:45:01 +0200 Subject: [PATCH 5/8] small lex-gen fix --- .changeset/feed-targets-shape.md | 4 +++- packages/lexicons/src/generate.ts | 12 ++++++++---- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/.changeset/feed-targets-shape.md b/.changeset/feed-targets-shape.md index 357f1bf..3607331 100644 --- a/.changeset/feed-targets-shape.md +++ b/.changeset/feed-targets-shape.md @@ -2,4 +2,6 @@ "@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. +Handle the new `FeedConfig.targets` shape (`string | { collection, maxItems? }`) when generating the feed lexicon and computing pull NSIDs. + +When `FeedConfig.follow` is unset, mirror contrail's runtime default and emit `app.bsky.graph.follow` into the generated `lex.config.js` pull list — previously the auto-added follow collection was missing from `pull.sources[0].nsids`, so lex-cli wouldn't fetch its schema. diff --git a/packages/lexicons/src/generate.ts b/packages/lexicons/src/generate.ts index 93151f8..8b39b6d 100644 --- a/packages/lexicons/src/generate.ts +++ b/packages/lexicons/src/generate.ts @@ -1040,12 +1040,16 @@ export function generateLexicons(options: GenerateOptions): Record config.collections[f.follow ?? "follow"]?.collection) + .map((f) => { + if (!f.follow) return DEFAULT_FOLLOW_NSID; + return config.collections[f.follow]?.collection; + }) .filter((nsid): nsid is string => typeof nsid === "string") : []; const pullNsids = new Set([...collectionNsids, ...profileNsids, ...feedFollowNsids]); -- 2.51.2 From d6661e51da03815dd5b817940d8f7f6dbb13ba34 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 3 May 2026 19:46:54 +0000 Subject: [PATCH 6/8] Version Packages --- .changeset/lex-pull-default-follow.md | 5 ----- packages/lexicons/CHANGELOG.md | 6 ++++++ packages/lexicons/package.json | 2 +- 3 files changed, 7 insertions(+), 6 deletions(-) delete mode 100644 .changeset/lex-pull-default-follow.md diff --git a/.changeset/lex-pull-default-follow.md b/.changeset/lex-pull-default-follow.md deleted file mode 100644 index 1c9a6d3..0000000 --- a/.changeset/lex-pull-default-follow.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@atmo-dev/contrail-lexicons": patch ---- - -Include `app.bsky.graph.follow` in the generated `lex.config.js` pull list when a feed leaves `FeedConfig.follow` unset. Mirrors the runtime default that `resolveConfig` auto-adds, so `lex-cli pull` fetches the schema instead of skipping it. diff --git a/packages/lexicons/CHANGELOG.md b/packages/lexicons/CHANGELOG.md index d7ae8fa..9a1c5d1 100644 --- a/packages/lexicons/CHANGELOG.md +++ b/packages/lexicons/CHANGELOG.md @@ -1,5 +1,11 @@ # @atmo-dev/contrail-lexicons +## 0.4.5 + +### Patch Changes + +- Include `app.bsky.graph.follow` in the generated `lex.config.js` pull list when a feed leaves `FeedConfig.follow` unset. Mirrors the runtime default that `resolveConfig` auto-adds, so `lex-cli pull` fetches the schema instead of skipping it. + ## 0.4.4 ### Patch Changes diff --git a/packages/lexicons/package.json b/packages/lexicons/package.json index 3053c3d..7a5dab9 100644 --- a/packages/lexicons/package.json +++ b/packages/lexicons/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail-lexicons", - "version": "0.4.4", + "version": "0.4.5", "description": "Generate atproto lexicon JSON (and optionally TypeScript types via @atcute/lex-cli) from a Contrail config.", "type": "module", "files": [ -- 2.51.2 From af2471448d21d6ed2ad7d795e0002c72951e539a Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Tue, 5 May 2026 21:29:14 +0200 Subject: [PATCH 7/8] add record filter, add identity events --- .../record-filter-and-identity-events.md | 8 ++ packages/contrail/src/core/identity.ts | 18 ++++ packages/contrail/src/core/jetstream.ts | 42 ++++++++-- packages/contrail/src/core/persistent.ts | 27 ++++-- packages/contrail/src/core/types.ts | 8 ++ packages/contrail/tests/persistent.test.ts | 82 ++++++++++++++++++- 6 files changed, 172 insertions(+), 13 deletions(-) create mode 100644 .changeset/record-filter-and-identity-events.md diff --git a/.changeset/record-filter-and-identity-events.md b/.changeset/record-filter-and-identity-events.md new file mode 100644 index 0000000..69925f7 --- /dev/null +++ b/.changeset/record-filter-and-identity-events.md @@ -0,0 +1,8 @@ +--- +"@atmo-dev/contrail": minor +--- + +Add per-collection `recordFilter` and apply Jetstream `#identity` handle changes during ingest. + +- `CollectionConfig.recordFilter?: (record) => boolean` runs against each create/update during ingest; returning false drops the record before it reaches the DB. Useful for narrowing high-volume collections to just the records you care about (e.g. only `app.bsky.feed.post` records mentioning a particular URL). Deletes are not filtered, so they still tear down any record the filter previously let through. Throws are caught, logged, and treated as drops. +- Jetstream `#identity` events (handle changes) now flow through to the `identities` table via a new `applyIdentityEvent` helper. UPDATE-only — unknown DIDs are no-ops so we don't materialize partial rows lacking PDS. diff --git a/packages/contrail/src/core/identity.ts b/packages/contrail/src/core/identity.ts index 05040fd..fe207b3 100644 --- a/packages/contrail/src/core/identity.ts +++ b/packages/contrail/src/core/identity.ts @@ -119,6 +119,24 @@ export async function resolveActor( return resolved.did; } +/** + * Apply a handle change from a Jetstream `#identity` event. + * + * UPDATE-only — does not create a row for unknown DIDs (we'd lack PDS, and + * partial rows confuse the rest of the pipeline). PDS column is left + * untouched; it gets refreshed lazily via `getPDS` / next slingshot resolve. + */ +export async function applyIdentityEvent( + db: Database, + did: string, + handle: string +): Promise { + await db + .prepare("UPDATE identities SET handle = ?, resolved_at = ? WHERE did = ?") + .bind(handle, Date.now(), did) + .run(); +} + export async function refreshStaleIdentities( db: Database, dids: string[] diff --git a/packages/contrail/src/core/jetstream.ts b/packages/contrail/src/core/jetstream.ts index 566c406..abc0287 100644 --- a/packages/contrail/src/core/jetstream.ts +++ b/packages/contrail/src/core/jetstream.ts @@ -7,7 +7,7 @@ import { buildFeedTargetCaps, } from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; -import { refreshStaleIdentities } from "./identity"; +import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; const BATCH_SIZE = 50; @@ -37,6 +37,7 @@ export async function ingestEvents( events: IngestEvent[]; lastCursor: number | null; newlyKnownDids: string[]; + identityUpdates: Map; }> { const log = getLogger(config); const startTimeUs = Date.now() * 1000; @@ -56,6 +57,7 @@ export async function ingestEvents( const seenUris = new Map(); // uri -> time_us of first occurrence const duplicateUris: string[] = []; const newlyKnownDids = new Set(); + const identityUpdates = new Map(); const subscription = new JetstreamSubscription({ url: urls, @@ -86,6 +88,9 @@ export async function ingestEvents( const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`; + const short = shortNameForNsid(config, commit.collection); + const collectionCfg = short ? config.collections[short] : undefined; + if (dependentCollections.has(commit.collection) && knownDids) { if (!knownDids.has(event.did)) { filteredUnknownDid++; @@ -96,10 +101,7 @@ export async function ingestEvents( // 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; + const subjectField = collectionCfg?.subjectField; if (subjectField && commit.operation !== "delete") { const subj = (commit.record as Record | undefined)?.[ subjectField @@ -110,6 +112,17 @@ export async function ingestEvents( } } + if (collectionCfg?.recordFilter && commit.operation !== "delete") { + const rec = commit.record as Record | undefined; + let keep = false; + try { + keep = !!(rec && collectionCfg.recordFilter(rec)); + } catch (err) { + log.warn(`[ingest] recordFilter threw for ${uri}: ${err}`); + } + if (!keep) continue; + } + const prev = seenUris.get(uri); if (prev !== undefined) { duplicateUris.push(uri); @@ -147,6 +160,8 @@ export async function ingestEvents( newlyKnownDids.add(event.did); } } + } else if (event.kind === "identity") { + identityUpdates.set(event.did, event.identity.handle); } if (event.time_us >= startTimeUs) { @@ -202,7 +217,7 @@ export async function ingestEvents( ); } - return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids] }; + return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids], identityUpdates }; } // Run a full ingest cycle: init schema, load cursor, ingest, apply, save cursor @@ -250,7 +265,7 @@ export async function runIngestCycle( } } - const { events, lastCursor, newlyKnownDids } = await ingestEvents( + const { events, lastCursor, newlyKnownDids, identityUpdates } = await ingestEvents( config, cursor, timeoutMs, @@ -275,6 +290,19 @@ export async function runIngestCycle( await applyEvents(db, batch, config, { pubsub }); } + // Apply handle changes from #identity events. UPDATE-only, so unknown + // DIDs are no-ops — we don't want to create partial rows lacking PDS. + if (identityUpdates.size > 0) { + for (const [did, handle] of identityUpdates) { + try { + await applyIdentityEvent(db, did, handle); + } catch (err) { + log.warn(`[ingest] identity update failed for ${did}: ${err}`); + } + } + log.log(`[ingest] applied ${identityUpdates.size} identity event(s)`); + } + // Refresh stale/missing identities for DIDs in this batch const uniqueDids = [...new Set(events.map((e) => e.did))]; if (uniqueDids.length > 0) { diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts index e53740c..882c9b8 100644 --- a/packages/contrail/src/core/persistent.ts +++ b/packages/contrail/src/core/persistent.ts @@ -8,7 +8,7 @@ import { shortNameForNsid, } from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; -import { refreshStaleIdentities } from "./identity"; +import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; import { createIngestState } from "./jetstream"; import type { IngestState } from "./jetstream"; @@ -230,13 +230,13 @@ async function streamAndFlush( if (event.kind === "commit") { const { commit } = event; + const short = shortNameForNsid(config, commit.collection); + const collectionCfg = short ? config.collections[short] : undefined; + 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; + const subjectField = collectionCfg?.subjectField; if (subjectField && commit.operation !== "delete") { const subj = (commit.record as Record | undefined)?.[ subjectField @@ -245,6 +245,17 @@ async function streamAndFlush( } } + if (collectionCfg?.recordFilter && commit.operation !== "delete") { + const rec = commit.record as Record | undefined; + let keep = false; + try { + keep = !!(rec && collectionCfg.recordFilter(rec)); + } catch (err) { + log.warn(`recordFilter threw for ${commit.collection}/${commit.rkey}: ${err}`); + } + if (!keep) continue; + } + const now = Date.now(); const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`; @@ -266,6 +277,12 @@ async function streamAndFlush( opts.newlyKnownDids?.add(event.did); } } + } else if (event.kind === "identity") { + try { + await applyIdentityEvent(db, event.did, event.identity.handle); + } catch (err) { + log.warn(`Identity update failed for ${event.did}: ${err}`); + } } if (buffer.length >= batchSize) { diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts index 48cea2a..c59f116 100644 --- a/packages/contrail/src/core/types.ts +++ b/packages/contrail/src/core/types.ts @@ -153,6 +153,14 @@ export interface CollectionConfig { * knownDids — useful for trimming network-wide social graphs to the * subjects we care about. */ subjectField?: string; + /** Per-record predicate run during ingest. Returning false drops the + * record before it hits the buffer / DB. Runs only for create/update; + * deletes always pass through (the delete may target a record that *did* + * pass an earlier version of the filter). Thrown errors are caught, + * logged, and treated as "drop". Note: Jetstream filters only by + * `wantedCollections`, so non-matching records still travel over the wire + * — this trims what gets persisted, not bandwidth. */ + recordFilter?: (record: Record) => boolean; } export interface ProfileConfig { diff --git a/packages/contrail/tests/persistent.test.ts b/packages/contrail/tests/persistent.test.ts index f248cf0..c48c7db 100644 --- a/packages/contrail/tests/persistent.test.ts +++ b/packages/contrail/tests/persistent.test.ts @@ -1,13 +1,16 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import type { ContrailConfig, Database } from "../src/core/types"; +import { resolveConfig } from "../src/core/types"; import { createTestDb, createTestDbWithSchema, TEST_CONFIG } from "./helpers"; import { runPersistent } from "../src/core/persistent"; import { getLastCursor, queryRecords } from "../src/core/db/records"; import { initSchema } from "../src/core/db/schema"; // Mock identity resolution to avoid network calls in tests +const applyIdentityEventMock = vi.fn().mockResolvedValue(undefined); vi.mock("../src/core/identity", () => ({ refreshStaleIdentities: vi.fn().mockResolvedValue(undefined), + applyIdentityEvent: (...args: unknown[]) => applyIdentityEventMock(...args), })); let db: Database; @@ -292,9 +295,79 @@ describe("runPersistent", () => { expect(row!.count_rsvp_going).toBe(1); }); + it("drops records that fail a collection's recordFilter", async () => { + // Filter accepts only events whose `name` contains "keep". The other + // events have well-formed records but should never reach the DB. + const filterConfig = resolveConfig({ + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + recordFilter: (r) => + typeof r.name === "string" && r.name.includes("keep"), + }, + }, + }); + + const freshDb = createTestDb(); + await initSchema(freshDb, filterConfig); + + const events = [ + { + kind: "commit" as const, + did: "did:plc:a", + time_us: 7000, + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "drop1", + cid: "c1", + record: { name: "drop me", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }, + { + kind: "commit" as const, + did: "did:plc:b", + time_us: 7001, + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "keep1", + cid: "c2", + record: { name: "keep this", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }, + ]; + + const controller = new AbortController(); + const promise = runPersistent(freshDb, filterConfig, { + batchSize: 100, + flushIntervalMs: 50, + signal: controller.signal, + createSubscription: () => mockSubscription(events) as any, + }); + + await new Promise((r) => setTimeout(r, 200)); + controller.abort(); + await promise; + + const result = await queryRecords(freshDb, filterConfig, { + collection: "community.lexicon.calendar.event", + limit: 100, + }); + expect(result.records.length).toBe(1); + expect(result.records[0]!.uri).toContain("/keep1"); + }); + it("skips non-commit events", async () => { + applyIdentityEventMock.mockClear(); const events = [ - { kind: "identity" as const, did: "did:plc:someone", time_us: 4000 }, + { + kind: "identity" as const, + did: "did:plc:someone", + time_us: 4000, + identity: { did: "did:plc:someone", handle: "newhandle.test", seq: 1, time: "2026-04-01T10:00:00Z" }, + }, { kind: "commit" as const, did: "did:plc:real", @@ -327,5 +400,12 @@ describe("runPersistent", () => { limit: 100, }); expect(result.records.length).toBe(1); + // Identity events update the identities table even though they don't + // produce records. + expect(applyIdentityEventMock).toHaveBeenCalledWith( + expect.anything(), + "did:plc:someone", + "newhandle.test" + ); }); }); -- 2.51.2 From 3caa00ab33803de32f9b977827ded25c9509655e Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 5 May 2026 19:30:16 +0000 Subject: [PATCH 8/8] Version Packages --- .changeset/record-filter-and-identity-events.md | 8 -------- packages/contrail/CHANGELOG.md | 9 +++++++++ packages/contrail/package.json | 2 +- packages/lexicons/CHANGELOG.md | 7 +++++++ packages/lexicons/package.json | 2 +- 5 files changed, 18 insertions(+), 10 deletions(-) delete mode 100644 .changeset/record-filter-and-identity-events.md diff --git a/.changeset/record-filter-and-identity-events.md b/.changeset/record-filter-and-identity-events.md deleted file mode 100644 index 69925f7..0000000 --- a/.changeset/record-filter-and-identity-events.md +++ /dev/null @@ -1,8 +0,0 @@ ---- -"@atmo-dev/contrail": minor ---- - -Add per-collection `recordFilter` and apply Jetstream `#identity` handle changes during ingest. - -- `CollectionConfig.recordFilter?: (record) => boolean` runs against each create/update during ingest; returning false drops the record before it reaches the DB. Useful for narrowing high-volume collections to just the records you care about (e.g. only `app.bsky.feed.post` records mentioning a particular URL). Deletes are not filtered, so they still tear down any record the filter previously let through. Throws are caught, logged, and treated as drops. -- Jetstream `#identity` events (handle changes) now flow through to the `identities` table via a new `applyIdentityEvent` helper. UPDATE-only — unknown DIDs are no-ops so we don't materialize partial rows lacking PDS. diff --git a/packages/contrail/CHANGELOG.md b/packages/contrail/CHANGELOG.md index 01982cf..63e68f6 100644 --- a/packages/contrail/CHANGELOG.md +++ b/packages/contrail/CHANGELOG.md @@ -1,5 +1,14 @@ # @atmo-dev/contrail +## 0.6.0 + +### Minor Changes + +- af24714: Add per-collection `recordFilter` and apply Jetstream `#identity` handle changes during ingest. + + - `CollectionConfig.recordFilter?: (record) => boolean` runs against each create/update during ingest; returning false drops the record before it reaches the DB. Useful for narrowing high-volume collections to just the records you care about (e.g. only `app.bsky.feed.post` records mentioning a particular URL). Deletes are not filtered, so they still tear down any record the filter previously let through. Throws are caught, logged, and treated as drops. + - Jetstream `#identity` events (handle changes) now flow through to the `identities` table via a new `applyIdentityEvent` helper. UPDATE-only — unknown DIDs are no-ops so we don't materialize partial rows lacking PDS. + ## 0.5.0 ### Minor Changes diff --git a/packages/contrail/package.json b/packages/contrail/package.json index 49123de..3245e17 100644 --- a/packages/contrail/package.json +++ b/packages/contrail/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail", - "version": "0.5.0", + "version": "0.6.0", "description": "Index AT Protocol records with typed XRPC endpoints. Cloudflare Workers + D1, SvelteKit, Node.js.", "type": "module", "sideEffects": false, diff --git a/packages/lexicons/CHANGELOG.md b/packages/lexicons/CHANGELOG.md index 9a1c5d1..d5f674f 100644 --- a/packages/lexicons/CHANGELOG.md +++ b/packages/lexicons/CHANGELOG.md @@ -1,5 +1,12 @@ # @atmo-dev/contrail-lexicons +## 0.4.6 + +### Patch Changes + +- Updated dependencies [af24714] + - @atmo-dev/contrail@0.6.0 + ## 0.4.5 ### Patch Changes diff --git a/packages/lexicons/package.json b/packages/lexicons/package.json index 7a5dab9..9cc3ea4 100644 --- a/packages/lexicons/package.json +++ b/packages/lexicons/package.json @@ -1,6 +1,6 @@ { "name": "@atmo-dev/contrail-lexicons", - "version": "0.4.5", + "version": "0.4.6", "description": "Generate atproto lexicon JSON (and optionally TypeScript types via @atcute/lex-cli) from a Contrail config.", "type": "module", "files": [ -- 2.51.2