From 4c4ab49004f66f85dd9f399a6cae371651cc91a3 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Tue, 4 Aug 2026 03:05:55 +0200 Subject: [PATCH] Fix adapter exports and stable pagination --- .changeset/lean-contrail.md | 2 +- .github/workflows/ci.yml | 3 + docs/02-querying.md | 2 +- package.json | 1 + packages/contrail/README.md | 2 +- packages/contrail/package.json | 1 + packages/contrail/src/core/db/records.ts | 192 +++++++++++++----- packages/contrail/src/core/dialect.ts | 9 +- packages/contrail/src/core/router/notify.ts | 84 +++++--- packages/contrail/tests/built-sqlite.mjs | 6 + packages/contrail/tests/notify.test.ts | 39 ++++ .../contrail/tests/query-pagination.test.ts | 141 +++++++++++++ packages/contrail/tests/records.test.ts | 51 ++++- packages/contrail/tests/search.test.ts | 59 ++++++ packages/contrail/tsup.config.ts | 1 + 15 files changed, 501 insertions(+), 92 deletions(-) create mode 100644 packages/contrail/tests/built-sqlite.mjs create mode 100644 packages/contrail/tests/query-pagination.test.ts diff --git a/.changeset/lean-contrail.md b/.changeset/lean-contrail.md index 1dcedb3..943bec0 100644 --- a/.changeset/lean-contrail.md +++ b/.changeset/lean-contrail.md @@ -2,4 +2,4 @@ "@atmo-dev/contrail": minor --- -Collapse Contrail into one public package and one AppView implementation. Remove the spaces, authority, record-host, community, realtime, sync, and custom Lexicon-tooling products. Route Jetstream, persistent, backfill, refresh, and immediate synchronization records through the shared `ingestRecords` admission and projection path. Make materialized relation counts converge when children arrive before parents or move during refresh, and prevent transient PDS failures from being interpreted as authoritative deletions. Keep dependent-subject filtering scoped to dependent collections and restore typed example XRPC clients with Atcute's generator. +Collapse Contrail into one public package and one AppView implementation. Remove the spaces, authority, record-host, community, realtime, sync, and custom Lexicon-tooling products. Route Jetstream, persistent, backfill, refresh, and immediate synchronization records through the shared `ingestRecords` admission and projection path. Make materialized relation counts converge when children arrive before parents or move during refresh, and prevent transient PDS failures from being interpreted as authoritative deletions. Keep dependent-subject filtering scoped to dependent collections and restore typed example XRPC clients with Atcute's generator. Preserve `node:sqlite` in the published adapter, make all query and search cursors stable across tied rows, use Worker-safe cursor encoding, and bound the complete notify resolution/fetch/body operation. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index af8aae8..d026ac1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,5 +44,8 @@ jobs: - name: Build run: pnpm build + - name: Test published adapters + run: pnpm test:built + - name: Test run: pnpm test diff --git a/docs/02-querying.md b/docs/02-querying.md index e8d288e..5d7d691 100644 --- a/docs/02-querying.md +++ b/docs/02-querying.md @@ -54,7 +54,7 @@ For count filters / sorts, the HTTP side is nicer than the programmatic side — ?limit=25&cursor= ``` -`cursor` is opaque — pass back whatever `listRecords` returned in its `cursor` field. `limit` is 1–200 (default 50). Cursors embed the sort kind, so a cursor from a `sort=startsAt` query is ignored by a `sort=rsvpsCount` query instead of silently returning wrong results. +`cursor` is opaque — pass back whatever `listRecords` returned in its `cursor` field. `limit` is 1–200 (default 50). Cursors embed the complete ordering, including relevance rank for search and URI as the final unique tiebreaker. They also embed the sort kind, so a cursor from a `sort=startsAt` query is ignored by a `sort=rsvpsCount` query instead of silently returning wrong results. ```ts let cursor: string | undefined; diff --git a/package.json b/package.json index bc8d3b8..7fb113a 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "build": "turbo run build --filter=./packages/*", "build:apps": "turbo run build", "test": "turbo run test", + "test:built": "pnpm --filter @atmo-dev/contrail test:built", "typecheck": "turbo run typecheck", "dev": "turbo run dev", "clean": "turbo run clean", diff --git a/packages/contrail/README.md b/packages/contrail/README.md index 3ac626a..dbc331d 100644 --- a/packages/contrail/README.md +++ b/packages/contrail/README.md @@ -88,4 +88,4 @@ import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite"; import { createPostgresDatabase } from "@atmo-dev/contrail/postgres"; ``` -D1 implements Contrail's database interface directly. +The SQLite adapter uses the built-in `node:sqlite` module and therefore requires Node.js 22.13 or newer. D1 implements Contrail's database interface directly. diff --git a/packages/contrail/package.json b/packages/contrail/package.json index ba1c135..928e8a1 100644 --- a/packages/contrail/package.json +++ b/packages/contrail/package.json @@ -61,6 +61,7 @@ "clean": "rm -rf dist", "typecheck": "tsc --noEmit", "test": "vitest run", + "test:built": "node tests/built-sqlite.mjs", "test:watch": "vitest" }, "dependencies": { diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts index 2d16a43..f3f26e1 100644 --- a/packages/contrail/src/core/db/records.ts +++ b/packages/contrail/src/core/db/records.ts @@ -783,13 +783,13 @@ export interface SortOption { direction: "asc" | "desc"; } -/** Opaque keyset cursor. `t` is the tiebreaker (time_us of the last row), - * `v` is the sort-key value (string for record fields, number for counts), - * `k` identifies the sort so we can reject mismatched cursors. */ +/** Opaque keyset cursor. `v` is the primary sort value, `t` is time, + * and `u` is the final unique tiebreaker. */ interface CursorPayload { t: number; + u: string; v?: string | number; - k: "time" | string; // "time" | `field:` | `count:` + k: "time" | "search" | string; } function sortKind(sort?: SortOption): "time" | string { @@ -798,15 +798,40 @@ function sortKind(sort?: SortOption): "time" | string { return "time"; } +function encodeBase64Url(value: string): string { + const bytes = new TextEncoder().encode(value); + let binary = ""; + for (let index = 0; index < bytes.length; index += 0x8000) { + binary += String.fromCharCode(...bytes.subarray(index, index + 0x8000)); + } + return btoa(binary) + .replaceAll("+", "-") + .replaceAll("/", "_") + .replace(/=+$/, ""); +} + +function decodeBase64Url(value: string): string { + const base64 = value.replaceAll("-", "+").replaceAll("_", "/"); + const padded = base64.padEnd(Math.ceil(base64.length / 4) * 4, "="); + const binary = atob(padded); + const bytes = Uint8Array.from(binary, (character) => character.charCodeAt(0)); + return new TextDecoder().decode(bytes); +} + function encodeCursor(payload: CursorPayload): string { - return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url"); + return encodeBase64Url(JSON.stringify(payload)); } function decodeCursor(cursor: string): CursorPayload | null { try { - const json = Buffer.from(cursor, "base64url").toString("utf8"); - const p = JSON.parse(json); - if (typeof p?.t !== "number" || typeof p?.k !== "string") return null; + const p = JSON.parse(decodeBase64Url(cursor)); + if ( + typeof p?.t !== "number" || + typeof p?.u !== "string" || + typeof p?.k !== "string" + ) { + return null; + } return p as CursorPayload; } catch { return null; @@ -865,44 +890,95 @@ export async function queryRecords( bindings.push(did); } - // Opaque keyset cursor encoding { t, v?, k }. Silently ignored if it doesn't - // match the current sort — callers shouldn't mix sort params with stale cursors. - const expectedKind = sortKind(sort); + const dialect = getDialect(db); + let ftsJoin = ""; + let ftsClause: ReturnType | null = null; + if (search) { + const collectionConfig = config.collections[collection]; + const fields = collectionConfig + ? getSearchableFields(collection, collectionConfig) + : null; + if (fields && fields.length > 0) { + ftsClause = ftsQueryClause(dialect, recordsTableName(collection)); + ftsJoin = ftsClause.join; + conditions.push(ftsClause.condition); + bindings.push(search); + } + } + + // Cursors are accepted only for the ordering that created them. URI is the + // final key so rows with identical sort values and timestamps cannot vanish. + const expectedKind = sort + ? sortKind(sort) + : ftsClause + ? "search" + : "time"; if (cursor) { const payload = decodeCursor(cursor); if (payload && payload.k === expectedKind) { + const stableTail = + "(r.time_us < ? OR (r.time_us = ? AND r.uri > ?))"; if (sort?.recordField) { - const sortExpr = getDialect(db).jsonExtract('r.record', sort.recordField); + const sortExpr = dialect.jsonExtract("r.record", sort.recordField); const cmp = sort.direction === "desc" ? "<" : ">"; - conditions.push(`(${sortExpr} ${cmp} ? OR (${sortExpr} = ? AND r.time_us < ?))`); - const v = payload.v ?? ""; - bindings.push(v as string | number, v as string | number, payload.t); + const value = payload.v ?? ""; + conditions.push( + `(${sortExpr} ${cmp} ? OR (${sortExpr} = ? AND ${stableTail}))`, + ); + bindings.push(value, value, payload.t, payload.t, payload.u); } else if (sort?.countType) { const sortCol = countColumnForType(config, collection, sort.countType); if (!sortCol) throw new Error(`Unknown countType: ${sort.countType}`); const cmp = sort.direction === "desc" ? "<" : ">"; - conditions.push(`(r.${sortCol} ${cmp} ? OR (r.${sortCol} = ? AND r.time_us < ?))`); - const v = Number(payload.v ?? 0); - bindings.push(v, v, payload.t); - } else { - conditions.push("r.time_us < ?"); - bindings.push(payload.t); + const value = Number(payload.v ?? 0); + conditions.push( + `(r.${sortCol} ${cmp} ? OR (r.${sortCol} = ? AND ${stableTail}))`, + ); + bindings.push(value, value, payload.t, payload.t, payload.u); + } else if (ftsClause && search && typeof payload.v === "number") { + const rankExpr = ftsClause.orderExpr; + const cmp = ftsClause.orderDirection === "desc" ? "<" : ">"; + conditions.push( + `(${rankExpr} ${cmp} ? OR (${rankExpr} = ? AND ${stableTail}))`, + ); + if (dialect.ftsStrategy === "generated-column") { + bindings.push( + search, + payload.v, + search, + payload.v, + payload.t, + payload.t, + payload.u, + ); + } else { + bindings.push( + payload.v, + payload.v, + payload.t, + payload.t, + payload.u, + ); + } + } else if (!ftsClause) { + conditions.push(stableTail); + bindings.push(payload.t, payload.t, payload.u); } } } for (const [field, value] of Object.entries(filters)) { - conditions.push(`${getDialect(db).jsonExtract('r.record', field)} = ?`); + conditions.push(`${dialect.jsonExtract("r.record", field)} = ?`); bindings.push(value); } for (const [field, range] of Object.entries(rangeFilters)) { if (range.min != null) { - conditions.push(`${getDialect(db).jsonExtract('r.record', field)} >= ?`); + conditions.push(`${dialect.jsonExtract("r.record", field)} >= ?`); bindings.push(range.min); } if (range.max != null) { - conditions.push(`${getDialect(db).jsonExtract('r.record', field)} <= ?`); + conditions.push(`${dialect.jsonExtract("r.record", field)} <= ?`); bindings.push(range.max); } } @@ -914,59 +990,53 @@ export async function queryRecords( bindings.push(minCount); } - let ftsJoin = ""; - let ftsClause: ReturnType | null = null; - if (search) { - const colConfig2 = config.collections[collection]; - const fields = colConfig2 ? getSearchableFields(collection, colConfig2) : null; - if (fields && fields.length > 0) { - ftsClause = ftsQueryClause(getDialect(db), recordsTableName(collection)); - ftsJoin = ftsClause.join; - conditions.push(ftsClause.condition); - // SECURITY: `search` is user input bound as a parameter, not interpolated. - bindings.push(search); - } - } - const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; const countSelect = countCols.length > 0 ? ", " + countCols.map(({ column }) => `r.${column}`).join(", ") : ""; - const select = `r.uri, r.did, r.rkey, r.cid, r.record, r.time_us, r.indexed_at${countSelect}`; + const selectBindings: (string | number)[] = []; + let searchRankSelect = ""; + if (expectedKind === "search" && ftsClause && search) { + searchRankSelect = `, ${ftsClause.orderExpr} AS __search_rank`; + if (dialect.ftsStrategy === "generated-column") { + selectBindings.push(search); + } + } + const select = `r.uri, r.did, r.rkey, r.cid, r.record, r.time_us, r.indexed_at${countSelect}${searchRankSelect}`; const join = [source?.joins, ftsJoin].filter(Boolean).join(" "); + const orderBindings: (string | number)[] = []; let orderBy: string; if (sort?.recordField) { const dir = sort.direction === "desc" ? "DESC" : "ASC"; - orderBy = `${getDialect(db).jsonExtract('r.record', sort.recordField)} ${dir}, r.time_us DESC`; + orderBy = `${dialect.jsonExtract("r.record", sort.recordField)} ${dir}, r.time_us DESC, r.uri ASC`; } else if (sort?.countType) { const dir = sort.direction === "desc" ? "DESC" : "ASC"; const sortCol = countColumnForType(config, collection, sort.countType); if (!sortCol) throw new Error(`Unknown countType: ${sort.countType}`); - orderBy = `r.${sortCol} ${dir}, r.time_us DESC`; + orderBy = `r.${sortCol} ${dir}, r.time_us DESC, r.uri ASC`; } else if (ftsClause) { - orderBy = `${ftsClause.orderExpr}, r.time_us DESC`; - // PG ts_rank needs the search term bound again for ORDER BY - if (getDialect(db).ftsStrategy === "generated-column" && search) { - bindings.push(search); + const dir = ftsClause.orderDirection.toUpperCase(); + orderBy = `${ftsClause.orderExpr} ${dir}, r.time_us DESC, r.uri ASC`; + if (dialect.ftsStrategy === "generated-column" && search) { + orderBindings.push(search); } } else { - orderBy = "r.time_us DESC"; + orderBy = "r.time_us DESC, r.uri ASC"; } - bindings.push(limit); - const query = `SELECT ${select} FROM ${table} r ${join} ${where} ORDER BY ${orderBy} LIMIT ?`; const result = await db .prepare(query) - .bind(...bindings) + .bind(...selectBindings, ...bindings, ...orderBindings, limit) .all(); + const rows = result.results ?? []; const nsid = nsidForShortName(config, collection) ?? collection; - const records = (result.results ?? []).map((row: any) => { + const records = rows.map((row: any) => { const rec: RecordRow & { counts?: Record } = { uri: row.uri, did: row.did, @@ -990,7 +1060,12 @@ export async function queryRecords( const nextCursor = records.length === limit - ? buildCursor(records[records.length - 1], sort, expectedKind) + ? buildCursor( + records[records.length - 1], + sort, + expectedKind, + rows[rows.length - 1]?.__search_rank, + ) : undefined; return { records, cursor: nextCursor }; @@ -1000,17 +1075,24 @@ export async function queryRecords( function buildCursor( row: RecordRow & { counts?: Record }, sort: SortOption | undefined, - kind: string + kind: string, + searchRank?: unknown, ): string { const t = Number(row.time_us); + const u = row.uri; if (sort?.recordField) { const parsed = row.record ? JSON.parse(row.record) : null; const v = parsed ? getNestedValue(parsed, sort.recordField) : undefined; - return encodeCursor({ t, v: v == null ? "" : String(v), k: kind }); + return encodeCursor({ t, u, v: v == null ? "" : String(v), k: kind }); } if (sort?.countType) { const v = row.counts?.[sort.countType] ?? 0; - return encodeCursor({ t, v, k: kind }); + return encodeCursor({ t, u, v, k: kind }); + } + if (kind === "search") { + const v = Number(searchRank); + if (!Number.isFinite(v)) throw new Error("Search rank missing from result"); + return encodeCursor({ t, u, v, k: kind }); } - return encodeCursor({ t, k: kind }); + return encodeCursor({ t, u, k: kind }); } diff --git a/packages/contrail/src/core/dialect.ts b/packages/contrail/src/core/dialect.ts index e3bb790..f5c3371 100644 --- a/packages/contrail/src/core/dialect.ts +++ b/packages/contrail/src/core/dialect.ts @@ -109,19 +109,26 @@ export function buildFtsSchema( export function ftsQueryClause( dialect: SqlDialect, recordsTable: string -): { join: string; condition: string; orderExpr: string } { +): { + join: string; + condition: string; + orderExpr: string; + orderDirection: "asc" | "desc"; +} { if (dialect.ftsStrategy === "virtual-table") { const ftsTable = recordsTable.replace("records_", "fts_"); return { join: `JOIN ${ftsTable} fts ON fts.uri = r.uri`, condition: "fts.content MATCH ?", orderExpr: "fts.rank", + orderDirection: "asc", }; } else { return { join: "", condition: "r.search_vector @@ plainto_tsquery('english', ?)", orderExpr: "ts_rank(r.search_vector, plainto_tsquery('english', ?))", + orderDirection: "desc", }; } } diff --git a/packages/contrail/src/core/router/notify.ts b/packages/contrail/src/core/router/notify.ts index e14e33a..72782f6 100644 --- a/packages/contrail/src/core/router/notify.ts +++ b/packages/contrail/src/core/router/notify.ts @@ -25,12 +25,24 @@ type RecordFetchResult = | { kind: "not-found" } | { kind: "error"; status?: number; message: string }; +function abortable(operation: Promise, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + operation.then(resolve, reject).finally(() => { + signal.removeEventListener("abort", onAbort); + }); + }); +} + /** Fetch one authoritative record without conflating failure with deletion. */ async function fetchRecordFromPDS( pds: string, did: string, collection: string, - rkey: string + rkey: string, + signal: AbortSignal, ): Promise { const url = new URL(`/xrpc/com.atproto.repo.getRecord`, pds); url.searchParams.set("repo", did); @@ -38,28 +50,18 @@ async function fetchRecordFromPDS( url.searchParams.set("rkey", rkey); let response: Response; - const controller = new AbortController(); - const timeout = setTimeout( - () => controller.abort(), - NOTIFY_FETCH_TIMEOUT_MS, - ); try { - response = await fetch(url.toString(), { signal: controller.signal }); + response = await fetch(url.toString(), { signal }); } catch (error) { - return { - kind: "error", - message: controller.signal.aborted - ? `PDS request timed out after ${NOTIFY_FETCH_TIMEOUT_MS}ms` - : `network failure: ${String(error)}`, - }; - } finally { - clearTimeout(timeout); + if (signal.aborted) throw error; + return { kind: "error", message: `network failure: ${String(error)}` }; } if (!response.ok) { - const body = await response - .json() - .catch(() => null) as { error?: unknown; message?: unknown } | null; + const body = await response.json().catch((error) => { + if (signal.aborted) throw error; + return null; + }) as { error?: unknown; message?: unknown } | null; const errorCode = typeof body?.error === "string" ? body.error : undefined; if (errorCode === "RecordNotFound") { return { kind: "not-found" }; @@ -78,6 +80,7 @@ async function fetchRecordFromPDS( try { data = await response.json(); } catch (error) { + if (signal.aborted) throw error; return { kind: "error", status: response.status, @@ -156,26 +159,43 @@ export async function processNotifyUris( ); for (const { uri, parsed } of validUris) { - let pds: string | null | undefined; + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(new Error("notify deadline exceeded")), + NOTIFY_FETCH_TIMEOUT_MS, + ); + let result: RecordFetchResult; try { - pds = await getPDS(parsed.did as Did, db, config); + const pds = await abortable( + getPDS(parsed.did as Did, db, config), + controller.signal, + ); + if (!pds) { + errors.push(`${uri}: could not resolve PDS for ${parsed.did}`); + continue; + } + result = await abortable( + fetchRecordFromPDS( + pds, + parsed.did, + parsed.collection, + parsed.rkey, + controller.signal, + ), + controller.signal, + ); } catch (error) { - errors.push(`${uri}: could not resolve PDS: ${String(error)}`); - continue; - } - if (!pds) { - errors.push(`${uri}: could not resolve PDS for ${parsed.did}`); + errors.push( + controller.signal.aborted + ? `${uri}: PDS request timed out after ${NOTIFY_FETCH_TIMEOUT_MS}ms` + : `${uri}: could not resolve or fetch from PDS: ${String(error)}`, + ); continue; + } finally { + clearTimeout(timeout); } - const result = await fetchRecordFromPDS( - pds, - parsed.did, - parsed.collection, - parsed.rkey - ); const existingInfo = existing.get(uri); - if (result.kind === "error") { errors.push(`${uri}: ${result.message}`); continue; diff --git a/packages/contrail/tests/built-sqlite.mjs b/packages/contrail/tests/built-sqlite.mjs new file mode 100644 index 0000000..ec7f4b3 --- /dev/null +++ b/packages/contrail/tests/built-sqlite.mjs @@ -0,0 +1,6 @@ +import assert from "node:assert/strict"; +import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite"; + +const db = createSqliteDatabase(":memory:"); +const row = await db.prepare("SELECT 1 AS ok").first(); +assert.equal(row?.ok, 1); diff --git a/packages/contrail/tests/notify.test.ts b/packages/contrail/tests/notify.test.ts index 85c6680..7cbc5f4 100644 --- a/packages/contrail/tests/notify.test.ts +++ b/packages/contrail/tests/notify.test.ts @@ -302,6 +302,45 @@ describe("POST notifyOfUpdate", () => { } }); + it("times out when response headers arrive but the body stalls", async () => { + vi.useFakeTimers(); + try { + const did = "did:plc:test"; + const uri = `at://${did}/community.lexicon.calendar.event/stalled-body`; + await seedLocalRecord(uri); + await seedIdentity(did, "https://pds.example.com"); + vi.stubGlobal( + "fetch", + vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + let bodyController!: ReadableStreamDefaultController; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + }); + init?.signal?.addEventListener( + "abort", + () => bodyController.error(new DOMException("aborted", "AbortError")), + { once: true }, + ); + return new Response(body, { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }), + ); + + const response = notify(uri); + await vi.advanceTimersByTimeAsync(5_000); + const body = await (await response).json(); + expect(body).toMatchObject({ indexed: 0, deleted: 0 }); + expect(body.errors?.[0]).toContain("timed out"); + expect(await getLocalRecord(uri)).toBeDefined(); + } finally { + vi.useRealTimers(); + } + }); + it("preserves local state after a malformed successful response", async () => { const did = "did:plc:test"; const uri = `at://${did}/community.lexicon.calendar.event/malformed`; diff --git a/packages/contrail/tests/query-pagination.test.ts b/packages/contrail/tests/query-pagination.test.ts new file mode 100644 index 0000000..b59bae8 --- /dev/null +++ b/packages/contrail/tests/query-pagination.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, it } from "vitest"; +import { + postgresDialect, + queryRecords, + resolveConfig, + sqliteDialect, + type Database, + type Statement, +} from "../src/index"; + +const config = resolveConfig({ + namespace: "com.example", + profiles: [], + collections: { + event: { + collection: "com.example.event", + searchable: ["name"], + }, + }, +}); + +function createQueryRecorder( + pages: Record[][], + dialect = sqliteDialect, +) { + const calls: { sql: string; bindings: unknown[] }[] = []; + const db: Database = { + dialect, + prepare(sql: string): Statement { + let bindings: unknown[] = []; + const statement: Statement = { + bind(...values: unknown[]) { + bindings = values; + return statement; + }, + async run() { + throw new Error("unexpected run"); + }, + async all() { + calls.push({ sql, bindings }); + return { results: (pages.shift() ?? []) as T[] }; + }, + async first() { + throw new Error("unexpected first"); + }, + }; + return statement; + }, + async batch() { + throw new Error("unexpected batch"); + }, + }; + return { db, calls }; +} + +describe("query keyset pagination", () => { + it("carries search rank, time, and URI into the next-page predicate", async () => { + const row = { + uri: "at://did:plc:a/com.example.event/a", + did: "did:plc:a", + rkey: "a", + cid: "cid-a", + record: JSON.stringify({ name: "Rust" }), + time_us: 1000, + indexed_at: 1000, + __search_rank: -2.5, + }; + const { db, calls } = createQueryRecorder([[row], []]); + + const first = await queryRecords(db, config, { + collection: "event", + search: "Rust", + limit: 1, + }); + await queryRecords(db, config, { + collection: "event", + search: "Rust", + limit: 1, + cursor: first.cursor, + }); + + expect(calls[1].sql).toContain( + "fts.rank > ? OR (fts.rank = ? AND (r.time_us < ? OR (r.time_us = ? AND r.uri > ?)))", + ); + expect(calls[1].sql).toContain( + "ORDER BY fts.rank ASC, r.time_us DESC, r.uri ASC", + ); + expect(calls[1].bindings).toEqual([ + "Rust", + -2.5, + -2.5, + 1000, + 1000, + row.uri, + 1, + ]); + }); + + it("binds PostgreSQL rank expressions in SQL placeholder order", async () => { + const row = { + uri: "at://did:plc:a/com.example.event/a", + did: "did:plc:a", + rkey: "a", + cid: "cid-a", + record: JSON.stringify({ name: "Rust" }), + time_us: 1000, + indexed_at: 1000, + __search_rank: 2.5, + }; + const { db, calls } = createQueryRecorder([[row], []], postgresDialect); + + const first = await queryRecords(db, config, { + collection: "event", + search: "Rust", + limit: 1, + }); + await queryRecords(db, config, { + collection: "event", + search: "Rust", + limit: 1, + cursor: first.cursor, + }); + + expect(calls[1].sql).toContain( + "ORDER BY ts_rank(r.search_vector, plainto_tsquery('english', ?)) DESC, r.time_us DESC, r.uri ASC", + ); + expect(calls[1].bindings).toEqual([ + "Rust", + "Rust", + "Rust", + 2.5, + "Rust", + 2.5, + 1000, + 1000, + row.uri, + "Rust", + 1, + ]); + }); +}); diff --git a/packages/contrail/tests/records.test.ts b/packages/contrail/tests/records.test.ts index b7d846d..9ecca87 100644 --- a/packages/contrail/tests/records.test.ts +++ b/packages/contrail/tests/records.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, beforeEach } from "vitest"; +import { describe, it, expect, beforeEach, vi } from "vitest"; import type { Database } from "../src/index"; import { ingestRecords, createTestDbWithSchema, makeEvent, TEST_CONFIG } from "./helpers"; import { queryRecords, getLastCursor, saveCursor } from "../src/index"; @@ -268,6 +268,55 @@ describe("queryRecords", () => { expect(page2.cursor).toBeUndefined(); }); + it("paginates records with tied timestamps without dropping any", async () => { + const tied = ["a", "b", "c"].map((rkey) => + makeEvent({ + uri: `at://did:plc:tied/community.lexicon.calendar.event/${rkey}`, + did: "did:plc:tied", + rkey, + record: { name: "Same" }, + time_us: 1000, + }), + ); + await ingestRecords(db, tied, TEST_CONFIG); + + const page1 = await queryRecords(db, TEST_CONFIG, { + collection: "community.lexicon.calendar.event", + did: "did:plc:tied", + limit: 2, + }); + const page2 = await queryRecords(db, TEST_CONFIG, { + collection: "community.lexicon.calendar.event", + did: "did:plc:tied", + limit: 2, + cursor: page1.cursor, + }); + + expect([...page1.records, ...page2.records].map((record) => record.rkey)).toEqual([ + "a", + "b", + "c", + ]); + }); + + it("encodes and decodes cursors without Node Buffer", async () => { + vi.stubGlobal("Buffer", undefined); + try { + const page1 = await queryRecords(db, TEST_CONFIG, { + collection: "community.lexicon.calendar.event", + limit: 2, + }); + const page2 = await queryRecords(db, TEST_CONFIG, { + collection: "community.lexicon.calendar.event", + limit: 2, + cursor: page1.cursor, + }); + expect(page2.records).toHaveLength(1); + } finally { + vi.unstubAllGlobals(); + } + }); + it("filters by equality", async () => { const result = await queryRecords(db, TEST_CONFIG, { collection: "community.lexicon.calendar.event", diff --git a/packages/contrail/tests/search.test.ts b/packages/contrail/tests/search.test.ts index f744d0c..3a0ed69 100644 --- a/packages/contrail/tests/search.test.ts +++ b/packages/contrail/tests/search.test.ts @@ -251,4 +251,63 @@ describe.skipIf(!hasFts)("search pagination", () => { expect(page2.records).toHaveLength(2); expect(page2.cursor).toBeUndefined(); }); + + it("uses rank, time, and URI together when paginating", async () => { + await ingestRecords( + db, + [ + makeEvent({ + uri: `at://did:plc:a/${collection}/rank-a`, + collection, + rkey: "rank-a", + record: { name: "Quokka", mode: "", description: "" }, + time_us: 1000, + }), + makeEvent({ + uri: `at://did:plc:a/${collection}/rank-b`, + collection, + rkey: "rank-b", + record: { + name: "Quokka community gathering with a deliberately long title", + mode: "online", + description: "many unrelated words make this a weaker match", + }, + time_us: 3000, + }), + makeEvent({ + uri: `at://did:plc:a/${collection}/rank-c`, + collection, + rkey: "rank-c", + record: { + name: "Quokka event with another deliberately long title", + mode: "online", + description: "more unrelated words make this weaker too", + }, + time_us: 2000, + }), + ], + SEARCH_CONFIG, + ); + + const complete = await queryRecords(db, SEARCH_CONFIG, { + collection, + search: "Quokka", + limit: 50, + }); + const paged: string[] = []; + let cursor: string | undefined; + do { + const page = await queryRecords(db, SEARCH_CONFIG, { + collection, + search: "Quokka", + limit: 1, + cursor, + }); + paged.push(...page.records.map((record) => record.uri)); + cursor = page.cursor; + } while (cursor && paged.length < 10); + + expect(paged).toEqual(complete.records.map((record) => record.uri)); + expect(paged).toHaveLength(3); + }); }); diff --git a/packages/contrail/tsup.config.ts b/packages/contrail/tsup.config.ts index 7cd50c0..e127492 100644 --- a/packages/contrail/tsup.config.ts +++ b/packages/contrail/tsup.config.ts @@ -17,4 +17,5 @@ export default defineConfig({ clean: true, tsconfig: "tsconfig.build.json", external: ["node:sqlite", "pg", "wrangler"], + removeNodeProtocol: false, }); -- 2.51.2