Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393import type { DevNiblingFeedDefs } from '@nibling/lexicons';
/** The one escape hatch between the pure model and its storage. The DO implements it over * `ctx.storage.sql.exec(...).toArray()`; tests implement it over `node:sqlite`. */export interface Sql { exec<T = Record<string, unknown>>(query: string, ...bindings: unknown[]): T[];}
interface ProjectRecord { name: string; slug?: string; description?: string; tags?: readonly string[]; forkOf?: { uri: string; cid: string }; createdAt: string; updatedAt?: string;}
interface LikeRecord { subject: { uri: string; cid: string }; createdAt: string;}
interface FollowRecord { subject: string; via?: string; createdAt: string;}
interface ProfileRecord { displayName?: string; description?: string; createdAt?: string;}
/** One put or tombstone, already validated at the boundary (jetstream.ts, backfill.ts). */export type IndexEvent = | { action: 'put'; collection: 'dev.nibling.project'; uri: string; cid: string; did: string; record: ProjectRecord; indexedAt: string } | { action: 'put'; collection: 'dev.nibling.feed.like'; uri: string; cid: string; did: string; record: LikeRecord; indexedAt: string } | { action: 'put'; collection: 'dev.nibling.graph.follow'; uri: string; cid: string; did: string; record: FollowRecord } | { action: 'put'; collection: 'dev.nibling.actor.profile'; uri: string; cid: string; did: string; record: ProfileRecord } | { action: 'delete'; collection: IndexCollection; uri: string };
type PutEvent<C extends IndexCollection> = Extract<IndexEvent, { action: 'put'; collection: C }>;
export const INDEX_COLLECTIONS = [ 'dev.nibling.project', 'dev.nibling.feed.like', 'dev.nibling.graph.follow', 'dev.nibling.actor.profile',] as const;
export type IndexCollection = (typeof INDEX_COLLECTIONS)[number];
/** Where each collection's rows live, keyed by the repo DID that owns them. */const TABLES: Record<IndexCollection, string> = { 'dev.nibling.project': 'projects', 'dev.nibling.feed.like': 'likes', 'dev.nibling.graph.follow': 'follows', 'dev.nibling.actor.profile': 'profiles',};
export interface RecentOptions { limit?: number; cursor?: string; tag?: string;}
export interface TrendingOptions { limit?: number; cursor?: string; /** Epoch ms. Threaded through explicitly so ranking is deterministic in tests. */ now: number;}
type ProjectView = DevNiblingFeedDefs.ProjectView;
const KNOWN_CHUNK = 50;
// One statement per array entry: `Sql.exec` runs exactly one statement per call (that's what// `ctx.storage.sql.exec` does in the DO), so the schema can't be a single multi-statement string.const SCHEMA = [ `CREATE TABLE IF NOT EXISTS projects ( uri TEXT PRIMARY KEY, cid TEXT NOT NULL, did TEXT NOT NULL, name TEXT NOT NULL, slug TEXT, description TEXT, forkOf TEXT, createdAt TEXT NOT NULL, updatedAt TEXT, sortAt TEXT NOT NULL, indexedAt TEXT NOT NULL, createdMs INTEGER NOT NULL )`, `CREATE INDEX IF NOT EXISTS projects_sortAt ON projects(sortAt DESC)`, `CREATE INDEX IF NOT EXISTS projects_forkOf ON projects(forkOf)`, `CREATE INDEX IF NOT EXISTS projects_did ON projects(did)`, `CREATE TABLE IF NOT EXISTS project_tags ( uri TEXT NOT NULL, tag TEXT NOT NULL, PRIMARY KEY (uri, tag) )`, `CREATE INDEX IF NOT EXISTS project_tags_tag ON project_tags(tag)`, `CREATE TABLE IF NOT EXISTS likes ( uri TEXT PRIMARY KEY, did TEXT NOT NULL, subjectUri TEXT NOT NULL, atMs INTEGER NOT NULL )`, `CREATE INDEX IF NOT EXISTS likes_subject ON likes(subjectUri)`, `CREATE INDEX IF NOT EXISTS likes_did ON likes(did)`, `CREATE TABLE IF NOT EXISTS follows ( uri TEXT PRIMARY KEY, did TEXT NOT NULL, subject TEXT NOT NULL, createdAt TEXT NOT NULL )`, `CREATE INDEX IF NOT EXISTS follows_did ON follows(did)`, `CREATE TABLE IF NOT EXISTS profiles ( did TEXT PRIMARY KEY, uri TEXT NOT NULL, createdAt TEXT )`, `CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL )`,];
/** When something happened, as far as the index will believe: the author's `createdAt`, but * never later than when the index first saw it. Authors can backdate a record (which only ages * it) but can't post-date one to stay fresh. */function observedMs(authorTime: string, indexedAt: string): number { return Math.min(Date.parse(authorTime), Date.parse(indexedAt));}
function encodeCursor(sortAt: string, uri: string): string { return Buffer.from(JSON.stringify([sortAt, uri])).toString('base64url');}
function decodeCursor(cursor: string): { sortAt: string; uri: string } | undefined { try { const [sortAt, uri] = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as [string, string]; if (typeof sortAt !== 'string' || typeof uri !== 'string') return undefined; return { sortAt, uri }; } catch { return undefined; }}
/** The index: a pure, replayable view over `dev.nibling.*` records. Never the source of truth * (the author's PDS is); this holds only what makes `dev.nibling.feed.*` and * `dev.nibling.actor.getKnown` fast to answer. */export class IndexModel { private readonly sql: Sql;
constructor(sql: Sql) { this.sql = sql; for (const statement of SCHEMA) this.sql.exec(statement); this.migrate(); }
/** `CREATE TABLE IF NOT EXISTS` above doesn't add columns to a table that already existed * (every Durable Object provisioned before `slug` shipped), so check and add it here instead. */ private migrate(): void { const columns = this.sql.exec<{ name: string }>('PRAGMA table_info(projects)'); if (!columns.some((c) => c.name === 'slug')) { this.sql.exec('ALTER TABLE projects ADD COLUMN slug TEXT'); } }
apply(event: IndexEvent): void { if (event.action === 'delete') { this.remove(event.collection, event.uri); return; } switch (event.collection) { case 'dev.nibling.project': this.applyProject(event); break; case 'dev.nibling.feed.like': this.applyLike(event); break; case 'dev.nibling.graph.follow': this.applyFollow(event); break; case 'dev.nibling.actor.profile': this.applyProfile(event); break; } }
private remove(collection: IndexCollection, uri: string): void { if (collection === 'dev.nibling.project') this.sql.exec('DELETE FROM project_tags WHERE uri = ?', uri); this.sql.exec(`DELETE FROM ${TABLES[collection]} WHERE uri = ?`, uri); }
private applyProject(event: PutEvent<'dev.nibling.project'>): void { const { uri, cid, did, record, indexedAt } = event; const existing = this.sql.exec<{ cid: string }>('SELECT cid FROM projects WHERE uri = ?', uri); if (existing[0]?.cid === cid) return; const sortAt = new Date(observedMs(record.updatedAt ?? record.createdAt, indexedAt)).toISOString(); this.sql.exec( `INSERT INTO projects (uri, cid, did, name, slug, description, forkOf, createdAt, updatedAt, sortAt, indexedAt, createdMs) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, did = excluded.did, name = excluded.name, slug = excluded.slug, description = excluded.description, forkOf = excluded.forkOf, createdAt = excluded.createdAt, updatedAt = excluded.updatedAt, sortAt = excluded.sortAt, indexedAt = excluded.indexedAt, createdMs = MIN(projects.createdMs, excluded.createdMs)`, uri, cid, did, record.name, record.slug ?? null, record.description ?? null, record.forkOf?.uri ?? null, record.createdAt, record.updatedAt ?? null, sortAt, indexedAt, observedMs(record.createdAt, indexedAt), ); this.sql.exec('DELETE FROM project_tags WHERE uri = ?', uri); for (const tag of record.tags ?? []) { this.sql.exec('INSERT OR IGNORE INTO project_tags (uri, tag) VALUES (?, ?)', uri, tag); } }
private applyLike(event: PutEvent<'dev.nibling.feed.like'>): void { const { uri, did, record, indexedAt } = event; this.sql.exec( `INSERT INTO likes (uri, did, subjectUri, atMs) VALUES (?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET did = excluded.did, subjectUri = excluded.subjectUri, atMs = MIN(likes.atMs, excluded.atMs)`, uri, did, record.subject.uri, observedMs(record.createdAt, indexedAt), ); }
private applyFollow(event: PutEvent<'dev.nibling.graph.follow'>): void { const { uri, did, record } = event; this.sql.exec( `INSERT INTO follows (uri, did, subject, createdAt) VALUES (?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET did = excluded.did, subject = excluded.subject, createdAt = excluded.createdAt`, uri, did, record.subject, record.createdAt, ); }
private applyProfile(event: PutEvent<'dev.nibling.actor.profile'>): void { const { uri, did, record } = event; this.sql.exec( `INSERT INTO profiles (did, uri, createdAt) VALUES (?, ?, ?) ON CONFLICT(did) DO UPDATE SET uri = excluded.uri, createdAt = excluded.createdAt`, did, uri, record.createdAt ?? null, ); }
private likeCount(uri: string): number { const rows = this.sql.exec<{ n: number }>('SELECT COUNT(DISTINCT did) as n FROM likes WHERE subjectUri = ?', uri); return rows[0]?.n ?? 0; }
private forkCount(uri: string): number { const rows = this.sql.exec<{ n: number }>('SELECT COUNT(DISTINCT did) as n FROM projects WHERE forkOf = ?', uri); return rows[0]?.n ?? 0; }
private toView(row: { uri: string; cid: string; did: string; name: string; slug: string | null; description: string | null; forkOf: string | null; sortAt: string; indexedAt: string; }): ProjectView { const tagRows = this.sql.exec<{ tag: string }>('SELECT tag FROM project_tags WHERE uri = ? ORDER BY tag', row.uri); const view: ProjectView = { uri: row.uri as ProjectView['uri'], cid: row.cid as ProjectView['cid'], did: row.did as ProjectView['did'], name: row.name, updatedAt: row.sortAt as ProjectView['updatedAt'], indexedAt: row.indexedAt as ProjectView['indexedAt'], likeCount: this.likeCount(row.uri), forkCount: this.forkCount(row.uri), }; if (row.slug !== null) view.slug = row.slug; if (row.description !== null) view.description = row.description; if (row.forkOf !== null) view.forkOf = row.forkOf as ProjectView['forkOf']; const tags = tagRows.map((r) => r.tag); if (tags.length > 0) view.tags = tags; return view; }
recent(options: RecentOptions = {}): { projects: ProjectView[]; cursor?: string } { const limit = options.limit ?? 30; const after = options.cursor ? decodeCursor(options.cursor) : undefined; const params: unknown[] = []; let query = options.tag ? `SELECT p.* FROM projects p JOIN project_tags t ON t.uri = p.uri WHERE t.tag = ?` : `SELECT p.* FROM projects p WHERE 1 = 1`; if (options.tag) params.push(options.tag); if (after) { query += ` AND (p.sortAt < ? OR (p.sortAt = ? AND p.uri < ?))`; params.push(after.sortAt, after.sortAt, after.uri); } query += ` ORDER BY p.sortAt DESC, p.uri DESC LIMIT ?`; params.push(limit + 1); const rows = this.sql.exec<{ uri: string; cid: string; did: string; name: string; slug: string | null; description: string | null; forkOf: string | null; sortAt: string; indexedAt: string; }>(query, ...params); const page = rows.slice(0, limit); const projects = page.map((row) => this.toView(row)); const cursor = rows.length > limit && page.length > 0 ? encodeCursor(page[page.length - 1]!.sortAt, page[page.length - 1]!.uri) : undefined; return cursor ? { projects, cursor } : { projects }; }
trending(options: TrendingOptions): { projects: ProjectView[]; cursor?: string } { const limit = options.limit ?? 30; const windowMs = 7 * 24 * 60 * 60 * 1000; const windowStart = options.now - windowMs; // Linear decay to zero at the edge of the 7-day window: a like or fork from just now // scores 1, one from 7 days ago scores 0. Deterministic given `now`. // Each account counts once per project: its freshest like, plus its freshest fork. const rows = this.sql.exec<{ uri: string; score: number }>( `WITH like_scores AS ( SELECT uri, SUM(s) AS score FROM ( SELECT subjectUri AS uri, MAX(MAX(0.0, ? - (? - atMs)) / ?) AS s FROM likes WHERE atMs >= ? GROUP BY subjectUri, did ) GROUP BY uri ), fork_scores AS ( SELECT uri, SUM(s) AS score FROM ( SELECT forkOf AS uri, MAX(MAX(0.0, ? - (? - createdMs)) / ?) AS s FROM projects WHERE forkOf IS NOT NULL AND createdMs >= ? GROUP BY forkOf, did ) GROUP BY uri ) SELECT p.uri AS uri, COALESCE(ls.score, 0) + COALESCE(fs.score, 0) AS score FROM projects p LEFT JOIN like_scores ls ON ls.uri = p.uri LEFT JOIN fork_scores fs ON fs.uri = p.uri ORDER BY score DESC, p.sortAt DESC, p.uri DESC`, windowMs, options.now, windowMs, windowStart, windowMs, options.now, windowMs, windowStart, ); // Trending is re-ranked on every call (scores decay continuously), so there is no stable sort // key to resume from. The cursor is the rank offset into that call's ranking. const offset = options.cursor ? Math.max(0, Number.parseInt(options.cursor, 10) || 0) : 0; const page = rows.slice(offset, offset + limit); const projects = page.map((r) => this.getProjectView(r.uri)).filter((v): v is ProjectView => v !== undefined); const cursor = offset + limit < rows.length ? String(offset + limit) : undefined; return cursor ? { projects, cursor } : { projects }; }
private getProjectView(uri: string): ProjectView | undefined { const rows = this.sql.exec<{ uri: string; cid: string; did: string; name: string; slug: string | null; description: string | null; forkOf: string | null; sortAt: string; indexedAt: string; }>('SELECT * FROM projects WHERE uri = ?', uri); return rows[0] ? this.toView(rows[0]) : undefined; }
known(dids: readonly string[]): string[] { const found = new Set<string>(); // Each DID binds twice, and Durable Object SQLite caps a query at 100 bound parameters. for (let i = 0; i < dids.length; i += KNOWN_CHUNK) { const chunk = dids.slice(i, i + KNOWN_CHUNK); const placeholders = chunk.map(() => '?').join(', '); const rows = this.sql.exec<{ did: string }>( `SELECT did FROM projects WHERE did IN (${placeholders}) UNION SELECT did FROM profiles WHERE did IN (${placeholders})`, ...chunk, ...chunk, ); for (const row of rows) found.add(row.did); } return dids.filter((did) => found.has(did)); }
/** Deletes the rows one repo owns in a collection that its PDS no longer lists, which is how * a backfill catches deletions that happened while the index wasn't listening. */ pruneRepo(collection: IndexCollection, did: string, keep: ReadonlySet<string>): void { const rows = this.sql.exec<{ uri: string }>(`SELECT uri FROM ${TABLES[collection]} WHERE did = ?`, did); for (const { uri } of rows) { if (!keep.has(uri)) this.apply({ action: 'delete', collection, uri }); } }
/** The Jetstream cursor, as `time_us` (microseconds since the epoch). */ cursor(): number | undefined { const rows = this.sql.exec<{ value: string }>("SELECT value FROM meta WHERE key = 'jetstreamCursor'"); const value = rows[0] ? Number(rows[0].value) : Number.NaN; return Number.isSafeInteger(value) ? value : undefined; }
setCursor(timeUs: number): void { this.sql.exec( `INSERT INTO meta (key, value) VALUES ('jetstreamCursor', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value`, String(timeUs), ); }}