merge: resolve conflicts with main into feat/another-big-refactor · flo-bit.dev/contrail@0a7cc54

+99 -8

packages/contrail-appview/src/core/backfill.ts

··· 4 4 5 5 import type { Client } from "@atcute/client"; 6 6 import type { ContrailConfig, Database, IngestEvent } from "./types"; 7 - import { getDiscoverableNsids, getDependentNsids, DEFAULT_RELAYS } from "./types"; 7 + import { 8 + getDiscoverableNsids, 9 + getDependentNsids, 10 + DEFAULT_RELAYS, 11 + shortNameForNsid, 12 + } from "./types"; 8 13 import { applyEvents, getLastCursor, saveCursor } from "./db"; 9 14 import { getClient, getPDS } from "./client"; 10 15 16 + const DEFAULT_TIME_FIELD = "createdAt"; 17 + 18 + /** Parse the record's canonical time (e.g. createdAt) and return microseconds. 19 + * Falls back to `nowUs` when missing/invalid. Clamps to nowUs to avoid 20 + * user-controlled future timestamps pinning records at the top of feeds. */ 21 + function recordTimeUs( 22 + record: unknown, 23 + collection: string, 24 + config: ContrailConfig | undefined, 25 + nowUs: number 26 + ): number { 27 + if (!config) return nowUs; 28 + const short = shortNameForNsid(config, collection); 29 + const colCfg = short ? config.collections[short] : undefined; 30 + const field = colCfg?.timeField ?? DEFAULT_TIME_FIELD; 31 + if (field === false) return nowUs; 32 + const raw = 33 + record && typeof record === "object" 34 + ? (record as Record<string, unknown>)[field] 35 + : undefined; 36 + if (typeof raw !== "string") return nowUs; 37 + const ms = Date.parse(raw); 38 + if (!Number.isFinite(ms) || ms <= 0) return nowUs; 39 + const us = ms * 1000; 40 + return us > nowUs ? nowUs : us; 41 + } 42 + 11 43 const PAGE_SIZE = 100; 12 44 const BATCH_SIZE = 100; 13 45 const MAX_RETRIES = 5; ··· 40 72 throw lastError; 41 73 } 42 74 75 + /** Drop events whose `subjectField` value is a DID we have no identity for. 76 + * One bulk SELECT per call, suitable for use after each backfill page. */ 77 + async function filterEventsBySubject( 78 + db: Database, 79 + events: IngestEvent[], 80 + subjectField: string 81 + ): Promise<IngestEvent[]> { 82 + const subjects = new Set<string>(); 83 + const eventSubjects = new Map<string, string>(); 84 + for (const e of events) { 85 + if (!e.record) continue; 86 + let subj: unknown; 87 + try { 88 + subj = JSON.parse(e.record)?.[subjectField]; 89 + } catch { 90 + continue; 91 + } 92 + if (typeof subj === "string" && isDid(subj)) { 93 + subjects.add(subj); 94 + eventSubjects.set(e.uri, subj); 95 + } 96 + } 97 + if (subjects.size === 0) return []; 98 + 99 + const known = new Set<string>(); 100 + const list = [...subjects]; 101 + const CHUNK = 100; 102 + for (let i = 0; i < list.length; i += CHUNK) { 103 + const chunk = list.slice(i, i + CHUNK); 104 + const placeholders = chunk.map(() => "?").join(","); 105 + const rows = await db 106 + .prepare(`SELECT did FROM identities WHERE did IN (${placeholders})`) 107 + .bind(...chunk) 108 + .all<{ did: string }>(); 109 + for (const r of rows.results ?? []) known.add(r.did); 110 + } 111 + 112 + return events.filter((e) => { 113 + const subj = eventSubjects.get(e.uri); 114 + return subj !== undefined && known.has(subj); 115 + }); 116 + } 117 + 43 118 async function markFailed( 44 119 db: Database, 45 120 did: string, ··· 125 200 let totalInserted = 0; 126 201 let done = false; 127 202 203 + // Lookup subject filter once: if this collection declares a subjectField, we 204 + // drop records whose subject DID isn't already in our identities table. 205 + const collectionShort = config 206 + ? shortNameForNsid(config, collection) 207 + : undefined; 208 + const subjectField = collectionShort 209 + ? config?.collections[collectionShort]?.subjectField 210 + : undefined; 211 + 128 212 try { 129 213 while (Date.now() < deadline) { 130 214 const response = await withRetry( ··· 157 241 } 158 242 159 243 const now = Date.now(); 160 - const events: IngestEvent[] = response.data.records.map((r) => ({ 244 + const nowUs = now * 1000; 245 + let events: IngestEvent[] = response.data.records.map((r) => ({ 161 246 uri: r.uri, 162 247 did, 163 248 collection, ··· 165 250 operation: "create" as const, 166 251 cid: r.cid, 167 252 record: JSON.stringify(r.value), 168 - time_us: now * 1000, 169 - indexed_at: now * 1000, 253 + time_us: recordTimeUs(r.value, collection, config, nowUs), 254 + indexed_at: nowUs, 170 255 })); 171 256 172 - await applyEvents(db, events, config, { 173 - skipReplayDetection: options?.skipReplayDetection, 174 - skipFeedFanout: true, 175 - }); 257 + if (subjectField) { 258 + events = await filterEventsBySubject(db, events, subjectField); 259 + } 260 + 261 + if (events.length > 0) { 262 + await applyEvents(db, events, config, { 263 + skipReplayDetection: options?.skipReplayDetection, 264 + skipFeedFanout: true, 265 + }); 266 + } 176 267 totalInserted += events.length; 177 268 178 269 currentCursor = response.data.cursor ?? undefined;

+181

packages/contrail-appview/src/core/constellation.ts

··· 1 + import { isDid } from "@atcute/lexicons/syntax"; 2 + import type { ContrailConfig, Database, Logger } from "./types"; 3 + import { 4 + DEFAULT_CONSTELLATION_URL, 5 + DEFAULT_FOLLOW_NSID, 6 + recordsTableName, 7 + shortNameForNsid, 8 + } from "./types"; 9 + 10 + const PAGE_LIMIT = 100; 11 + const DID_FILTER_CHUNK = 50; 12 + 13 + interface BacklinksPage { 14 + links?: Array<{ 15 + did?: string; 16 + rkey?: string; 17 + /** Some Constellation versions return the full URI rather than did/rkey split. */ 18 + uri?: string; 19 + }>; 20 + cursor?: string; 21 + } 22 + 23 + function getLogger(config: ContrailConfig): Logger { 24 + return config.logger ?? console; 25 + } 26 + 27 + /** Resolve effective Constellation config; null when disabled. */ 28 + function getConstellationSettings( 29 + config: ContrailConfig 30 + ): { url: string; userAgent: string } | null { 31 + const c = config.constellation; 32 + if (c === false) return null; 33 + if (c?.enabled === false) return null; 34 + return { 35 + url: c?.url ?? DEFAULT_CONSTELLATION_URL, 36 + userAgent: c?.userAgent ?? `contrail/${config.namespace}`, 37 + }; 38 + } 39 + 40 + /** Find the configured short name for `app.bsky.graph.follow`, if any. */ 41 + function getFollowShort(config: ContrailConfig): string | null { 42 + const short = shortNameForNsid(config, DEFAULT_FOLLOW_NSID); 43 + return short ?? null; 44 + } 45 + 46 + interface BacklinkRow { 47 + did: string; 48 + rkey: string; 49 + uri: string; 50 + } 51 + 52 + function parseBacklink(entry: NonNullable<BacklinksPage["links"]>[number]): BacklinkRow | null { 53 + if (entry.did && entry.rkey && isDid(entry.did)) { 54 + return { 55 + did: entry.did, 56 + rkey: entry.rkey, 57 + uri: entry.uri ?? `at://${entry.did}/${DEFAULT_FOLLOW_NSID}/${entry.rkey}`, 58 + }; 59 + } 60 + if (entry.uri) { 61 + const m = /^at:\/\/(did:[^/]+)\/[^/]+\/([^/]+)$/.exec(entry.uri); 62 + if (m && isDid(m[1])) { 63 + return { did: m[1], rkey: m[2], uri: entry.uri }; 64 + } 65 + } 66 + return null; 67 + } 68 + 69 + /** Filter a candidate-follower DID list to those already in our identities table. */ 70 + async function filterKnownDids( 71 + db: Database, 72 + candidates: string[] 73 + ): Promise<Set<string>> { 74 + const known = new Set<string>(); 75 + for (let i = 0; i < candidates.length; i += DID_FILTER_CHUNK) { 76 + const chunk = candidates.slice(i, i + DID_FILTER_CHUNK); 77 + const placeholders = chunk.map(() => "?").join(","); 78 + const rows = await db 79 + .prepare(`SELECT did FROM identities WHERE did IN (${placeholders})`) 80 + .bind(...chunk) 81 + .all<{ did: string }>(); 82 + for (const r of rows.results ?? []) known.add(r.did); 83 + } 84 + return known; 85 + } 86 + 87 + /** Fetch one page of getBacklinks. Returns null on non-2xx (caller decides whether to bail). */ 88 + async function fetchBacklinksPage( 89 + url: string, 90 + userAgent: string, 91 + subject: string, 92 + cursor?: string 93 + ): Promise<BacklinksPage | null> { 94 + const u = new URL("/xrpc/blue.microcosm.links.getBacklinks", url); 95 + u.searchParams.set("subject", subject); 96 + u.searchParams.set("source", `${DEFAULT_FOLLOW_NSID}:.subject`); 97 + u.searchParams.set("limit", String(PAGE_LIMIT)); 98 + if (cursor) u.searchParams.set("cursor", cursor); 99 + try { 100 + const res = await fetch(u.toString(), { 101 + headers: { "user-agent": userAgent, accept: "application/json" }, 102 + }); 103 + if (!res.ok) return null; 104 + return (await res.json()) as BacklinksPage; 105 + } catch { 106 + return null; 107 + } 108 + } 109 + 110 + /** For a newly-known subject DID, find existing followers via Constellation 111 + * and ingest synthesized follow records into the configured follow table. 112 + * Best-effort: failures are logged but not retried (caller can re-trigger). */ 113 + export async function backfillFollowersFromConstellation( 114 + db: Database, 115 + config: ContrailConfig, 116 + subjectDid: string 117 + ): Promise<number> { 118 + const settings = getConstellationSettings(config); 119 + if (!settings) return 0; 120 + if (!isDid(subjectDid)) return 0; 121 + const followShort = getFollowShort(config); 122 + if (!followShort) return 0; 123 + const log = getLogger(config); 124 + 125 + const followTable = recordsTableName(followShort); 126 + const recordJson = JSON.stringify({ 127 + $type: DEFAULT_FOLLOW_NSID, 128 + subject: subjectDid, 129 + createdAt: new Date().toISOString(), 130 + }); 131 + const nowUs = Date.now() * 1000; 132 + 133 + let cursor: string | undefined; 134 + let inserted = 0; 135 + let pages = 0; 136 + 137 + while (true) { 138 + const page = await fetchBacklinksPage( 139 + settings.url, 140 + settings.userAgent, 141 + subjectDid, 142 + cursor 143 + ); 144 + if (!page) break; 145 + pages++; 146 + 147 + const rows: BacklinkRow[] = (page.links ?? []) 148 + .map(parseBacklink) 149 + .filter((r): r is BacklinkRow => r !== null && r.did !== subjectDid); 150 + 151 + if (rows.length > 0) { 152 + const known = await filterKnownDids( 153 + db, 154 + rows.map((r) => r.did) 155 + ); 156 + const survivors = rows.filter((r) => known.has(r.did)); 157 + 158 + for (const r of survivors) { 159 + const result = await db 160 + .prepare( 161 + `INSERT INTO ${followTable} (uri, did, rkey, cid, record, time_us, indexed_at) 162 + VALUES (?, ?, ?, NULL, ?, ?, ?) 163 + ON CONFLICT(uri) DO NOTHING` 164 + ) 165 + .bind(r.uri, r.did, r.rkey, recordJson, nowUs, nowUs) 166 + .run(); 167 + inserted += (result as { changes?: number })?.changes ?? 0; 168 + } 169 + } 170 + 171 + cursor = page.cursor ?? undefined; 172 + if (!cursor) break; 173 + } 174 + 175 + if (inserted > 0) { 176 + log.log( 177 + `[constellation] subject=${subjectDid} pages=${pages} inserted=${inserted}` 178 + ); 179 + } 180 + return inserted; 181 + }

+54 -22

packages/contrail-appview/src/core/db/records.ts

··· 18 18 spacesRecordsTableName, 19 19 shortNameForNsid, 20 20 nsidForShortName, 21 + normalizeFeedTarget, 22 + feedTargetMaxItems, 23 + DEFAULT_FOLLOW_SHORT, 21 24 } from "../types"; 22 25 import { getSearchableFields, ftsTableName, buildFtsContent } from "../search"; 23 26 import { ftsQueryClause, getDialect } from "../dialect"; ··· 210 213 if (!eventShort) return []; 211 214 212 215 for (const [, feedConfig] of Object.entries(config.feeds)) { 213 - const followTable = recordsTableName(feedConfig.follow); 216 + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; 217 + const followTable = recordsTableName(followShort); 218 + const targets = feedConfig.targets.map(normalizeFeedTarget); 219 + const targetShorts = targets.map((t) => t.collection); 214 220 215 221 // Target collection: fan out to followers 216 - if (feedConfig.targets.includes(eventShort)) { 222 + if (targetShorts.includes(eventShort)) { 217 223 if (event.operation === "create" || event.operation === "update") { 218 224 stmts.push( 219 225 db ··· 235 241 } 236 242 237 243 // Follow collection: handle follow/unfollow 238 - if (eventShort === feedConfig.follow) { 244 + if (eventShort === followShort) { 239 245 if (event.operation === "create") { 240 246 const record = event.record ? JSON.parse(event.record) : null; 241 247 const subject = record?.subject; 242 248 if (subject) { 243 - for (const targetShort of feedConfig.targets) { 244 - const targetTable = recordsTableName(targetShort); 245 - const targetNsid = nsidForShortName(config, targetShort) ?? targetShort; 249 + for (const target of targets) { 250 + const targetTable = recordsTableName(target.collection); 251 + const targetNsid = nsidForShortName(config, target.collection) ?? target.collection; 252 + const cap = feedTargetMaxItems(feedConfig, target); 246 253 stmts.push( 247 254 db 248 255 .prepare( ··· 252 259 FROM ${targetTable} r 253 260 WHERE r.did = ? 254 261 ORDER BY r.time_us DESC 255 - LIMIT 100` 262 + LIMIT ${cap}` 256 263 ) 257 264 ) 258 265 .bind(event.did, targetNsid, subject) ··· 265 272 const parsed = JSON.parse(existingRecord); 266 273 const subject = parsed?.subject; 267 274 if (subject) { 268 - for (const targetShort of feedConfig.targets) { 269 - const targetTable = recordsTableName(targetShort); 275 + for (const target of targets) { 276 + const targetTable = recordsTableName(target.collection); 270 277 stmts.push( 271 278 db 272 279 .prepare( ··· 288 295 289 296 // --- Feed pruning --- 290 297 298 + /** Prune feed_items per (actor, collection) to the given cap. 299 + * 300 + * - If `caps` is a number: legacy behavior — global per-actor cap across all collections. 301 + * - If `caps` is a Map<collection-NSID, cap>: each collection is pruned independently per actor, 302 + * so high-volume collections (e.g. RSVPs) can't squeeze out lower-volume ones (e.g. events). 303 + * Collections not present in the map are left alone. 304 + */ 291 305 export async function pruneFeedItems( 292 306 db: Database, 293 - maxItems: number 307 + caps: number | Map<string, number> 294 308 ): Promise<number> { 295 - const result = await db 296 - .prepare( 297 - `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( 298 - SELECT actor, uri FROM ( 299 - SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn 300 - FROM feed_items 301 - ) sub WHERE rn <= ? 302 - )` 303 - ) 304 - .bind(maxItems) 305 - .run(); 306 - return (result as any)?.changes ?? 0; 309 + if (typeof caps === "number") { 310 + const result = await db 311 + .prepare( 312 + `DELETE FROM feed_items WHERE (actor, uri) NOT IN ( 313 + SELECT actor, uri FROM ( 314 + SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn 315 + FROM feed_items 316 + ) sub WHERE rn <= ? 317 + )` 318 + ) 319 + .bind(caps) 320 + .run(); 321 + return (result as any)?.changes ?? 0; 322 + } 323 + let total = 0; 324 + for (const [collection, cap] of caps) { 325 + const result = await db 326 + .prepare( 327 + `DELETE FROM feed_items WHERE collection = ? AND (actor, uri) NOT IN ( 328 + SELECT actor, uri FROM ( 329 + SELECT actor, uri, ROW_NUMBER() OVER (PARTITION BY actor ORDER BY time_us DESC) as rn 330 + FROM feed_items WHERE collection = ? 331 + ) sub WHERE rn <= ? 332 + )` 333 + ) 334 + .bind(collection, collection, cap) 335 + .run(); 336 + total += (result as any)?.changes ?? 0; 337 + } 338 + return total; 307 339 } 308 340 309 341 // --- Cursor ---

+9 -1

packages/contrail-appview/src/core/db/schema.ts

··· 213 213 actor TEXT NOT NULL, 214 214 feed TEXT NOT NULL, 215 215 completed INTEGER NOT NULL DEFAULT 0, 216 + retries INTEGER NOT NULL DEFAULT 0, 217 + last_error TEXT, 218 + started_at ${dialect.bigintType}, 216 219 PRIMARY KEY (actor, feed) 217 220 )`, 218 221 ]; 219 222 220 - const followCollections = new Set(Object.values(config.feeds).map((f) => f.follow)); 223 + const followCollections = new Set( 224 + Object.values(config.feeds).map((f) => f.follow ?? "follow") 225 + ); 221 226 for (const col of followCollections) { 222 227 const table = recordsTableName(col); 223 228 const safe = sanitizeName(col); ··· 249 254 "ALTER TABLE backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", 250 255 "ALTER TABLE backfills ADD COLUMN last_error TEXT", 251 256 "ALTER TABLE spaces_invites ADD COLUMN kind TEXT NOT NULL DEFAULT 'join'", 257 + "ALTER TABLE feed_backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", 258 + "ALTER TABLE feed_backfills ADD COLUMN last_error TEXT", 259 + "ALTER TABLE feed_backfills ADD COLUMN started_at BIGINT", 252 260 ]; 253 261 254 262 async function runMigrations(db: Database): Promise<void> {

+83 -12

packages/contrail-appview/src/core/jetstream.ts

··· 1 1 import { JetstreamSubscription } from "@atcute/jetstream"; 2 2 import type { ContrailConfig, IngestEvent, Database, Logger } from "./types"; 3 - import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS } from "./types"; 3 + import { 4 + getCollectionNsids, 5 + getDependentNsids, 6 + shortNameForNsid, 7 + buildFeedTargetCaps, 8 + } from "./types"; 4 9 import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; 5 - import { refreshStaleIdentities } from "./identity"; 10 + import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; 11 + import { backfillFollowersFromConstellation } from "./constellation"; 6 12 7 13 const BATCH_SIZE = 50; 8 14 const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; // 1 hour ··· 27 33 cursor: number | null, 28 34 safetyTimeoutMs: number = 25_000, 29 35 knownDids?: Set<string> 30 - ): Promise<{ events: IngestEvent[]; lastCursor: number | null }> { 36 + ): Promise<{ 37 + events: IngestEvent[]; 38 + lastCursor: number | null; 39 + newlyKnownDids: string[]; 40 + identityUpdates: Map<string, string>; 41 + }> { 31 42 const log = getLogger(config); 32 43 const startTimeUs = Date.now() * 1000; 33 44 const deadline = Date.now() + safetyTimeoutMs; ··· 45 56 let connectCount = 0; 46 57 const seenUris = new Map<string, number>(); // uri -> time_us of first occurrence 47 58 const duplicateUris: string[] = []; 59 + const newlyKnownDids = new Set<string>(); 60 + const identityUpdates = new Map<string, string>(); 48 61 49 62 const subscription = new JetstreamSubscription({ 50 63 url: urls, ··· 75 88 76 89 const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`; 77 90 91 + const short = shortNameForNsid(config, commit.collection); 92 + const collectionCfg = short ? config.collections[short] : undefined; 93 + 78 94 if (dependentCollections.has(commit.collection) && knownDids) { 79 95 if (!knownDids.has(event.did)) { 80 96 filteredUnknownDid++; 81 97 if (filteredDidSamples.size < 10) filteredDidSamples.add(event.did); 82 98 continue; 83 99 } 100 + // Subject filter: for collections with subjectField (e.g. follows 101 + // pointing at a `subject` DID), drop records whose subject isn't a 102 + // DID we care about. Trims network-wide social graph to the 103 + // subjects our discoverable users overlap with. 104 + const subjectField = collectionCfg?.subjectField; 105 + if (subjectField && commit.operation !== "delete") { 106 + const subj = (commit.record as Record<string, unknown> | undefined)?.[ 107 + subjectField 108 + ]; 109 + if (typeof subj === "string" && !knownDids.has(subj)) { 110 + continue; 111 + } 112 + } 113 + } 114 + 115 + if (collectionCfg?.recordFilter && commit.operation !== "delete") { 116 + const rec = commit.record as Record<string, unknown> | undefined; 117 + let keep = false; 118 + try { 119 + keep = !!(rec && collectionCfg.recordFilter(rec)); 120 + } catch (err) { 121 + log.warn(`[ingest] recordFilter threw for ${uri}: ${err}`); 122 + } 123 + if (!keep) continue; 84 124 } 85 125 86 126 const prev = seenUris.get(uri); ··· 115 155 ); 116 156 117 157 if (knownDids && !dependentCollections.has(commit.collection)) { 118 - knownDids.add(event.did); 158 + if (!knownDids.has(event.did)) { 159 + knownDids.add(event.did); 160 + newlyKnownDids.add(event.did); 161 + } 119 162 } 163 + } else if (event.kind === "identity") { 164 + identityUpdates.set(event.did, event.identity.handle); 120 165 } 121 166 122 167 if (event.time_us >= startTimeUs) { ··· 172 217 ); 173 218 } 174 219 175 - return { events: collected, lastCursor }; 220 + return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids], identityUpdates }; 176 221 } 177 222 178 223 // Run a full ingest cycle: init schema, load cursor, ingest, apply, save cursor ··· 220 265 } 221 266 } 222 267 223 - const { events, lastCursor } = await ingestEvents( 268 + const { events, lastCursor, newlyKnownDids, identityUpdates } = await ingestEvents( 224 269 config, 225 270 cursor, 226 271 timeoutMs, ··· 245 290 await applyEvents(db, batch, config, { pubsub }); 246 291 } 247 292 293 + // Apply handle changes from #identity events. UPDATE-only, so unknown 294 + // DIDs are no-ops — we don't want to create partial rows lacking PDS. 295 + if (identityUpdates.size > 0) { 296 + for (const [did, handle] of identityUpdates) { 297 + try { 298 + await applyIdentityEvent(db, did, handle); 299 + } catch (err) { 300 + log.warn(`[ingest] identity update failed for ${did}: ${err}`); 301 + } 302 + } 303 + log.log(`[ingest] applied ${identityUpdates.size} identity event(s)`); 304 + } 305 + 248 306 // Refresh stale/missing identities for DIDs in this batch 249 307 const uniqueDids = [...new Set(events.map((e) => e.did))]; 250 308 if (uniqueDids.length > 0) { ··· 266 324 log.log(`[ingest] no cursor returned from subscription; not saving`); 267 325 } 268 326 269 - // Prune feed items hourly 327 + // Newly-discovered DIDs: ask Constellation for back-edges so they 328 + // immediately appear in existing followers' feeds (best-effort, opt-out). 329 + if (config.feeds && newlyKnownDids.length > 0) { 330 + for (const subj of newlyKnownDids) { 331 + try { 332 + await backfillFollowersFromConstellation(db, config, subj); 333 + } catch (err) { 334 + log.warn(`[constellation] subject=${subj} failed: ${err}`); 335 + } 336 + } 337 + } 338 + 339 + // Prune feed items hourly, per-target so high-volume targets don't 340 + // squeeze out lower-volume ones. 270 341 if (config.feeds && Date.now() - s.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { 271 - const maxItems = Math.max( 272 - ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS) 273 - ); 274 - const pruned = await pruneFeedItems(db, maxItems); 275 - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 342 + const caps = buildFeedTargetCaps(config); 343 + if (caps.size > 0) { 344 + const pruned = await pruneFeedItems(db, caps); 345 + if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 346 + } 276 347 s.lastFeedPruneMs = Date.now(); 277 348 } 278 349

+63 -8

packages/contrail-appview/src/core/persistent.ts

··· 1 1 import type { JetstreamSubscription } from "@atcute/jetstream"; 2 2 import type { ContrailConfig, IngestEvent, Database, Logger, ResolvedContrailConfig } from "./types"; 3 - import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS, resolveConfig } from "./types"; 3 + import { 4 + getCollectionNsids, 5 + getDependentNsids, 6 + buildFeedTargetCaps, 7 + resolveConfig, 8 + shortNameForNsid, 9 + } from "./types"; 4 10 import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; 5 - import { refreshStaleIdentities } from "./identity"; 11 + import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; 12 + import { backfillFollowersFromConstellation } from "./constellation"; 6 13 import { createIngestState } from "./jetstream"; 7 14 import type { IngestState } from "./jetstream"; 8 15 ··· 75 82 collections, 76 83 dependentCollections, 77 84 knownDids, 85 + newlyKnownDids: new Set<string>(), 78 86 state, 79 87 log, 80 88 createSubscription: options?.createSubscription, ··· 101 109 collections: string[]; 102 110 dependentCollections: Set<string>; 103 111 knownDids?: Set<string>; 112 + /** DIDs that crossed from unknown→known during this stream's lifetime. 113 + * Drained on each flush so Constellation reverse-lookups can run for them. */ 114 + newlyKnownDids?: Set<string>; 104 115 state: IngestState; 105 116 log: Logger; 106 117 createSubscription?: (cursor: number | null) => any; ··· 152 163 } 153 164 } 154 165 166 + // Drain newly-known DIDs and ask Constellation for back-edges. 167 + if (config.feeds && opts.newlyKnownDids && opts.newlyKnownDids.size > 0) { 168 + const drained = [...opts.newlyKnownDids]; 169 + opts.newlyKnownDids.clear(); 170 + for (const subj of drained) { 171 + try { 172 + await backfillFollowersFromConstellation(db, config, subj); 173 + } catch (err) { 174 + log.warn(`[constellation] subject=${subj} failed: ${err}`); 175 + } 176 + } 177 + } 178 + 155 179 if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { 156 - const maxItems = Math.max( 157 - ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS) 158 - ); 159 - const pruned = await pruneFeedItems(db, maxItems); 160 - if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 180 + const caps = buildFeedTargetCaps(config); 181 + if (caps.size > 0) { 182 + const pruned = await pruneFeedItems(db, caps); 183 + if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); 184 + } 161 185 state.lastFeedPruneMs = Date.now(); 162 186 } 163 187 ··· 206 230 if (event.kind === "commit") { 207 231 const { commit } = event; 208 232 233 + const short = shortNameForNsid(config, commit.collection); 234 + const collectionCfg = short ? config.collections[short] : undefined; 235 + 209 236 if (dependentCollections.has(commit.collection) && knownDids) { 210 237 if (!knownDids.has(event.did)) continue; 238 + // Subject filter: skip records whose subject DID isn't known. 239 + const subjectField = collectionCfg?.subjectField; 240 + if (subjectField && commit.operation !== "delete") { 241 + const subj = (commit.record as Record<string, unknown> | undefined)?.[ 242 + subjectField 243 + ]; 244 + if (typeof subj === "string" && !knownDids.has(subj)) continue; 245 + } 246 + } 247 + 248 + if (collectionCfg?.recordFilter && commit.operation !== "delete") { 249 + const rec = commit.record as Record<string, unknown> | undefined; 250 + let keep = false; 251 + try { 252 + keep = !!(rec && collectionCfg.recordFilter(rec)); 253 + } catch (err) { 254 + log.warn(`recordFilter threw for ${commit.collection}/${commit.rkey}: ${err}`); 255 + } 256 + if (!keep) continue; 211 257 } 212 258 213 259 const now = Date.now(); ··· 226 272 }); 227 273 228 274 if (knownDids && !dependentCollections.has(commit.collection)) { 229 - knownDids.add(event.did); 275 + if (!knownDids.has(event.did)) { 276 + knownDids.add(event.did); 277 + opts.newlyKnownDids?.add(event.did); 278 + } 279 + } 280 + } else if (event.kind === "identity") { 281 + try { 282 + await applyIdentityEvent(db, event.did, event.identity.handle); 283 + } catch (err) { 284 + log.warn(`Identity update failed for ${event.did}: ${err}`); 230 285 } 231 286 } 232 287

+220 -55

packages/contrail-appview/src/core/router/feed.ts

··· 1 - import type { Hono } from "hono"; 2 - import type { ContrailConfig, Database, FeedConfig } from "../types"; 1 + import type { Context, Hono } from "hono"; 2 + import type { 3 + ContrailConfig, 4 + Database, 5 + FeedConfig, 6 + FeedTargetConfig, 7 + } from "../types"; 3 8 import { getDialect } from "../dialect"; 4 - import { DEFAULT_FEED_MAX_ITEMS, recordsTableName } from "../types"; 9 + import { 10 + DEFAULT_FOLLOW_SHORT, 11 + feedTargetMaxItems, 12 + normalizeFeedTarget, 13 + recordsTableName, 14 + shortNameForNsid, 15 + } from "../types"; 5 16 import { resolveActor } from "../identity"; 6 17 import { backfillUser } from "../backfill"; 7 18 import { runPipeline } from "./collection"; 8 19 20 + const BACKFILL_TIMEOUT_MS = 30_000; 21 + const BACKFILL_REQUEST_TIMEOUT_MS = 10_000; 22 + const BACKFILL_MAX_RETRIES = 3; 23 + /** Re-arm a stuck in-progress row after this long (covers process crashes mid-backfill). */ 24 + const BACKFILL_STALE_MS = 5 * 60 * 1000; 25 + 26 + interface FeedBackfillStatus { 27 + completed: number; 28 + retries: number; 29 + last_error: string | null; 30 + started_at: number | null; 31 + } 32 + 33 + /** Schedule async work, preferring waitUntil on Cloudflare Workers so the 34 + * runtime keeps the request alive until the promise settles. Falls back to 35 + * fire-and-forget with a logged catch. */ 36 + function scheduleBackground( 37 + c: Context, 38 + config: ContrailConfig, 39 + task: () => Promise<unknown> 40 + ): void { 41 + const log = config.logger ?? console; 42 + const promise = task().catch((err) => 43 + log.error(`[feed] background task failed: ${err}`) 44 + ); 45 + try { 46 + c.executionCtx.waitUntil(promise); 47 + } catch { 48 + // No executionCtx (Node/Bun); promise runs detached. 49 + } 50 + } 51 + 52 + /** Run the bootstrap copy + per-target prune. Returns rows inserted. */ 53 + async function bootstrapFeedItems( 54 + db: Database, 55 + config: ContrailConfig, 56 + actor: string, 57 + feedConfig: FeedConfig 58 + ): Promise<number> { 59 + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; 60 + const followTable = recordsTableName(followShort); 61 + const targets = feedConfig.targets.map(normalizeFeedTarget); 62 + let totalInserted = 0; 63 + 64 + for (const target of targets) { 65 + const targetTable = recordsTableName(target.collection); 66 + const targetCfg = config.collections[target.collection]; 67 + if (!targetCfg) continue; 68 + const cap = feedTargetMaxItems(feedConfig, target); 69 + 70 + const insert = await db 71 + .prepare( 72 + getDialect(db).insertOrIgnore( 73 + `INSERT INTO feed_items (actor, uri, collection, time_us) 74 + SELECT ?, r.uri, ?, r.time_us 75 + FROM ${targetTable} r 76 + WHERE r.did IN ( 77 + SELECT ${getDialect(db).jsonExtract("f.record", "subject")} 78 + FROM ${followTable} f 79 + WHERE f.did = ? 80 + ) 81 + ORDER BY r.time_us DESC 82 + LIMIT ${cap}` 83 + ) 84 + ) 85 + .bind(actor, targetCfg.collection, actor) 86 + .run(); 87 + totalInserted += (insert as { changes?: number })?.changes ?? 0; 88 + 89 + // Per-target prune so high-volume targets don't squeeze out lower-volume ones. 90 + await db 91 + .prepare( 92 + `DELETE FROM feed_items WHERE actor = ? AND collection = ? AND uri NOT IN ( 93 + SELECT uri FROM feed_items WHERE actor = ? AND collection = ? 94 + ORDER BY time_us DESC LIMIT ? 95 + )` 96 + ) 97 + .bind(actor, targetCfg.collection, actor, targetCfg.collection, cap) 98 + .run(); 99 + } 100 + 101 + return totalInserted; 102 + } 103 + 104 + /** Run a full backfill cycle: walk follow records → bootstrap feed_items → mark complete. 105 + * Updates feed_backfills row with retries/last_error on failure. */ 106 + async function runFeedBackfill( 107 + db: Database, 108 + config: ContrailConfig, 109 + actor: string, 110 + feedName: string, 111 + feedConfig: FeedConfig 112 + ): Promise<void> { 113 + const followShort = feedConfig.follow ?? DEFAULT_FOLLOW_SHORT; 114 + const followCfg = config.collections[followShort]; 115 + if (!followCfg) return; 116 + 117 + try { 118 + const inserted = await backfillUser( 119 + db, 120 + actor, 121 + followCfg.collection, 122 + Date.now() + BACKFILL_TIMEOUT_MS, 123 + config, 124 + { 125 + skipReplayDetection: true, 126 + maxRetries: BACKFILL_MAX_RETRIES, 127 + requestTimeout: BACKFILL_REQUEST_TIMEOUT_MS, 128 + } 129 + ); 130 + 131 + // Bootstrap from whatever follow records we now have (may be from this 132 + // backfill, from earlier live ingest, or both). 133 + await bootstrapFeedItems(db, config, actor, feedConfig); 134 + 135 + // Only mark complete if the underlying follow backfill actually finished 136 + // (backfills.completed = 1). Avoids the old bug where timeouts/empty 137 + // walks would lock the user out of any retry. 138 + const followStatus = await db 139 + .prepare( 140 + "SELECT completed FROM backfills WHERE did = ? AND collection = ?" 141 + ) 142 + .bind(actor, followCfg.collection) 143 + .first<{ completed: number }>(); 144 + 145 + if (followStatus?.completed) { 146 + await db 147 + .prepare( 148 + "UPDATE feed_backfills SET completed = 1, last_error = NULL WHERE actor = ? AND feed = ?" 149 + ) 150 + .bind(actor, feedName) 151 + .run(); 152 + } else { 153 + // Walk didn't complete (timeout/error), but record the partial progress 154 + // so the next request retries. 155 + await db 156 + .prepare( 157 + "UPDATE feed_backfills SET retries = retries + 1, started_at = NULL, last_error = ? WHERE actor = ? AND feed = ?" 158 + ) 159 + .bind( 160 + `follow backfill incomplete (inserted=${inserted})`, 161 + actor, 162 + feedName 163 + ) 164 + .run(); 165 + } 166 + } catch (err) { 167 + await db 168 + .prepare( 169 + "UPDATE feed_backfills SET retries = retries + 1, started_at = NULL, last_error = ? WHERE actor = ? AND feed = ?" 170 + ) 171 + .bind(String(err), actor, feedName) 172 + .run(); 173 + } 174 + } 175 + 176 + /** Decide whether to (re)kick off a background backfill, and do so if needed. 177 + * Always returns immediately so the request path stays cheap. */ 9 178 async function maybeBackfillFeed( 179 + c: Context, 10 180 db: Database, 11 181 config: ContrailConfig, 12 182 actor: string, ··· 14 184 feedConfig: FeedConfig 15 185 ): Promise<void> { 16 186 const status = await db 17 - .prepare("SELECT completed FROM feed_backfills WHERE actor = ? AND feed = ?") 187 + .prepare( 188 + "SELECT completed, retries, last_error, started_at FROM feed_backfills WHERE actor = ? AND feed = ?" 189 + ) 18 190 .bind(actor, feedName) 19 - .first<{ completed: number }>(); 191 + .first<FeedBackfillStatus>(); 20 192 21 193 if (status?.completed) return; 22 194 23 - // Ensure the user's follow records are backfilled first 24 - await backfillUser(db, actor, feedConfig.follow, Date.now() + 3_000, config, { 25 - maxRetries: 0, 26 - requestTimeout: 3_000, 27 - }); 195 + const now = Date.now(); 28 196 29 - // Mark as in-progress (idempotent) 30 - await db 31 - .prepare( 32 - "INSERT INTO feed_backfills (actor, feed, completed) VALUES (?, ?, 0) ON CONFLICT DO NOTHING" 33 - ) 34 - .bind(actor, feedName) 35 - .run(); 197 + // Skip if a backfill is already in flight (started_at recently set) — avoids 198 + // duplicate work from concurrent requests for the same actor. 199 + if (status?.started_at && now - status.started_at < BACKFILL_STALE_MS) return; 36 200 37 - const maxItems = feedConfig.maxItems ?? DEFAULT_FEED_MAX_ITEMS; 38 - 39 - // Populate feed from existing records by followed users 40 - const followTable = recordsTableName(feedConfig.follow); 41 - for (const targetCol of feedConfig.targets) { 42 - const targetTable = recordsTableName(targetCol); 201 + // Either no row, or stale started_at. Claim it. 202 + if (!status) { 203 + await db 204 + .prepare( 205 + "INSERT INTO feed_backfills (actor, feed, completed, started_at) VALUES (?, ?, 0, ?) ON CONFLICT DO NOTHING" 206 + ) 207 + .bind(actor, feedName, now) 208 + .run(); 209 + } else { 43 210 await db 44 211 .prepare( 45 - getDialect(db).insertOrIgnore( 46 - `INSERT INTO feed_items (actor, uri, collection, time_us) 47 - SELECT ?, r.uri, ?, r.time_us 48 - FROM ${targetTable} r 49 - WHERE r.did IN ( 50 - SELECT ${getDialect(db).jsonExtract('f.record', 'subject')} 51 - FROM ${followTable} f 52 - WHERE f.did = ? 53 - ) 54 - ORDER BY r.time_us DESC 55 - LIMIT ?` 56 - ) 212 + "UPDATE feed_backfills SET started_at = ? WHERE actor = ? AND feed = ?" 57 213 ) 58 - .bind(actor, targetCol, actor, maxItems) 214 + .bind(now, actor, feedName) 59 215 .run(); 60 216 } 61 217 62 - // Prune oldest items beyond the cap 63 - await db 64 - .prepare( 65 - `DELETE FROM feed_items WHERE actor = ? AND uri NOT IN ( 66 - SELECT uri FROM feed_items WHERE actor = ? ORDER BY time_us DESC LIMIT ? 67 - )` 68 - ) 69 - .bind(actor, actor, maxItems) 70 - .run(); 71 - 72 - await db 73 - .prepare("UPDATE feed_backfills SET completed = 1 WHERE actor = ? AND feed = ?") 74 - .bind(actor, feedName) 75 - .run(); 218 + scheduleBackground(c, config, () => 219 + runFeedBackfill(db, config, actor, feedName, feedConfig) 220 + ); 76 221 } 77 222 78 223 export function registerFeedRoutes( ··· 101 246 const did = await resolveActor(db, actor); 102 247 if (!did) return c.json({ error: "Could not resolve actor" }, 400); 103 248 104 - await maybeBackfillFeed(db, config, did, feedName, feedConfig); 249 + await maybeBackfillFeed(c, db, config, did, feedName, feedConfig); 105 250 106 - const collection = params.get("collection") || feedConfig.targets[0]; 107 - if (!feedConfig.targets.includes(collection)) { 108 - return c.json({ error: "Collection not in feed targets" }, 400); 251 + const targets = feedConfig.targets.map(normalizeFeedTarget); 252 + if (targets.length === 0) { 253 + return c.json({ error: "Feed has no targets configured" }, 500); 254 + } 255 + // Wire-level `collection` is an NSID (matches the generated lex enum and 256 + // what's stored in feed_items.collection). Internally runPipeline expects 257 + // the short name, so translate. 258 + const requestedRaw = params.get("collection"); 259 + let requestedShort: string; 260 + if (!requestedRaw) { 261 + requestedShort = targets[0].collection; 262 + } else if (targets.some((t) => t.collection === requestedRaw)) { 263 + // Tolerate callers passing the short name directly. 264 + requestedShort = requestedRaw; 265 + } else { 266 + const asShort = shortNameForNsid(config, requestedRaw); 267 + if (asShort && targets.some((t) => t.collection === asShort)) { 268 + requestedShort = asShort; 269 + } else { 270 + return c.json({ error: "Collection not in feed targets" }, 400); 271 + } 109 272 } 110 273 111 274 // Strip feed-specific params so runPipeline doesn't misinterpret them ··· 122 285 }; 123 286 124 287 try { 125 - const result = await runPipeline(db, config, collection, pipelineParams, source); 288 + const result = await runPipeline(db, config, requestedShort, pipelineParams, source); 126 289 return c.json(result); 127 290 } catch (e: any) { 128 291 if (e.message === "Could not resolve actor") { ··· 132 295 } 133 296 }); 134 297 } 298 + 299 + export type { FeedTargetConfig };

+1

packages/contrail-appview/src/index.ts

··· 33 33 export * from "./core/backfill"; 34 34 export * from "./core/refresh"; 35 35 export * from "./core/search"; 36 + export * from "./core/constellation"; 36 37 37 38 // DB 38 39 export * from "./core/db/schema";

+18

packages/contrail-base/src/identity.ts

··· 119 119 return resolved.did; 120 120 } 121 121 122 + /** 123 + * Apply a handle change from a Jetstream `#identity` event. 124 + * 125 + * UPDATE-only — does not create a row for unknown DIDs (we'd lack PDS, and 126 + * partial rows confuse the rest of the pipeline). PDS column is left 127 + * untouched; it gets refreshed lazily via `getPDS` / next slingshot resolve. 128 + */ 129 + export async function applyIdentityEvent( 130 + db: Database, 131 + did: string, 132 + handle: string 133 + ): Promise<void> { 134 + await db 135 + .prepare("UPDATE identities SET handle = ?, resolved_at = ? WHERE did = ?") 136 + .bind(handle, Date.now(), did) 137 + .run(); 138 + } 139 + 122 140 export async function refreshStaleIdentities( 123 141 db: Database, 124 142 dids: string[]

+1 -1

packages/contrail-base/src/index.ts

··· 13 13 // Dialect (SqlDialect, getDialect, sqliteDialect, postgresDialect, buildFtsSchema) 14 14 export * from "./dialect"; 15 15 16 - // Identity (resolveActor, resolveIdentities, refreshStaleIdentities) 16 + // Identity (resolveActor, resolveIdentities, refreshStaleIdentities, applyIdentityEvent) 17 17 export * from "./identity"; 18 18 19 19 // PDS client helpers (getPDS, getClient)

+134 -16

packages/contrail-base/src/types.ts

··· 60 60 config: ContrailConfig 61 61 ) => Promise<RecordSource>; 62 62 63 + export interface FeedTargetConfig { 64 + /** Short name of the target collection. */ 65 + collection: string; 66 + /** Per-target item cap. Falls back to FeedConfig.maxItems if unset. */ 67 + maxItems?: number; 68 + } 69 + 63 70 export interface FeedConfig { 64 - /** Short name of the follow collection. */ 65 - follow: string; 66 - /** Short names of target collections to fan out to. */ 67 - targets: string[]; 68 - /** Max feed items per user (default: 200). Oldest items are pruned after backfill. */ 71 + /** Short name of the follow collection. Defaults to "follow" 72 + * (auto-added with NSID `app.bsky.graph.follow`, `discover: false`). */ 73 + follow?: string; 74 + /** Target collections to fan out to. Each entry is either a short name 75 + * or `{ collection, maxItems? }` for per-target caps. */ 76 + targets: (string | FeedTargetConfig)[]; 77 + /** Default per-target item cap when a target doesn't specify its own 78 + * (default: 200). Oldest items per (actor, collection) are pruned. */ 69 79 maxItems?: number; 70 80 } 71 81 72 82 export const DEFAULT_FEED_MAX_ITEMS = 200; 83 + export const DEFAULT_FOLLOW_NSID = "app.bsky.graph.follow"; 84 + export const DEFAULT_FOLLOW_SHORT = "follow"; 85 + 86 + /** Normalize a feed target entry to FeedTargetConfig. */ 87 + export function normalizeFeedTarget( 88 + t: string | FeedTargetConfig 89 + ): FeedTargetConfig { 90 + return typeof t === "string" ? { collection: t } : t; 91 + } 92 + 93 + /** Resolve a feed's per-target item cap, falling back to FeedConfig.maxItems then global default. */ 94 + export function feedTargetMaxItems( 95 + feed: FeedConfig, 96 + target: FeedTargetConfig 97 + ): number { 98 + return target.maxItems ?? feed.maxItems ?? DEFAULT_FEED_MAX_ITEMS; 99 + } 100 + 101 + /** Build a Map<target-NSID, maxItems> across all configured feeds, taking the 102 + * largest cap if the same target collection appears in multiple feeds. */ 103 + export function buildFeedTargetCaps( 104 + config: ContrailConfig 105 + ): Map<string, number> { 106 + const caps = new Map<string, number>(); 107 + if (!config.feeds) return caps; 108 + for (const feed of Object.values(config.feeds)) { 109 + for (const t of feed.targets) { 110 + const target = normalizeFeedTarget(t); 111 + const colCfg = config.collections[target.collection]; 112 + if (!colCfg) continue; 113 + const cap = feedTargetMaxItems(feed, target); 114 + const existing = caps.get(colCfg.collection) ?? 0; 115 + if (cap > existing) caps.set(colCfg.collection, cap); 116 + } 117 + } 118 + return caps; 119 + } 73 120 74 121 export type CollectionMethod = "listRecords" | "getRecord"; 75 122 export const DEFAULT_COLLECTION_METHODS: CollectionMethod[] = [ ··· 96 143 /** When spaces are enabled globally, emit a parallel spaces_records_<short> table 97 144 * so this collection can also live inside spaces. Defaults to true. */ 98 145 allowInSpaces?: boolean; 146 + /** JSON field on the record used as the canonical event time, parsed and 147 + * written into `time_us` during backfill (and clamped to now). Default 148 + * `"createdAt"`. Set to `false` to disable parsing and keep ingest time. */ 149 + timeField?: string | false; 150 + /** JSON field on the record holding a DID that this record points at 151 + * (e.g. `"subject"` for follows). When set on a `discover: false` 152 + * collection, ingest also drops records whose subject DID is not in 153 + * knownDids — useful for trimming network-wide social graphs to the 154 + * subjects we care about. */ 155 + subjectField?: string; 156 + /** Per-record predicate run during ingest. Returning false drops the 157 + * record before it hits the buffer / DB. Runs only for create/update; 158 + * deletes always pass through (the delete may target a record that *did* 159 + * pass an earlier version of the filter). Thrown errors are caught, 160 + * logged, and treated as "drop". Note: Jetstream filters only by 161 + * `wantedCollections`, so non-matching records still travel over the wire 162 + * — this trims what gets persisted, not bandwidth. */ 163 + recordFilter?: (record: Record<string, unknown>) => boolean; 99 164 } 100 165 101 166 export interface ProfileConfig { ··· 170 235 labels?: import("./labels/types").LabelsConfig; 171 236 /** Customize the auto-generated `<namespace>.authFull` lexicon. */ 172 237 permissionSet?: PermissionSetConfig; 238 + /** Constellation-backed reverse-follower lookup (default: enabled). 239 + * When a DID is first seen producing a discoverable record, contrail 240 + * queries Constellation for follow records pointing at that DID and 241 + * ingests synthesized rows for any follower already in our identities 242 + * table. Lets newcomers immediately appear in existing users' feeds. */ 243 + constellation?: ConstellationConfig | false; 173 244 } 245 + 246 + export interface ConstellationConfig { 247 + /** Override the default Constellation instance URL. */ 248 + url?: string; 249 + /** Sent as the User-Agent header per Constellation's request that 250 + * callers identify themselves. Defaults to `contrail/<namespace>`. */ 251 + userAgent?: string; 252 + /** Set false to disable lookups while keeping the table around. */ 253 + enabled?: boolean; 254 + } 255 + 256 + export const DEFAULT_CONSTELLATION_URL = "https://constellation.microcosm.blue"; 174 257 175 258 /** Single entry in an atproto permission-set's `permissions` array. 176 259 * See https://atproto.com/guides/permission-sets for the full schema. */ ··· 218 301 const profiles = (config.profiles ?? DEFAULT_PROFILES).map( 219 302 normalizeProfileConfig 220 303 ); 221 - const collections = { ...config.collections }; 304 + const collections: Record<string, CollectionConfig> = {}; 305 + 306 + // Default `discover: false` for any collection whose NSID lives under the 307 + // `app.bsky.*` namespace, since these are external/network-wide records that 308 + // would otherwise blow up storage if left discoverable. 309 + for (const [short, c] of Object.entries(config.collections)) { 310 + collections[short] = 311 + c.discover === undefined && 312 + typeof c.collection === "string" && 313 + c.collection.startsWith("app.bsky.") 314 + ? { ...c, discover: false } 315 + : c; 316 + } 317 + 222 318 for (const p of profiles) { 223 319 const short = p.shortName!; 224 320 if (!collections[short]) { ··· 226 322 } 227 323 } 228 324 229 - // Auto-add follow collections from feed configs as dependent collections if they're 230 - // not already listed. Feed config already uses short names so nothing to resolve — 231 - // but if the user forgot to declare the follow collection, we can't auto-add it without 232 - // knowing its NSID. In that case we warn later via validateConfig. 325 + // Auto-add a follow collection for any feed that doesn't declare one. 326 + // Default short name `follow` → `app.bsky.graph.follow`, with a `subject` 327 + // filter so we only persist follows pointing at known DIDs. 328 + const feeds = config.feeds; 329 + if (feeds) { 330 + const usedFollowShorts = new Set<string>(); 331 + for (const [, feed] of Object.entries(feeds)) { 332 + const shortName = feed.follow ?? DEFAULT_FOLLOW_SHORT; 333 + usedFollowShorts.add(shortName); 334 + } 335 + for (const short of usedFollowShorts) { 336 + if (!collections[short]) { 337 + collections[short] = { 338 + collection: DEFAULT_FOLLOW_NSID, 339 + discover: false, 340 + subjectField: "subject", 341 + }; 342 + } 343 + } 344 + } 233 345 234 346 const base = { 235 347 ...config, ··· 279 391 280 392 export function getFeedFollowShortNames(config: ContrailConfig): string[] { 281 393 if (!config.feeds) return []; 282 - return [...new Set(Object.values(config.feeds).map((f) => f.follow))]; 394 + return [ 395 + ...new Set( 396 + Object.values(config.feeds).map((f) => f.follow ?? DEFAULT_FOLLOW_SHORT) 397 + ), 398 + ]; 283 399 } 284 400 285 401 /** Alias for getFeedFollowShortNames. */ ··· 377 493 378 494 if (config.feeds) { 379 495 for (const [feedName, feed] of Object.entries(config.feeds)) { 380 - if (!config.collections[feed.follow]) { 496 + const followShort = feed.follow ?? DEFAULT_FOLLOW_SHORT; 497 + if (!config.collections[followShort]) { 381 498 throw new Error( 382 - `Feed "${feedName}" references unknown follow collection "${feed.follow}"` 499 + `Feed "${feedName}" references unknown follow collection "${followShort}"` 383 500 ); 384 501 } 385 - for (const target of feed.targets) { 386 - if (!config.collections[target]) { 502 + for (const t of feed.targets) { 503 + const targetShort = normalizeFeedTarget(t).collection; 504 + if (!config.collections[targetShort]) { 387 505 throw new Error( 388 - `Feed "${feedName}" references unknown target collection "${target}"` 506 + `Feed "${feedName}" references unknown target collection "${targetShort}"` 389 507 ); 390 508 } 391 509 }

+33

packages/contrail/CHANGELOG.md

··· 1 1 # @atmo-dev/contrail 2 2 3 + ## 0.6.0 4 + 5 + ### Minor Changes 6 + 7 + - af24714: Add per-collection `recordFilter` and apply Jetstream `#identity` handle changes during ingest. 8 + 9 + - `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. 10 + - 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. 11 + 12 + ## 0.5.0 13 + 14 + ### Minor Changes 15 + 16 + - 1a6d8cf: Follow-feed overhaul. Several related changes that together fix correctness and storage problems with how follow-driven feeds are bootstrapped, ingested, and recovered. 17 + 18 + **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). 19 + 20 + **`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`. 21 + 22 + **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. 23 + 24 + **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<collection-NSID, cap>`; jetstream/persistent ingest cycles now compute the per-collection map via `buildFeedTargetCaps`. 25 + 26 + **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. 27 + 28 + **`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. 29 + 30 + **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. 31 + 32 + **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/<namespace>` per Constellation's request that callers identify themselves. 33 + 34 + **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. 35 + 3 36 ## 0.4.2 4 37 5 38 ### Patch Changes

+1 -1

packages/contrail/package.json

··· 1 1 { 2 2 "name": "@atmo-dev/contrail", 3 - "version": "0.4.2", 3 + "version": "0.6.0", 4 4 "description": "Index AT Protocol records with typed XRPC endpoints. Cloudflare Workers + D1, SvelteKit, Node.js.", 5 5 "type": "module", 6 6 "sideEffects": false,

+1

packages/contrail/src/core/constellation.ts

··· 1 + export * from "@atmo-dev/contrail-appview";

+81 -1

packages/contrail/tests/persistent.test.ts

··· 1 1 import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 import type { ContrailConfig, Database } from "../src/core/types"; 3 + import { resolveConfig } from "../src/core/types"; 3 4 import { createTestDb, createTestDbWithSchema, TEST_CONFIG } from "./helpers"; 4 5 import { runPersistent } from "../src/core/persistent"; 5 6 import { getLastCursor, queryRecords } from "../src/core/db/records"; 6 7 import { initSchema } from "../src/core/db/schema"; 7 8 8 9 // Identity helpers live in @atmo-dev/contrail-base post-split. Mock there. 10 + const applyIdentityEventMock = vi.fn().mockResolvedValue(undefined); 9 11 vi.mock("@atmo-dev/contrail-base", async (importOriginal) => { 10 12 const actual = await importOriginal<typeof import("@atmo-dev/contrail-base")>(); 11 13 return { 12 14 ...actual, 13 15 refreshStaleIdentities: vi.fn().mockResolvedValue(undefined), 16 + applyIdentityEvent: (...args: unknown[]) => applyIdentityEventMock(...args), 14 17 }; 15 18 }); 16 19 ··· 296 299 expect(row!.count_rsvp_going).toBe(1); 297 300 }); 298 301 302 + it("drops records that fail a collection's recordFilter", async () => { 303 + // Filter accepts only events whose `name` contains "keep". The other 304 + // events have well-formed records but should never reach the DB. 305 + const filterConfig = resolveConfig({ 306 + namespace: "com.example", 307 + collections: { 308 + event: { 309 + collection: "community.lexicon.calendar.event", 310 + recordFilter: (r) => 311 + typeof r.name === "string" && r.name.includes("keep"), 312 + }, 313 + }, 314 + }); 315 + 316 + const freshDb = createTestDb(); 317 + await initSchema(freshDb, filterConfig); 318 + 319 + const events = [ 320 + { 321 + kind: "commit" as const, 322 + did: "did:plc:a", 323 + time_us: 7000, 324 + commit: { 325 + collection: "community.lexicon.calendar.event", 326 + operation: "create", 327 + rkey: "drop1", 328 + cid: "c1", 329 + record: { name: "drop me", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, 330 + }, 331 + }, 332 + { 333 + kind: "commit" as const, 334 + did: "did:plc:b", 335 + time_us: 7001, 336 + commit: { 337 + collection: "community.lexicon.calendar.event", 338 + operation: "create", 339 + rkey: "keep1", 340 + cid: "c2", 341 + record: { name: "keep this", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, 342 + }, 343 + }, 344 + ]; 345 + 346 + const controller = new AbortController(); 347 + const promise = runPersistent(freshDb, filterConfig, { 348 + batchSize: 100, 349 + flushIntervalMs: 50, 350 + signal: controller.signal, 351 + createSubscription: () => mockSubscription(events) as any, 352 + }); 353 + 354 + await new Promise((r) => setTimeout(r, 200)); 355 + controller.abort(); 356 + await promise; 357 + 358 + const result = await queryRecords(freshDb, filterConfig, { 359 + collection: "community.lexicon.calendar.event", 360 + limit: 100, 361 + }); 362 + expect(result.records.length).toBe(1); 363 + expect(result.records[0]!.uri).toContain("/keep1"); 364 + }); 365 + 299 366 it("skips non-commit events", async () => { 367 + applyIdentityEventMock.mockClear(); 300 368 const events = [ 301 - { kind: "identity" as const, did: "did:plc:someone", time_us: 4000 }, 369 + { 370 + kind: "identity" as const, 371 + did: "did:plc:someone", 372 + time_us: 4000, 373 + identity: { did: "did:plc:someone", handle: "newhandle.test", seq: 1, time: "2026-04-01T10:00:00Z" }, 374 + }, 302 375 { 303 376 kind: "commit" as const, 304 377 did: "did:plc:real", ··· 331 404 limit: 100, 332 405 }); 333 406 expect(result.records.length).toBe(1); 407 + // Identity events update the identities table even though they don't 408 + // produce records. 409 + expect(applyIdentityEventMock).toHaveBeenCalledWith( 410 + expect.anything(), 411 + "did:plc:someone", 412 + "newhandle.test" 413 + ); 334 414 }); 335 415 });

+27

packages/lexicons/CHANGELOG.md

··· 1 1 # @atmo-dev/contrail-lexicons 2 2 3 + ## 0.4.6 4 + 5 + ### Patch Changes 6 + 7 + - Updated dependencies [af24714] 8 + - @atmo-dev/contrail@0.6.0 9 + 10 + ## 0.4.5 11 + 12 + ### Patch Changes 13 + 14 + - 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. 15 + 16 + ## 0.4.4 17 + 18 + ### Patch Changes 19 + 20 + - 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. 21 + - Updated dependencies [1a6d8cf] 22 + - @atmo-dev/contrail@0.5.0 23 + 24 + ## 0.4.3 25 + 26 + ### Patch Changes 27 + 28 + - 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`. 29 + 3 30 ## 0.4.2 4 31 5 32 ### Patch Changes

+1 -1

packages/lexicons/package.json

··· 1 1 { 2 2 "name": "@atmo-dev/contrail-lexicons", 3 - "version": "0.4.2", 3 + "version": "0.4.6", 4 4 "description": "Generate atproto lexicon JSON (and optionally TypeScript types via @atcute/lex-cli) from a Contrail config.", 5 5 "type": "module", 6 6 "files": [

+24 -4

packages/lexicons/src/generate.ts

··· 401 401 log("Generating feed endpoint..."); 402 402 403 403 const feedNames = Object.keys(config.feeds); 404 - // feedConfig.targets are short names; expose NSIDs in the lexicon since the 405 - // `collection` param filters by the record's NSID at the wire level. 406 - const allTargets = [...new Set(Object.values(config.feeds).flatMap((f) => f.targets))]; 404 + // feedConfig.targets are short names (or `{ collection, maxItems? }`); 405 + // expose NSIDs in the lexicon since the `collection` param filters by 406 + // the record's NSID at the wire level. 407 + const allTargets = [ 408 + ...new Set( 409 + Object.values(config.feeds).flatMap((f) => 410 + f.targets.map((t) => (typeof t === "string" ? t : t.collection)) 411 + ) 412 + ), 413 + ]; 407 414 const allTargetNsids = allTargets 408 415 .map((t) => config.collections[t]?.collection) 409 416 .filter((n): n is string => !!n); ··· 1031 1038 const profileNsids: string[] = (config.profiles ?? ["app.bsky.actor.profile"]).map( 1032 1039 (p) => (typeof p === "string" ? p : p.collection) 1033 1040 ); 1034 - const feedFollowNsids = config.feeds ? Object.values(config.feeds).map((f) => f.follow) : []; 1041 + // f.follow is a short name (a key in config.collections), not an NSID — resolve 1042 + // it before pushing into the pull list, otherwise lex-cli pull rejects it as 1043 + // "must be valid nsid". When `follow` is unset, contrail's resolveConfig 1044 + // auto-adds an `app.bsky.graph.follow` collection at runtime; mirror that 1045 + // here so the generated pull list still includes it. 1046 + const DEFAULT_FOLLOW_NSID = "app.bsky.graph.follow"; 1047 + const feedFollowNsids = config.feeds 1048 + ? Object.values(config.feeds) 1049 + .map((f) => { 1050 + if (!f.follow) return DEFAULT_FOLLOW_NSID; 1051 + return config.collections[f.follow]?.collection; 1052 + }) 1053 + .filter((nsid): nsid is string => typeof nsid === "string") 1054 + : []; 1035 1055 const pullNsids = new Set([...collectionNsids, ...profileNsids, ...feedFollowNsids]); 1036 1056 for (const ref of allRefs) { 1037 1057 if (!ref.startsWith("com.atproto.")) pullNsids.add(ref);

+78

packages/lexicons/tests/generate.test.ts

··· 376 376 } 377 377 }); 378 378 }); 379 + 380 + describe("runtime files: lex.config.js pull NSIDs", () => { 381 + let workdir: string; 382 + 383 + beforeAll(() => { 384 + workdir = mkdtempSync(join(tmpdir(), "contrail-feedpull-")); 385 + }); 386 + afterAll(() => { 387 + rmSync(workdir, { recursive: true, force: true }); 388 + }); 389 + 390 + function readPullNsids(): string[] { 391 + const lexConfig = readFileSync(join(workdir, "lex.config.js"), "utf-8"); 392 + const match = lexConfig.match(/nsids:\s*(\[[\s\S]*?\])/); 393 + if (!match) throw new Error("could not locate pull nsids array in lex.config.js"); 394 + return JSON.parse(match[1]); 395 + } 396 + 397 + it("resolves feed.follow short names to NSIDs (regression: lex-cli pull rejects bare short names)", () => { 398 + const config: ContrailConfig = { 399 + namespace: "test.app", 400 + collections: { 401 + follow: { collection: "app.bsky.graph.follow" }, 402 + event: { collection: "community.lexicon.calendar.event" }, 403 + }, 404 + feeds: { 405 + network: { follow: "follow", targets: ["event"] }, 406 + }, 407 + }; 408 + 409 + generateLexicons({ 410 + config, 411 + rootDir: workdir, 412 + lexiconDirs: [], 413 + writeRuntimeFiles: true, 414 + quiet: true, 415 + }); 416 + 417 + const nsids = readPullNsids(); 418 + 419 + // Every entry must be a valid NSID — at least one dot, and never equal a feed short name. 420 + const feedShortNames = Object.keys(config.collections); 421 + for (const nsid of nsids) { 422 + expect(nsid).toMatch(/\./); 423 + expect(feedShortNames).not.toContain(nsid); 424 + } 425 + 426 + // The follow collection's NSID is included via the feeds path. 427 + expect(nsids).toContain("app.bsky.graph.follow"); 428 + }); 429 + 430 + it("skips feeds whose follow short name is missing from collections (no undefineds in pull list)", () => { 431 + const config = { 432 + namespace: "test.app", 433 + collections: { 434 + event: { collection: "community.lexicon.calendar.event" }, 435 + }, 436 + feeds: { 437 + broken: { follow: "doesNotExist", targets: ["event"] }, 438 + }, 439 + } as unknown as ContrailConfig; 440 + 441 + generateLexicons({ 442 + config, 443 + rootDir: workdir, 444 + lexiconDirs: [], 445 + writeRuntimeFiles: true, 446 + quiet: true, 447 + }); 448 + 449 + const nsids = readPullNsids(); 450 + for (const nsid of nsids) { 451 + expect(typeof nsid).toBe("string"); 452 + expect(nsid).toMatch(/\./); 453 + } 454 + expect(nsids).not.toContain("doesNotExist"); 455 + }); 456 + });

Configure Feed

Configure Feed