+11

packages/contrail-appview/src/index.ts

Reviewed

··· 16 16 export * from "@atmo-dev/contrail-authority"; 17 17 export * from "@atmo-dev/contrail-record-host"; 18 18 19 19 + // Record-sync ingestion (consumer side of recordHost.sync) 20 20 + export { 21 21 + runRecordHostSync, 22 22 + buildRecordSyncSchema, 23 23 + applyRecordSyncSchema, 24 24 + } from "./sync"; 25 25 + export type { 26 26 + RecordHostSyncSource, 27 27 + RecordHostSyncOptions, 28 28 + } from "./sync"; 29 29 + 19 30 // Indexing pipeline (jetstream, persistent, backfill, refresh, ingest helpers) 20 31 export * from "./core/jetstream"; 21 32 export * from "./core/persistent";

+229

packages/contrail-appview/src/sync.ts

Reviewed

··· 1 1 + /** Appview-side ingestion loop for the recordHost.sync streaming endpoint. 2 2 + * 3 3 + * Opens an SSE connection to a remote host's `<ns>.recordHost.sync` endpoint 4 4 + * for a (host, space) pair, parses the event stream, writes each 5 5 + * `record.created` / `record.deleted` event into the local record host's 6 6 + * tables, and persists the cursor after every checkpoint. 7 7 + * 8 8 + * Designed to be called per-subscription. Reconnect logic is the caller's — 9 9 + * this function returns when the stream ends or an error throws. Wrap it in 10 10 + * a retry-with-backoff loop in your worker / persistent-process. */ 11 11 + 12 12 + import type { 13 13 + ContrailConfig, 14 14 + Database, 15 15 + RecordHost, 16 16 + SqlDialect, 17 17 + } from "@atmo-dev/contrail-base"; 18 18 + import { getDialect } from "@atmo-dev/contrail-base"; 19 19 + 20 20 + export interface RecordHostSyncSource { 21 21 + /** Remote host's base URL, e.g. "https://contrail-a.example.com". */ 22 22 + hostUrl: string; 23 23 + /** Space we want to sync. */ 24 24 + spaceUri: string; 25 25 + /** Authority DID — used for auto-enrolling locally on first connect. */ 26 26 + authorityDid: string; 27 27 + /** Credential the appview presents to read this space's stream. */ 28 28 + credential: string; 29 29 + /** Sync endpoint NSID; defaults to "<config.namespace>.recordHost.sync". */ 30 30 + endpointNsid?: string; 31 31 + } 32 32 + 33 33 + export interface RecordHostSyncOptions { 34 34 + /** Local DB the records go into (same DB the appview's RecordHost adapter uses). */ 35 35 + db: Database; 36 36 + /** Resolved config. Used to derive the remote endpoint NSID and to find 37 37 + * collection short names for table writes. */ 38 38 + config: ContrailConfig; 39 39 + /** Local record host — the destination for ingested events. The function 40 40 + * calls `putRecord` / `deleteRecord` / `enroll` on this. */ 41 41 + recordHost: RecordHost; 42 42 + /** fetch implementation — pass the remote host app's fetch directly for 43 43 + * in-process tests; defaults to globalThis.fetch. */ 44 44 + fetch?: typeof fetch; 45 45 + /** Aborts the stream when triggered. */ 46 46 + signal?: AbortSignal; 47 47 + /** Called on each cursor checkpoint, after persistence. */ 48 48 + onCursor?: (cursor: string) => void; 49 49 + } 50 50 + 51 51 + /** Run sync for a single (host, space) source until the stream ends or the 52 52 + * signal aborts. Reads the prior cursor from `record_sync_subscriptions` if 53 53 + * present; persists the new cursor as it advances. Auto-enrolls the space 54 54 + * locally on first connect using the source's `authorityDid`. */ 55 55 + export async function runRecordHostSync( 56 56 + source: RecordHostSyncSource, 57 57 + options: RecordHostSyncOptions 58 58 + ): Promise<void> { 59 59 + const fetchImpl = options.fetch ?? globalThis.fetch; 60 60 + await ensureSyncSchema(options.db); 61 61 + 62 62 + // Auto-enroll the space so the local record host accepts subsequent 63 63 + // queries against it. 64 64 + const existing = await options.recordHost.getEnrollment(source.spaceUri); 65 65 + if (!existing) { 66 66 + await options.recordHost.enroll({ 67 67 + spaceUri: source.spaceUri, 68 68 + authorityDid: source.authorityDid, 69 69 + enrolledAt: Date.now(), 70 70 + enrolledBy: source.authorityDid, 71 71 + }); 72 72 + } 73 73 + 74 74 + // Resume from the last persisted cursor for this subscription. 75 75 + const since = await readCursor(options.db, source.hostUrl, source.spaceUri); 76 76 + 77 77 + const endpoint = 78 78 + source.endpointNsid ?? `${options.config.namespace}.recordHost.sync`; 79 79 + const url = new URL(`${source.hostUrl}/xrpc/${endpoint}`); 80 80 + url.searchParams.set("spaceUri", source.spaceUri); 81 81 + if (since) url.searchParams.set("since", since); 82 82 + 83 83 + const res = await fetchImpl(url.toString(), { 84 84 + headers: { 85 85 + "X-Space-Credential": source.credential, 86 86 + accept: "text/event-stream", 87 87 + }, 88 88 + signal: options.signal, 89 89 + }); 90 90 + if (!res.ok) { 91 91 + const body = await res.text().catch(() => ""); 92 92 + throw new Error(`recordHost.sync ${res.status}: ${body}`); 93 93 + } 94 94 + if (!res.body) { 95 95 + throw new Error("recordHost.sync response has no body"); 96 96 + } 97 97 + 98 98 + const reader = res.body.getReader(); 99 99 + const decoder = new TextDecoder(); 100 100 + let buf = ""; 101 101 + try { 102 102 + while (true) { 103 103 + const { done, value } = await reader.read(); 104 104 + if (done) break; 105 105 + buf += decoder.decode(value, { stream: true }); 106 106 + while (true) { 107 107 + const idx = buf.indexOf("\n\n"); 108 108 + if (idx < 0) break; 109 109 + const block = buf.slice(0, idx); 110 110 + buf = buf.slice(idx + 2); 111 111 + const dataLine = block.split("\n").find((l) => l.startsWith("data:")); 112 112 + if (!dataLine) continue; 113 113 + const json = dataLine.slice(5).trim(); 114 114 + if (!json) continue; 115 115 + let event: any; 116 116 + try { 117 117 + event = JSON.parse(json); 118 118 + } catch { 119 119 + continue; 120 120 + } 121 121 + await applyEvent(event, options); 122 122 + if (event.kind === "cursor" && typeof event.value === "string") { 123 123 + await persistCursor( 124 124 + options.db, 125 125 + source.hostUrl, 126 126 + source.spaceUri, 127 127 + event.value 128 128 + ); 129 129 + options.onCursor?.(event.value); 130 130 + } 131 131 + } 132 132 + } 133 133 + } finally { 134 134 + await reader.cancel().catch(() => {}); 135 135 + } 136 136 + } 137 137 + 138 138 + async function applyEvent( 139 139 + event: any, 140 140 + options: RecordHostSyncOptions 141 141 + ): Promise<void> { 142 142 + if (event.kind === "record.created") { 143 143 + const p = event.payload; 144 144 + await options.recordHost.putRecord({ 145 145 + spaceUri: p.space, 146 146 + collection: p.collection, 147 147 + authorDid: p.did, 148 148 + rkey: p.rkey, 149 149 + cid: p.cid ?? null, 150 150 + record: p.record ?? {}, 151 151 + // The host's time_us is microseconds; createdAt on putRecord is ms-ish 152 152 + // historically. Keep the host's ordering by passing it through; the 153 153 + // host adapter writes time_us = createdAt * 1000 internally so this 154 154 + // round-trips. We store the source's time_us directly to preserve 155 155 + // ordering across hosts. 156 156 + createdAt: p.time_us != null ? Math.floor(p.time_us / 1000) : Date.now(), 157 157 + }); 158 158 + } else if (event.kind === "record.deleted") { 159 159 + const p = event.payload; 160 160 + await options.recordHost.deleteRecord(p.space, p.collection, p.did, p.rkey); 161 161 + } 162 162 + // cursor events are handled by the caller for persistence 163 163 + } 164 164 + 165 165 + // ---- Schema + cursor persistence ---- 166 166 + 167 167 + const SYNC_SCHEMA_APPLIED = new WeakSet<object>(); 168 168 + 169 169 + /** Idempotent: applies the `record_sync_subscriptions` table on first call 170 170 + * for a given DB. Tracks per-DB (by reference) so repeated calls in tests 171 171 + * don't re-issue DDL each time. */ 172 172 + async function ensureSyncSchema(db: Database): Promise<void> { 173 173 + if (SYNC_SCHEMA_APPLIED.has(db as unknown as object)) return; 174 174 + const dialect = getDialect(db); 175 175 + const stmts = buildRecordSyncSchema(dialect); 176 176 + await db.batch(stmts.map((s) => db.prepare(s))); 177 177 + SYNC_SCHEMA_APPLIED.add(db as unknown as object); 178 178 + } 179 179 + 180 180 + export function buildRecordSyncSchema(dialect: SqlDialect): string[] { 181 181 + return [ 182 182 + `CREATE TABLE IF NOT EXISTS record_sync_subscriptions ( 183 183 + host_url TEXT NOT NULL, 184 184 + space_uri TEXT NOT NULL, 185 185 + cursor TEXT, 186 186 + last_synced_at ${dialect.bigintType}, 187 187 + PRIMARY KEY (host_url, space_uri) 188 188 + )`, 189 189 + ]; 190 190 + } 191 191 + 192 192 + /** SchemaModule-shaped helper for `initSchema({ extraSchemas: [...] })`. */ 193 193 + export async function applyRecordSyncSchema(db: Database): Promise<void> { 194 194 + const dialect = getDialect(db); 195 195 + const stmts = buildRecordSyncSchema(dialect); 196 196 + await db.batch(stmts.map((s) => db.prepare(s))); 197 197 + } 198 198 + 199 199 + async function readCursor( 200 200 + db: Database, 201 201 + hostUrl: string, 202 202 + spaceUri: string 203 203 + ): Promise<string | null> { 204 204 + const row = await db 205 205 + .prepare( 206 206 + `SELECT cursor FROM record_sync_subscriptions WHERE host_url = ? AND space_uri = ?` 207 207 + ) 208 208 + .bind(hostUrl, spaceUri) 209 209 + .first<{ cursor: string | null } | null>(); 210 210 + return row?.cursor ?? null; 211 211 + } 212 212 + 213 213 + async function persistCursor( 214 214 + db: Database, 215 215 + hostUrl: string, 216 216 + spaceUri: string, 217 217 + cursor: string 218 218 + ): Promise<void> { 219 219 + await db 220 220 + .prepare( 221 221 + `INSERT INTO record_sync_subscriptions (host_url, space_uri, cursor, last_synced_at) 222 222 + VALUES (?, ?, ?, ?) 223 223 + ON CONFLICT (host_url, space_uri) DO UPDATE SET 224 224 + cursor = excluded.cursor, 225 225 + last_synced_at = excluded.last_synced_at` 226 226 + ) 227 227 + .bind(hostUrl, spaceUri, cursor, Date.now()) 228 228 + .run(); 229 229 + }

+3

packages/contrail-record-host/src/index.ts

Reviewed

··· 25 25 export { collectBlobCids } from "./blob-refs"; 26 26 27 27 export { registerRecordHostRoutes } from "./routes"; 28 28 + 29 29 + export { registerRecordHostSyncRoutes } from "./sync"; 30 30 + export type { RecordHostSyncOptions, SyncEvent } from "./sync";

+331

packages/contrail-record-host/src/sync.ts

Reviewed

··· 1 1 + /** Record-host sync endpoint — streams record events for a specific space. 2 2 + * 3 3 + * Two phases per connection: 4 4 + * 1. **Catch-up**: scan the per-collection `spaces_records_<short>` tables 5 5 + * for `time_us > since`, emit each row as `record.created`. After each 6 6 + * batch, emit a `cursor` checkpoint so the client can persist progress. 7 7 + * 2. **Live**: subscribe to the in-process pubsub for `space:<uri>` and 8 8 + * forward record.created / record.deleted events. 9 9 + * 10 10 + * Phase 7b MVP limitations: 11 11 + * - Catch-up only sees `record.created`; deletions in the past aren't 12 12 + * replayed (the row is gone). Live deletions are emitted. 13 13 + * - Brief race between catch-up end and live subscribe — if a write 14 14 + * lands in that window, the next reconnect catches it via catch-up. 15 15 + * - SSE only (no WS). Simpler. WS can be added if needed. 16 16 + * 17 17 + * Auth: requires a valid X-Space-Credential whose `space` claim matches 18 18 + * the requested spaceUri and whose scope is `read` or `rw`. */ 19 19 + 20 20 + import type { Hono, MiddlewareHandler } from "hono"; 21 21 + import type { 22 22 + ContrailConfig, 23 23 + CredentialClaims, 24 24 + CredentialVerifier, 25 25 + PubSub, 26 26 + RealtimeEvent, 27 27 + RecordHost, 28 28 + } from "@atmo-dev/contrail-base"; 29 29 + import { 30 30 + DEFAULT_KEEPALIVE_MS, 31 31 + extractSpaceCredential, 32 32 + shortNameForNsid, 33 33 + spacesRecordsTableName, 34 34 + spaceTopic, 35 35 + } from "@atmo-dev/contrail-base"; 36 36 + import type { Database } from "@atmo-dev/contrail-base"; 37 37 + 38 38 + /** Events emitted on the wire. RealtimeEvent kinds (record.created / 39 39 + * record.deleted) plus our own `cursor` checkpoint. */ 40 40 + export type SyncEvent = 41 41 + | RealtimeEvent 42 42 + | { kind: "cursor"; value: string }; 43 43 + 44 44 + export interface RecordHostSyncOptions { 45 45 + /** Database the host's record tables live on. */ 46 46 + db: Database; 47 47 + /** Optional pubsub for live mode. When omitted, sync is catch-up only — 48 48 + * the stream ends after catch-up rather than tailing for new writes. */ 49 49 + pubsub?: PubSub | null; 50 50 + /** Required: verifier for the X-Space-Credential header. */ 51 51 + credentialVerifier: CredentialVerifier; 52 52 + /** Page size for catch-up scans. Default 100. */ 53 53 + batchSize?: number; 54 54 + /** SSE keepalive interval in ms. Default uses the realtime module's. */ 55 55 + keepaliveMs?: number; 56 56 + } 57 57 + 58 58 + export function registerRecordHostSyncRoutes( 59 59 + app: Hono, 60 60 + recordHost: RecordHost, 61 61 + config: ContrailConfig, 62 62 + options: RecordHostSyncOptions 63 63 + ): void { 64 64 + const RECORD_HOST = `${config.namespace}.recordHost`; 65 65 + const batchSize = options.batchSize ?? 100; 66 66 + const keepaliveMs = options.keepaliveMs ?? DEFAULT_KEEPALIVE_MS; 67 67 + 68 68 + app.get(`/xrpc/${RECORD_HOST}.sync`, async (c) => { 69 69 + // ---- Auth: credential required ---- 70 70 + const credToken = extractSpaceCredential(c.req.raw); 71 71 + if (!credToken) { 72 72 + return c.json( 73 73 + { error: "AuthRequired", reason: "credential-required" }, 74 74 + 401 75 75 + ); 76 76 + } 77 77 + const verified = await options.credentialVerifier.verify(credToken); 78 78 + if (!verified.ok) { 79 79 + return c.json({ error: "AuthRequired", reason: verified.reason }, 401); 80 80 + } 81 81 + const claims = verified.claims; 82 82 + 83 83 + const spaceUri = c.req.query("spaceUri"); 84 84 + if (!spaceUri) { 85 85 + return c.json( 86 86 + { error: "InvalidRequest", message: "spaceUri required" }, 87 87 + 400 88 88 + ); 89 89 + } 90 90 + if (claims.space !== spaceUri) { 91 91 + return c.json( 92 92 + { error: "Forbidden", reason: "credential-wrong-space" }, 93 93 + 403 94 94 + ); 95 95 + } 96 96 + 97 97 + const enrollment = await recordHost.getEnrollment(spaceUri); 98 98 + if (!enrollment) { 99 99 + return c.json( 100 100 + { error: "NotFound", reason: "not-enrolled" }, 101 101 + 404 102 102 + ); 103 103 + } 104 104 + 105 105 + const since = parseSince(c.req.query("since")); 106 106 + 107 107 + // Build the SSE response with a hand-rolled stream so we can interleave 108 108 + // catch-up batches and live events under one cursor sequence. 109 109 + const ac = new AbortController(); 110 110 + c.req.raw.signal.addEventListener("abort", () => ac.abort(), { once: true }); 111 111 + 112 112 + const stream = new ReadableStream<Uint8Array>({ 113 113 + async start(controller) { 114 114 + const enc = new TextEncoder(); 115 115 + let closed = false; 116 116 + const close = () => { 117 117 + if (closed) return; 118 118 + closed = true; 119 119 + try { controller.close(); } catch { /* already closed */ } 120 120 + }; 121 121 + ac.signal.addEventListener("abort", close, { once: true }); 122 122 + 123 123 + const keepalive = setInterval(() => { 124 124 + if (closed) return; 125 125 + try { 126 126 + controller.enqueue(enc.encode(`: keepalive\n\n`)); 127 127 + } catch { 128 128 + close(); 129 129 + } 130 130 + }, keepaliveMs); 131 131 + 132 132 + const writeEvent = (e: SyncEvent) => { 133 133 + if (closed) return false; 134 134 + try { 135 135 + controller.enqueue(enc.encode(frameEvent(e))); 136 136 + return true; 137 137 + } catch { 138 138 + close(); 139 139 + return false; 140 140 + } 141 141 + }; 142 142 + 143 143 + try { 144 144 + controller.enqueue(enc.encode(`: open\n\n`)); 145 145 + 146 146 + // ---- Phase 1: catch-up ---- 147 147 + const lastCursor = await streamCatchup({ 148 148 + db: options.db, 149 149 + config, 150 150 + spaceUri, 151 151 + since, 152 152 + batchSize, 153 153 + writeEvent, 154 154 + isClosed: () => closed, 155 155 + }); 156 156 + if (closed) return; 157 157 + 158 158 + // ---- Phase 2: live ---- 159 159 + if (options.pubsub) { 160 160 + const liveCutoff = lastCursor ?? since; 161 161 + for await (const event of options.pubsub.subscribe( 162 162 + spaceTopic(spaceUri), 163 163 + ac.signal 164 164 + )) { 165 165 + if (closed) break; 166 166 + // Ignore events older than what catch-up already covered. 167 167 + const ts = (event as any).payload?.time_us ?? null; 168 168 + if ( 169 169 + liveCutoff != null && 170 170 + typeof ts === "number" && 171 171 + ts <= liveCutoff 172 172 + ) { 173 173 + continue; 174 174 + } 175 175 + if ( 176 176 + event.kind !== "record.created" && 177 177 + event.kind !== "record.deleted" 178 178 + ) { 179 179 + continue; 180 180 + } 181 181 + const ok = writeEvent(event); 182 182 + if (!ok) break; 183 183 + // Emit a cursor checkpoint after each live event so consumers 184 184 + // can resume from the latest seen point. 185 185 + if (typeof ts === "number") { 186 186 + writeEvent({ kind: "cursor", value: String(ts) }); 187 187 + } 188 188 + } 189 189 + } 190 190 + } catch (err) { 191 191 + if (!closed) { 192 192 + try { 193 193 + controller.enqueue( 194 194 + enc.encode( 195 195 + `event: error\ndata: ${JSON.stringify({ 196 196 + message: err instanceof Error ? err.message : String(err), 197 197 + })}\n\n` 198 198 + ) 199 199 + ); 200 200 + } catch { 201 201 + /* already torn down */ 202 202 + } 203 203 + } 204 204 + } finally { 205 205 + clearInterval(keepalive); 206 206 + close(); 207 207 + } 208 208 + }, 209 209 + cancel() { 210 210 + ac.abort(); 211 211 + }, 212 212 + }); 213 213 + 214 214 + return new Response(stream, { 215 215 + status: 200, 216 216 + headers: { 217 217 + "content-type": "text/event-stream", 218 218 + "cache-control": "no-cache, no-transform", 219 219 + connection: "keep-alive", 220 220 + "x-accel-buffering": "no", 221 221 + }, 222 222 + }); 223 223 + }); 224 224 + } 225 225 + 226 226 + /** Catch-up phase: scan every configured per-collection spaces_records table 227 227 + * for rows with `time_us > since`, emit them in time_us order. Returns the 228 228 + * highest time_us emitted (or null if nothing emitted). */ 229 229 + async function streamCatchup(args: { 230 230 + db: Database; 231 231 + config: ContrailConfig; 232 232 + spaceUri: string; 233 233 + since: number | null; 234 234 + batchSize: number; 235 235 + writeEvent: (e: SyncEvent) => boolean; 236 236 + isClosed: () => boolean; 237 237 + }): Promise<number | null> { 238 238 + const { db, config, spaceUri, since, batchSize, writeEvent, isClosed } = args; 239 239 + let highest = since; 240 240 + 241 241 + for (const [_short, colCfg] of Object.entries(config.collections)) { 242 242 + if (colCfg.allowInSpaces === false) continue; 243 243 + if (isClosed()) return highest; 244 244 + 245 245 + const collectionNsid = colCfg.collection; 246 246 + const short = shortNameForNsid(config, collectionNsid); 247 247 + if (!short) continue; 248 248 + const table = spacesRecordsTableName(short); 249 249 + 250 250 + let cursor = since; 251 251 + while (true) { 252 252 + if (isClosed()) return highest; 253 253 + 254 254 + let rows: any[]; 255 255 + try { 256 256 + const sql = 257 257 + cursor != null 258 258 + ? `SELECT * FROM ${table} WHERE space_uri = ? AND time_us > ? ORDER BY time_us LIMIT ?` 259 259 + : `SELECT * FROM ${table} WHERE space_uri = ? ORDER BY time_us LIMIT ?`; 260 260 + const result = await (cursor != null 261 261 + ? db.prepare(sql).bind(spaceUri, cursor, batchSize).all<any>() 262 262 + : db.prepare(sql).bind(spaceUri, batchSize).all<any>()); 263 263 + rows = result.results; 264 264 + } catch { 265 265 + // Table missing — skip this collection silently. 266 266 + break; 267 267 + } 268 268 + 269 269 + if (rows.length === 0) break; 270 270 + 271 271 + for (const row of rows) { 272 272 + if (isClosed()) return highest; 273 273 + const time_us = numericish(row.time_us); 274 274 + const event: RealtimeEvent = { 275 275 + topic: spaceTopic(spaceUri), 276 276 + kind: "record.created", 277 277 + payload: { 278 278 + uri: row.uri, 279 279 + did: row.did, 280 280 + collection: collectionNsid, 281 281 + rkey: row.rkey, 282 282 + cid: row.cid ?? null, 283 283 + record: parseRecordJson(row.record), 284 284 + time_us, 285 285 + space: spaceUri, 286 286 + }, 287 287 + ts: Date.now(), 288 288 + }; 289 289 + if (!writeEvent(event)) return highest; 290 290 + if (highest == null || time_us > highest) highest = time_us; 291 291 + cursor = time_us; 292 292 + } 293 293 + 294 294 + // Cursor checkpoint after the batch. 295 295 + if (highest != null) { 296 296 + writeEvent({ kind: "cursor", value: String(highest) }); 297 297 + } 298 298 + 299 299 + if (rows.length < batchSize) break; 300 300 + } 301 301 + } 302 302 + 303 303 + return highest; 304 304 + } 305 305 + 306 306 + function frameEvent(event: SyncEvent): string { 307 307 + return `event: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`; 308 308 + } 309 309 + 310 310 + function parseSince(raw: string | undefined): number | null { 311 311 + if (!raw) return null; 312 312 + const n = Number(raw); 313 313 + if (Number.isNaN(n) || !Number.isFinite(n)) return null; 314 314 + return n; 315 315 + } 316 316 + 317 317 + function numericish(v: unknown): number { 318 318 + return typeof v === "string" ? Number(v) : (v as number); 319 319 + } 320 320 + 321 321 + function parseRecordJson(value: unknown): Record<string, unknown> { 322 322 + if (value == null) return {}; 323 323 + if (typeof value === "string") { 324 324 + try { 325 325 + return JSON.parse(value) as Record<string, unknown>; 326 326 + } catch { 327 327 + return {}; 328 328 + } 329 329 + } 330 330 + return value as Record<string, unknown>; 331 331 + }

+304

packages/contrail/tests/sync-e2e.test.ts

Reviewed

··· 1 1 + /** End-to-end multi-host sync test. 2 2 + * 3 3 + * Two Contrail instances (host A and appview B) running in one process, 4 4 + * each with its own SQLite DB. Host A creates a space and writes records. 5 5 + * Appview B opens a sync stream against host A and ingests the records 6 6 + * into its own tables. We verify B can query the records locally. 7 7 + * 8 8 + * The "network" between them is host A's hono fetch, threaded through 9 9 + * appview B's fetch parameter. Real deployments would use the actual 10 10 + * fetch over HTTPS. */ 11 11 + 12 12 + import { describe, it, expect, beforeAll } from "vitest"; 13 13 + import { Hono } from "hono"; 14 14 + import { createSqliteDatabase } from "../src/adapters/sqlite"; 15 15 + import { initSchema } from "../src/core/db/schema"; 16 16 + import { resolveConfig } from "../src/core/types"; 17 17 + import type { ContrailConfig } from "../src/core/types"; 18 18 + import { HostedAdapter } from "../src/core/spaces/adapter"; 19 19 + import { registerRecordHostSyncRoutes } from "@atmo-dev/contrail-record-host"; 20 20 + import { runRecordHostSync, applyRecordSyncSchema } from "@atmo-dev/contrail-appview"; 21 21 + import { 22 22 + generateAuthoritySigningKey, 23 23 + issueCredential, 24 24 + createInProcessVerifier, 25 25 + InMemoryPubSub, 26 26 + } from "@atmo-dev/contrail-base"; 27 27 + import type { CredentialKeyMaterial } from "@atmo-dev/contrail-base"; 28 28 + import { wrapWithPublishing } from "../src/core/realtime/publishing-adapter"; 29 29 + 30 30 + const ALICE = "did:plc:alice"; 31 31 + const SERVICE_DID = "did:web:test.example#svc"; 32 32 + const SPACE_TYPE = "tools.atmo.event.space"; 33 33 + const SPACE_KEY = "main"; 34 34 + const SPACE_URI = `ats://${ALICE}/${SPACE_TYPE}/${SPACE_KEY}`; 35 35 + 36 36 + let SIGNING: CredentialKeyMaterial; 37 37 + 38 38 + beforeAll(async () => { 39 39 + SIGNING = await generateAuthoritySigningKey(); 40 40 + }); 41 41 + 42 42 + const HOST_CONFIG: ContrailConfig = { 43 43 + namespace: "test.sync", 44 44 + collections: { message: { collection: "app.event.message" } }, 45 45 + spaces: { 46 46 + authority: { type: SPACE_TYPE, serviceDid: SERVICE_DID }, 47 47 + recordHost: {}, 48 48 + }, 49 49 + }; 50 50 + 51 51 + async function makeHost(): Promise<{ 52 52 + app: Hono; 53 53 + adapter: HostedAdapter; 54 54 + db: any; 55 55 + }> { 56 56 + const db = createSqliteDatabase(":memory:"); 57 57 + const cfg = { ...HOST_CONFIG }; 58 58 + cfg.spaces!.authority!.signing = SIGNING; 59 59 + const resolved = resolveConfig(cfg); 60 60 + await initSchema(db, resolved); 61 61 + 62 62 + const baseAdapter = new HostedAdapter(db, resolved); 63 63 + const pubsub = new InMemoryPubSub(); 64 64 + const adapter = wrapWithPublishing(baseAdapter, pubsub) as HostedAdapter; 65 65 + 66 66 + // Provision a space + enroll on this host. 67 67 + await adapter.createSpace({ 68 68 + uri: SPACE_URI, 69 69 + ownerDid: ALICE, 70 70 + type: SPACE_TYPE, 71 71 + key: SPACE_KEY, 72 72 + serviceDid: SERVICE_DID, 73 73 + appPolicyRef: null, 74 74 + appPolicy: null, 75 75 + }); 76 76 + await adapter.addMember(SPACE_URI, ALICE, ALICE); 77 77 + await adapter.enroll({ 78 78 + spaceUri: SPACE_URI, 79 79 + authorityDid: SERVICE_DID, 80 80 + enrolledAt: Date.now(), 81 81 + enrolledBy: ALICE, 82 82 + }); 83 83 + 84 84 + const app = new Hono(); 85 85 + const verifier = createInProcessVerifier({ 86 86 + authorityDid: SERVICE_DID, 87 87 + publicKey: SIGNING.publicKey, 88 88 + }); 89 89 + registerRecordHostSyncRoutes(app, adapter, resolved, { 90 90 + db, 91 91 + pubsub, 92 92 + credentialVerifier: verifier, 93 93 + keepaliveMs: 60_000, 94 94 + batchSize: 50, 95 95 + }); 96 96 + 97 97 + return { app, adapter, db }; 98 98 + } 99 99 + 100 100 + async function makeAppview(): Promise<{ db: any; adapter: HostedAdapter; resolved: any }> { 101 101 + const db = createSqliteDatabase(":memory:"); 102 102 + const cfg = { ...HOST_CONFIG }; 103 103 + cfg.spaces!.authority!.signing = SIGNING; 104 104 + const resolved = resolveConfig(cfg); 105 105 + await initSchema(db, resolved, { 106 106 + extraSchemas: [applyRecordSyncSchema], 107 107 + }); 108 108 + const adapter = new HostedAdapter(db, resolved); 109 109 + return { db, adapter, resolved }; 110 110 + } 111 111 + 112 112 + async function mintCredentialForAlice(): Promise<string> { 113 113 + const { credential } = await issueCredential( 114 114 + { 115 115 + iss: SERVICE_DID, 116 116 + sub: ALICE, 117 117 + space: SPACE_URI, 118 118 + scope: "rw", 119 119 + ttlMs: 60_000, 120 120 + }, 121 121 + SIGNING 122 122 + ); 123 123 + return credential; 124 124 + } 125 125 + 126 126 + describe("recordHost.sync — end-to-end (host → appview)", () => { 127 127 + it("appview ingests historical records into its own tables", async () => { 128 128 + const host = await makeHost(); 129 129 + const appview = await makeAppview(); 130 130 + 131 131 + // Plant 3 records on the host. 132 132 + const t0 = Date.now(); 133 133 + for (let i = 0; i < 3; i++) { 134 134 + await host.adapter.putRecord({ 135 135 + spaceUri: SPACE_URI, 136 136 + collection: "app.event.message", 137 137 + authorDid: ALICE, 138 138 + rkey: `rk${i}`, 139 139 + cid: null, 140 140 + record: { $type: "app.event.message", text: `msg-${i}` }, 141 141 + createdAt: t0 + i, 142 142 + }); 143 143 + } 144 144 + 145 145 + // Appview opens a sync stream with a 200ms abort — long enough for 146 146 + // catch-up to drain, then we tear down before the live phase blocks. 147 147 + const ac = new AbortController(); 148 148 + const credential = await mintCredentialForAlice(); 149 149 + const cursors: string[] = []; 150 150 + 151 151 + const syncPromise = runRecordHostSync( 152 152 + { 153 153 + hostUrl: "http://host", 154 154 + spaceUri: SPACE_URI, 155 155 + authorityDid: SERVICE_DID, 156 156 + credential, 157 157 + }, 158 158 + { 159 159 + db: appview.db, 160 160 + config: appview.resolved, 161 161 + recordHost: appview.adapter, 162 162 + // Route fetch to host A's hono app instead of the network. 163 163 + fetch: ((input: any, init?: any) => { 164 164 + const req = typeof input === "string" || input instanceof URL 165 165 + ? new Request(input, init) 166 166 + : (input as Request); 167 167 + return host.app.fetch(req); 168 168 + }) as typeof fetch, 169 169 + signal: ac.signal, 170 170 + onCursor: (c) => { 171 171 + cursors.push(c); 172 172 + // Once we've seen at least one cursor checkpoint, catch-up is 173 173 + // making progress. Schedule abort to break out of live mode. 174 174 + setTimeout(() => ac.abort(), 50); 175 175 + }, 176 176 + } 177 177 + ).catch((err) => { 178 178 + // AbortError is expected; rethrow other errors. 179 179 + if (err.name !== "AbortError") throw err; 180 180 + }); 181 181 + 182 182 + await syncPromise; 183 183 + 184 184 + // Verify the appview's local tables now have the host's records. 185 185 + const ingested = await appview.adapter.listRecords( 186 186 + SPACE_URI, 187 187 + "app.event.message" 188 188 + ); 189 189 + expect(ingested.records).toHaveLength(3); 190 190 + expect(ingested.records.map((r) => r.rkey).sort()).toEqual([ 191 191 + "rk0", 192 192 + "rk1", 193 193 + "rk2", 194 194 + ]); 195 195 + expect(cursors.length).toBeGreaterThan(0); 196 196 + 197 197 + // Cursor was persisted in the subscriptions table. 198 198 + const subRow = await appview.db 199 199 + .prepare( 200 200 + `SELECT cursor FROM record_sync_subscriptions WHERE host_url = ? AND space_uri = ?` 201 201 + ) 202 202 + .bind("http://host", SPACE_URI) 203 203 + .first<{ cursor: string | null }>(); 204 204 + expect(subRow?.cursor).toBeTruthy(); 205 205 + 206 206 + // Auto-enrolled on the appview side. 207 207 + const enrollment = await appview.adapter.getEnrollment(SPACE_URI); 208 208 + expect(enrollment).toBeTruthy(); 209 209 + expect(enrollment?.authorityDid).toBe(SERVICE_DID); 210 210 + }); 211 211 + 212 212 + it("appview resumes from persisted cursor — no re-emission of old records", async () => { 213 213 + const host = await makeHost(); 214 214 + const appview = await makeAppview(); 215 215 + const credential = await mintCredentialForAlice(); 216 216 + 217 217 + // First batch. 218 218 + const t0 = Date.now(); 219 219 + await host.adapter.putRecord({ 220 220 + spaceUri: SPACE_URI, 221 221 + collection: "app.event.message", 222 222 + authorDid: ALICE, 223 223 + rkey: "first", 224 224 + cid: null, 225 225 + record: { text: "first" }, 226 226 + createdAt: t0, 227 227 + }); 228 228 + 229 229 + const networkFetch = ((input: any, init?: any) => { 230 230 + const req = typeof input === "string" || input instanceof URL 231 231 + ? new Request(input, init) 232 232 + : (input as Request); 233 233 + return host.app.fetch(req); 234 234 + }) as typeof fetch; 235 235 + 236 236 + const ac1 = new AbortController(); 237 237 + const seenCount: { count: number } = { count: 0 }; 238 238 + await runRecordHostSync( 239 239 + { 240 240 + hostUrl: "http://host", 241 241 + spaceUri: SPACE_URI, 242 242 + authorityDid: SERVICE_DID, 243 243 + credential, 244 244 + }, 245 245 + { 246 246 + db: appview.db, 247 247 + config: appview.resolved, 248 248 + recordHost: appview.adapter, 249 249 + fetch: networkFetch, 250 250 + signal: ac1.signal, 251 251 + onCursor: () => { 252 252 + seenCount.count++; 253 253 + setTimeout(() => ac1.abort(), 30); 254 254 + }, 255 255 + } 256 256 + ).catch((err) => { 257 257 + if (err.name !== "AbortError") throw err; 258 258 + }); 259 259 + 260 260 + // Insert second batch on the host AFTER the first sync completed. 261 261 + await host.adapter.putRecord({ 262 262 + spaceUri: SPACE_URI, 263 263 + collection: "app.event.message", 264 264 + authorDid: ALICE, 265 265 + rkey: "second", 266 266 + cid: null, 267 267 + record: { text: "second" }, 268 268 + createdAt: t0 + 100, 269 269 + }); 270 270 + 271 271 + // Second sync run — should pick up only the new record, using the 272 272 + // persisted cursor. 273 273 + const ac2 = new AbortController(); 274 274 + const ingestedRkeys: string[] = []; 275 275 + // Wrap putRecord to observe what gets re-ingested. 276 276 + const origPut = appview.adapter.putRecord.bind(appview.adapter); 277 277 + appview.adapter.putRecord = async (record) => { 278 278 + ingestedRkeys.push(record.rkey); 279 279 + return origPut(record); 280 280 + }; 281 281 + 282 282 + await runRecordHostSync( 283 283 + { 284 284 + hostUrl: "http://host", 285 285 + spaceUri: SPACE_URI, 286 286 + authorityDid: SERVICE_DID, 287 287 + credential, 288 288 + }, 289 289 + { 290 290 + db: appview.db, 291 291 + config: appview.resolved, 292 292 + recordHost: appview.adapter, 293 293 + fetch: networkFetch, 294 294 + signal: ac2.signal, 295 295 + onCursor: () => setTimeout(() => ac2.abort(), 30), 296 296 + } 297 297 + ).catch((err) => { 298 298 + if (err.name !== "AbortError") throw err; 299 299 + }); 300 300 + 301 301 + // The second run should have ingested ONLY "second" (not re-ingested "first"). 302 302 + expect(ingestedRkeys).toEqual(["second"]); 303 303 + }); 304 304 + });

+282

packages/contrail/tests/sync-host.test.ts

Reviewed

··· 1 1 + /** Tests for the recordHost.sync SSE endpoint. */ 2 2 + 3 3 + import { describe, it, expect, beforeAll } from "vitest"; 4 4 + import { Hono } from "hono"; 5 5 + import type { MiddlewareHandler } from "hono"; 6 6 + import { createSqliteDatabase } from "../src/adapters/sqlite"; 7 7 + import { initSchema } from "../src/core/db/schema"; 8 8 + import { resolveConfig } from "../src/core/types"; 9 9 + import type { ContrailConfig } from "../src/core/types"; 10 10 + import { HostedAdapter } from "../src/core/spaces/adapter"; 11 11 + import { registerRecordHostSyncRoutes } from "@atmo-dev/contrail-record-host"; 12 12 + import { 13 13 + generateAuthoritySigningKey, 14 14 + issueCredential, 15 15 + createInProcessVerifier, 16 16 + InMemoryPubSub, 17 17 + spaceTopic, 18 18 + } from "@atmo-dev/contrail-base"; 19 19 + import type { CredentialKeyMaterial } from "@atmo-dev/contrail-base"; 20 20 + import { wrapWithPublishing } from "../src/core/realtime/publishing-adapter"; 21 21 + 22 22 + const ALICE = "did:plc:alice"; 23 23 + const SERVICE_DID = "did:web:test.example#svc"; 24 24 + const SPACE_TYPE = "tools.atmo.event.space"; 25 25 + 26 26 + let SIGNING: CredentialKeyMaterial; 27 27 + 28 28 + beforeAll(async () => { 29 29 + SIGNING = await generateAuthoritySigningKey(); 30 30 + }); 31 31 + 32 32 + const CONFIG: ContrailConfig = { 33 33 + namespace: "test.sync", 34 34 + collections: { message: { collection: "app.event.message" } }, 35 35 + spaces: { 36 36 + authority: { 37 37 + type: SPACE_TYPE, 38 38 + serviceDid: SERVICE_DID, 39 39 + signing: undefined as any, // filled in beforeAll 40 40 + }, 41 41 + recordHost: {}, 42 42 + }, 43 43 + }; 44 44 + 45 45 + async function makeHost(): Promise<{ 46 46 + app: Hono; 47 47 + adapter: HostedAdapter; 48 48 + pubsub: InMemoryPubSub; 49 49 + spaceUri: string; 50 50 + }> { 51 51 + const db = createSqliteDatabase(":memory:"); 52 52 + const cfg = { ...CONFIG }; 53 53 + cfg.spaces!.authority!.signing = SIGNING; 54 54 + const resolved = resolveConfig(cfg); 55 55 + await initSchema(db, resolved); 56 56 + 57 57 + const baseAdapter = new HostedAdapter(db, resolved); 58 58 + const pubsub = new InMemoryPubSub(); 59 59 + // Wrap so writes also publish onto pubsub topics — mirrors what the 60 60 + // umbrella createApp does in realtime mode. 61 61 + const adapter = wrapWithPublishing(baseAdapter, pubsub) as HostedAdapter; 62 62 + 63 63 + // Create a space + enroll 64 64 + const spaceUri = `ats://${ALICE}/${SPACE_TYPE}/main`; 65 65 + await adapter.createSpace({ 66 66 + uri: spaceUri, 67 67 + ownerDid: ALICE, 68 68 + type: SPACE_TYPE, 69 69 + key: "main", 70 70 + serviceDid: SERVICE_DID, 71 71 + appPolicyRef: null, 72 72 + appPolicy: null, 73 73 + }); 74 74 + await adapter.addMember(spaceUri, ALICE, ALICE); 75 75 + await adapter.enroll({ 76 76 + spaceUri, 77 77 + authorityDid: SERVICE_DID, 78 78 + enrolledAt: Date.now(), 79 79 + enrolledBy: ALICE, 80 80 + }); 81 81 + 82 82 + // Build the SSE app. 83 83 + const app = new Hono(); 84 84 + const verifier = createInProcessVerifier({ 85 85 + authorityDid: SERVICE_DID, 86 86 + publicKey: SIGNING.publicKey, 87 87 + }); 88 88 + registerRecordHostSyncRoutes(app, adapter, resolved, { 89 89 + db, 90 90 + pubsub, 91 91 + credentialVerifier: verifier, 92 92 + keepaliveMs: 60_000, 93 93 + }); 94 94 + 95 95 + return { app, adapter, pubsub, spaceUri }; 96 96 + } 97 97 + 98 98 + async function mintCredential(spaceUri: string): Promise<string> { 99 99 + const { credential } = await issueCredential( 100 100 + { 101 101 + iss: SERVICE_DID, 102 102 + sub: ALICE, 103 103 + space: spaceUri, 104 104 + scope: "rw", 105 105 + ttlMs: 60_000, 106 106 + }, 107 107 + SIGNING 108 108 + ); 109 109 + return credential; 110 110 + } 111 111 + 112 112 + /** Read the SSE response, return the first N parsed events. */ 113 113 + async function readEvents( 114 114 + res: Response, 115 115 + count: number, 116 116 + timeoutMs = 2000 117 117 + ): Promise<Array<{ kind: string; payload?: any; value?: string }>> { 118 118 + const reader = res.body!.getReader(); 119 119 + const decoder = new TextDecoder(); 120 120 + const events: Array<{ kind: string; payload?: any; value?: string }> = []; 121 121 + let buf = ""; 122 122 + const start = Date.now(); 123 123 + while (events.length < count) { 124 124 + if (Date.now() - start > timeoutMs) { 125 125 + throw new Error(`timeout waiting for ${count} events; got ${events.length}`); 126 126 + } 127 127 + const { done, value } = await reader.read(); 128 128 + if (done) break; 129 129 + buf += decoder.decode(value, { stream: true }); 130 130 + while (true) { 131 131 + const idx = buf.indexOf("\n\n"); 132 132 + if (idx < 0) break; 133 133 + const block = buf.slice(0, idx); 134 134 + buf = buf.slice(idx + 2); 135 135 + // Skip comment-only blocks (`: open`, `: keepalive`). 136 136 + const dataLine = block 137 137 + .split("\n") 138 138 + .find((l) => l.startsWith("data:")); 139 139 + if (!dataLine) continue; 140 140 + const json = dataLine.slice(5).trim(); 141 141 + if (!json) continue; 142 142 + events.push(JSON.parse(json)); 143 143 + if (events.length >= count) break; 144 144 + } 145 145 + } 146 146 + await reader.cancel().catch(() => {}); 147 147 + return events; 148 148 + } 149 149 + 150 150 + describe("recordHost.sync — catch-up phase", () => { 151 151 + it("emits historical records as record.created in time_us order", async () => { 152 152 + const { app, adapter, spaceUri } = await makeHost(); 153 153 + 154 154 + // Plant 3 records with deterministic timestamps. 155 155 + const now = Date.now(); 156 156 + for (let i = 0; i < 3; i++) { 157 157 + await adapter.putRecord({ 158 158 + spaceUri, 159 159 + collection: "app.event.message", 160 160 + authorDid: ALICE, 161 161 + rkey: `rk${i}`, 162 162 + cid: null, 163 163 + record: { $type: "app.event.message", text: `msg-${i}` }, 164 164 + createdAt: now + i, 165 165 + }); 166 166 + } 167 167 + 168 168 + const credential = await mintCredential(spaceUri); 169 169 + const res = await app.fetch( 170 170 + new Request( 171 171 + `http://localhost/xrpc/test.sync.recordHost.sync?spaceUri=${encodeURIComponent(spaceUri)}`, 172 172 + { headers: { "X-Space-Credential": credential } } 173 173 + ) 174 174 + ); 175 175 + expect(res.status).toBe(200); 176 176 + 177 177 + // Expect 3 record.created + at least one cursor checkpoint. 178 178 + const events = await readEvents(res, 4); 179 179 + const records = events.filter((e) => e.kind === "record.created"); 180 180 + expect(records).toHaveLength(3); 181 181 + expect(records.map((e) => e.payload.rkey)).toEqual(["rk0", "rk1", "rk2"]); 182 182 + expect(records.every((e) => e.payload.space === spaceUri)).toBe(true); 183 183 + const cursors = events.filter((e) => e.kind === "cursor"); 184 184 + expect(cursors.length).toBeGreaterThan(0); 185 185 + }); 186 186 + 187 187 + it("respects the since cursor — records at or before are skipped", async () => { 188 188 + const { app, adapter, spaceUri } = await makeHost(); 189 189 + 190 190 + const t0 = 1_700_000_000_000; 191 191 + await adapter.putRecord({ 192 192 + spaceUri, 193 193 + collection: "app.event.message", 194 194 + authorDid: ALICE, 195 195 + rkey: "old", 196 196 + cid: null, 197 197 + record: { text: "old" }, 198 198 + createdAt: t0, 199 199 + }); 200 200 + await adapter.putRecord({ 201 201 + spaceUri, 202 202 + collection: "app.event.message", 203 203 + authorDid: ALICE, 204 204 + rkey: "new", 205 205 + cid: null, 206 206 + record: { text: "new" }, 207 207 + createdAt: t0 + 100, 208 208 + }); 209 209 + 210 210 + const credential = await mintCredential(spaceUri); 211 211 + const res = await app.fetch( 212 212 + new Request( 213 213 + `http://localhost/xrpc/test.sync.recordHost.sync?spaceUri=${encodeURIComponent(spaceUri)}&since=${t0}`, 214 214 + { headers: { "X-Space-Credential": credential } } 215 215 + ) 216 216 + ); 217 217 + expect(res.status).toBe(200); 218 218 + 219 219 + const events = await readEvents(res, 2); 220 220 + const records = events.filter((e) => e.kind === "record.created"); 221 221 + expect(records).toHaveLength(1); 222 222 + expect(records[0]!.payload.rkey).toBe("new"); 223 223 + }); 224 224 + }); 225 225 + 226 226 + describe("recordHost.sync — auth + enrollment guards", () => { 227 227 + it("rejects without a credential", async () => { 228 228 + const { app, spaceUri } = await makeHost(); 229 229 + const res = await app.fetch( 230 230 + new Request( 231 231 + `http://localhost/xrpc/test.sync.recordHost.sync?spaceUri=${encodeURIComponent(spaceUri)}` 232 232 + ) 233 233 + ); 234 234 + expect(res.status).toBe(401); 235 235 + expect((await res.json() as any).reason).toBe("credential-required"); 236 236 + }); 237 237 + 238 238 + it("rejects credential whose space doesn't match", async () => { 239 239 + const { app, spaceUri } = await makeHost(); 240 240 + const wrongSpaceCred = await issueCredential( 241 241 + { 242 242 + iss: SERVICE_DID, 243 243 + sub: ALICE, 244 244 + space: "ats://did:plc:alice/x/y", 245 245 + scope: "rw", 246 246 + ttlMs: 60_000, 247 247 + }, 248 248 + SIGNING 249 249 + ); 250 250 + const res = await app.fetch( 251 251 + new Request( 252 252 + `http://localhost/xrpc/test.sync.recordHost.sync?spaceUri=${encodeURIComponent(spaceUri)}`, 253 253 + { headers: { "X-Space-Credential": wrongSpaceCred.credential } } 254 254 + ) 255 255 + ); 256 256 + expect(res.status).toBe(403); 257 257 + expect((await res.json() as any).reason).toBe("credential-wrong-space"); 258 258 + }); 259 259 + 260 260 + it("rejects un-enrolled spaces", async () => { 261 261 + const { app } = await makeHost(); 262 262 + const otherSpaceUri = `ats://${ALICE}/${SPACE_TYPE}/different`; 263 263 + const cred = await issueCredential( 264 264 + { 265 265 + iss: SERVICE_DID, 266 266 + sub: ALICE, 267 267 + space: otherSpaceUri, 268 268 + scope: "rw", 269 269 + ttlMs: 60_000, 270 270 + }, 271 271 + SIGNING 272 272 + ); 273 273 + const res = await app.fetch( 274 274 + new Request( 275 275 + `http://localhost/xrpc/test.sync.recordHost.sync?spaceUri=${encodeURIComponent(otherSpaceUri)}`, 276 276 + { headers: { "X-Space-Credential": cred.credential } } 277 277 + ) 278 278 + ); 279 279 + expect(res.status).toBe(404); 280 280 + expect((await res.json() as any).reason).toBe("not-enrolled"); 281 281 + }); 282 282 + });

+86

packages/lexicons/lexicon-templates/recordHost/sync.json

Reviewed

··· 1 1 + { 2 2 + "lexicon": 1, 3 3 + "id": "tools.atmo.recordHost.sync", 4 4 + "defs": { 5 5 + "main": { 6 6 + "type": "subscription", 7 7 + "description": "SSE stream of record events for a specific space. Emits a catch-up phase (every record with `time_us > since`, in time_us order) followed by a live phase (record.created / record.deleted as they happen). Cursor checkpoints are emitted after each catch-up batch and after each live event so consumers can persist progress and resume on reconnect. Auth: requires X-Space-Credential whose `space` claim matches `spaceUri`.", 8 8 + "parameters": { 9 9 + "type": "params", 10 10 + "required": ["spaceUri"], 11 11 + "properties": { 12 12 + "spaceUri": { 13 13 + "type": "string", 14 14 + "description": "Space to sync." 15 15 + }, 16 16 + "since": { 17 17 + "type": "string", 18 18 + "description": "Opaque cursor (last `value` from a `cursor` event). Server interprets as time_us; events with time_us > since are emitted." 19 19 + } 20 20 + } 21 21 + }, 22 22 + "message": { 23 23 + "schema": { 24 24 + "type": "union", 25 25 + "refs": [ 26 26 + "#recordCreated", 27 27 + "#recordDeleted", 28 28 + "#cursor" 29 29 + ] 30 30 + } 31 31 + } 32 32 + }, 33 33 + "recordCreated": { 34 34 + "type": "object", 35 35 + "required": ["topic", "kind", "payload", "ts"], 36 36 + "properties": { 37 37 + "topic": { "type": "string" }, 38 38 + "kind": { "type": "string", "const": "record.created" }, 39 39 + "payload": { 40 40 + "type": "object", 41 41 + "required": ["uri", "did", "collection", "rkey", "record", "time_us", "space"], 42 42 + "properties": { 43 43 + "uri": { "type": "string" }, 44 44 + "did": { "type": "string", "format": "did" }, 45 45 + "collection": { "type": "string", "format": "nsid" }, 46 46 + "rkey": { "type": "string" }, 47 47 + "cid": { "type": "string", "format": "cid" }, 48 48 + "record": { "type": "unknown" }, 49 49 + "time_us": { "type": "integer" }, 50 50 + "space": { "type": "string" } 51 51 + } 52 52 + }, 53 53 + "ts": { "type": "integer" } 54 54 + } 55 55 + }, 56 56 + "recordDeleted": { 57 57 + "type": "object", 58 58 + "required": ["topic", "kind", "payload", "ts"], 59 59 + "properties": { 60 60 + "topic": { "type": "string" }, 61 61 + "kind": { "type": "string", "const": "record.deleted" }, 62 62 + "payload": { 63 63 + "type": "object", 64 64 + "required": ["uri", "did", "collection", "rkey", "space"], 65 65 + "properties": { 66 66 + "uri": { "type": "string" }, 67 67 + "did": { "type": "string", "format": "did" }, 68 68 + "collection": { "type": "string", "format": "nsid" }, 69 69 + "rkey": { "type": "string" }, 70 70 + "space": { "type": "string" } 71 71 + } 72 72 + }, 73 73 + "ts": { "type": "integer" } 74 74 + } 75 75 + }, 76 76 + "cursor": { 77 77 + "type": "object", 78 78 + "required": ["kind", "value"], 79 79 + "description": "Checkpoint: persist `value` and pass it as `since` on next reconnect to resume.", 80 80 + "properties": { 81 81 + "kind": { "type": "string", "const": "cursor" }, 82 82 + "value": { "type": "string" } 83 83 + } 84 84 + } 85 85 + } 86 86 + }