Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192// The Node-only half of @radial/ingest, imported as `@radial/ingest/node`.//// The poller and the space ingestor are isomorphic — they read repos over `fetch` — so a browser// runs them unchanged. Only the SQLite checkpoint store needs a Node process; a browser supplies// its own `SyncStateStore` (IndexedDB) instead.
import { DatabaseSync } from 'node:sqlite'import type { CursorStore, RepoCheckpoint, SyncStateStore } from './state.js'
export class SqliteSyncStateStore implements SyncStateStore, CursorStore { readonly #database: DatabaseSync constructor(path = ':memory:') { this.#database = new DatabaseSync(path) this.#database.exec(` CREATE TABLE IF NOT EXISTS repo_checkpoints ( did TEXT PRIMARY KEY, rev TEXT NOT NULL, commit_cid TEXT NOT NULL, first_observed_rev TEXT NOT NULL, updated_at TEXT NOT NULL ) STRICT `) // A second table rather than a column on the first: a Jetstream cursor is not a repo rev, and // the two have different lifetimes (a cursor survives `radiald index reset`'s concerns about // records because it is only a resume hint — a stale one costs a replay, never a gap). this.#database.exec(` CREATE TABLE IF NOT EXISTS jetstream_cursors ( key TEXT PRIMARY KEY, cursor TEXT NOT NULL, updated_at TEXT NOT NULL ) STRICT `) }
getCursor(key: string): string | undefined { const row = this.#database.prepare('SELECT cursor FROM jetstream_cursors WHERE key = ?').get(key) as | Record<string, string> | undefined return row ? (row.cursor as string) : undefined }
setCursor(key: string, cursor: string): void { this.#database .prepare(` INSERT INTO jetstream_cursors (key, cursor, updated_at) VALUES (?, ?, ?) ON CONFLICT (key) DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at `) .run(key, cursor, new Date().toISOString()) } get(did: string): RepoCheckpoint | undefined { const row = this.#database .prepare('SELECT did, rev, commit_cid, first_observed_rev, updated_at FROM repo_checkpoints WHERE did = ?') .get(did) as Record<string, string> | undefined return row ? { did: row.did as string, rev: row.rev as string, commitCid: row.commit_cid as string, firstObservedRev: row.first_observed_rev as string, updatedAt: row.updated_at as string, } : undefined } set(value: RepoCheckpoint): void { this.#database .prepare(` INSERT INTO repo_checkpoints (did, rev, commit_cid, first_observed_rev, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT (did) DO UPDATE SET rev = excluded.rev, commit_cid = excluded.commit_cid, first_observed_rev = excluded.first_observed_rev, updated_at = excluded.updated_at `) .run(value.did, value.rev, value.commitCid, value.firstObservedRev, value.updatedAt) } all(): RepoCheckpoint[] { return (this.#database .prepare('SELECT did, rev, commit_cid, first_observed_rev, updated_at FROM repo_checkpoints ORDER BY did') .all() as Array<Record<string, string>>).map((row) => ({ did: row.did as string, rev: row.rev as string, commitCid: row.commit_cid as string, firstObservedRev: row.first_observed_rev as string, updatedAt: row.updated_at as string, })) } close(): void { this.#database.close() }}