Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112import { safeParse } from '@atcute/lexicons';import { DevNiblingActorProfile, DevNiblingFeedLike, DevNiblingGraphFollow, DevNiblingProject } from '@nibling/lexicons';import { INDEX_COLLECTIONS, type IndexCollection, type IndexEvent } from './model.ts';
/** The four collections the index cares about (D22). */export const WANTED_COLLECTIONS = INDEX_COLLECTIONS;
function isIndexCollection(value: string): value is IndexCollection { return (INDEX_COLLECTIONS as readonly string[]).includes(value);}
/** * Jetstream v2 is `/xrpc/network.bsky.jetstream.subscribeEvents` with repeated `collections` * params, `kinds`, and a `cursor` that's either a `seq` or a `time_us`, told apart by magnitude. * The index passes a `time_us` (see ingest.ts). `collections` supports NSID-prefix wildcards * (an `app.bsky.feed.*`-style namespace wildcard), so `dev.nibling.*` would also work, but since * there are only four collections to ask for, this lists them explicitly instead. */export function subscribeUrl(baseUrl: string, cursor?: string): string { const url = new URL('/xrpc/network.bsky.jetstream.subscribeEvents', baseUrl); for (const collection of WANTED_COLLECTIONS) url.searchParams.append('collections', collection); url.searchParams.set('kinds', 'commit'); if (cursor) url.searchParams.set('cursor', cursor); return url.toString();}
interface JetstreamCommit { $type: 'network.bsky.jetstream.subscribeEvents#commit'; did: string; seq: number; time: string; operation: 'create' | 'update' | 'delete'; collection: string; rkey: string; cid?: string; record?: unknown;}
function isCommit(payload: unknown): payload is JetstreamCommit { return ( typeof payload === 'object' && payload !== null && (payload as { $type?: unknown }).$type === 'network.bsky.jetstream.subscribeEvents#commit' && typeof (payload as { did?: unknown }).did === 'string' && typeof (payload as { collection?: unknown }).collection === 'string' && typeof (payload as { rkey?: unknown }).rkey === 'string' && typeof (payload as { seq?: unknown }).seq === 'number' );}
/** * Parses one Jetstream frame into an `IndexEvent`, or `undefined` to drop it. Frames are * untrusted: a different collection, a non-commit event, or a record that fails its lexicon's * schema is dropped rather than indexed. `now` is injectable for deterministic tests. */export function parseJetstreamMessage(raw: unknown, now: () => string = () => new Date().toISOString()): IndexEvent | undefined { if (typeof raw !== 'object' || raw === null) return undefined; const payload = (raw as { payload?: unknown }).payload; if (!isCommit(payload)) return undefined;
const uri = `at://${payload.did}/${payload.collection}/${payload.rkey}`;
if (payload.operation === 'delete') { return isIndexCollection(payload.collection) ? { action: 'delete', collection: payload.collection, uri } : undefined; }
if (payload.operation !== 'create' && payload.operation !== 'update') return undefined; if (typeof payload.cid !== 'string') return undefined;
return toPutEvent(payload.collection, uri, payload.cid, payload.did, payload.record, now);}
/** * Validates an untrusted record against its lexicon and builds a `put` `IndexEvent`, or * `undefined` if the collection is unknown or the record fails validation. Shared by the live * tail (above) and `backfill.ts`, so both apply the same boundary check (D22, "validate with * the generated @nibling/lexicons schemas at the boundary"). */export function toPutEvent( collection: string, uri: string, cid: string, did: string, record: unknown, now: () => string = () => new Date().toISOString(),): IndexEvent | undefined { switch (collection) { case 'dev.nibling.project': { const result = safeParse(DevNiblingProject.mainSchema, record); if (!result.ok) return undefined; return { action: 'put', collection: 'dev.nibling.project', uri, cid, did, record: result.value, indexedAt: now() }; } case 'dev.nibling.feed.like': { const result = safeParse(DevNiblingFeedLike.mainSchema, record); if (!result.ok) return undefined; return { action: 'put', collection: 'dev.nibling.feed.like', uri, cid, did, record: result.value, indexedAt: now() }; } case 'dev.nibling.graph.follow': { const result = safeParse(DevNiblingGraphFollow.mainSchema, record); if (!result.ok) return undefined; return { action: 'put', collection: 'dev.nibling.graph.follow', uri, cid, did, record: result.value }; } case 'dev.nibling.actor.profile': { const result = safeParse(DevNiblingActorProfile.mainSchema, record); if (!result.ok) return undefined; return { action: 'put', collection: 'dev.nibling.actor.profile', uri, cid, did, record: result.value }; } default: return undefined; }}