From ccb970f2003dc79fc5d14d5e43422be8d755f00d Mon Sep 17 00:00:00 2001 From: Codinget Date: Wed, 22 Jul 2026 22:10:29 +0000 Subject: [PATCH 1/4] feat: add export/import streaming to the pluggable backends Add backend-agnostic data export/import over an NDJSON wire format, plus an inspect utility, exposed via three new CLIs and an HTTP export endpoint. - ExportImport types + filter helpers (kind/record scoping, hard-intersection of filters) in src/db/{types/ExportImport,export/filter}.ts - SqliteInterface implements Exportable + Importable: signal-checked async generator export (one query per table, per-record yield, trailing error sentinel on mid-stream failure) and a manually-driven import transaction that rolls back on any error/abort and never commits partial data - ApiInterface implements Exportable via its own fetch({signal}) - computeForcedExportFilter enforces non-global-admin scope (resided-in abodes + co-resident users, intersected with apikey restrict_*); GET /export intersects it with the caller's filter and wires an AbortController to the response socket - inspectExportStream reports kinds/counts from any stream without a db - abode-export / abode-import / abode-inspect CLIs (import is sqlite-only) - Secrets are not exported: imported users default to '#unset' passwords and apikeys are re-minted a token (ClientApikey view round-trips exactly) - test/tools/export-import.test.ts: round-trip, filter narrowing, forced-scope, export/import cancellation, in-process Koa endpoint, inspect Co-Authored-By: Claude Opus 4.8 --- package-lock.json | 3 + package.json | 8 +- src/bin/abode-export.ts | 97 +++++ src/bin/abode-import.ts | 106 +++++ src/bin/abode-inspect.ts | 68 +++ src/db/api/ApiInterface.ts | 46 +- src/db/export/filter.ts | 79 ++++ src/db/export/inspect.ts | 60 +++ src/db/sqlite/SqliteInterface.ts | 231 +++++++++- src/db/sqlite/query.ts | 4 +- src/db/types/ExportImport.ts | 68 +++ src/webapi/apirouter.ts | 48 +++ src/webapi/exportScope.ts | 58 +++ test/tools/export-import.test.ts | 715 +++++++++++++++++++++++++++++++ 14 files changed, 1586 insertions(+), 5 deletions(-) create mode 100644 src/bin/abode-export.ts create mode 100644 src/bin/abode-import.ts create mode 100644 src/bin/abode-inspect.ts create mode 100644 src/db/export/filter.ts create mode 100644 src/db/export/inspect.ts create mode 100644 src/db/types/ExportImport.ts create mode 100644 src/webapi/exportScope.ts create mode 100644 test/tools/export-import.test.ts diff --git a/package-lock.json b/package-lock.json index afe0d6a..579cf8b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -25,6 +25,9 @@ "react-redux": "^9.2.0" }, "bin": { + "abode-export": "dist/bin/abode-export.cjs", + "abode-import": "dist/bin/abode-import.cjs", + "abode-inspect": "dist/bin/abode-inspect.cjs", "abode-migrate": "dist/bin/abode-migrate.cjs", "abode-repl": "dist/bin/abode-repl.cjs", "abode-tui": "dist/bin/abode-tui.cjs", diff --git a/package.json b/package.json index 9c234a9..bc19a34 100644 --- a/package.json +++ b/package.json @@ -9,7 +9,10 @@ "abode-migrate": "dist/bin/abode-migrate.cjs", "abode-repl": "dist/bin/abode-repl.cjs", "abode-web": "dist/bin/abode-web.cjs", - "abode-tui": "dist/bin/abode-tui.cjs" + "abode-tui": "dist/bin/abode-tui.cjs", + "abode-export": "dist/bin/abode-export.cjs", + "abode-import": "dist/bin/abode-import.cjs", + "abode-inspect": "dist/bin/abode-inspect.cjs" }, "scripts": { "repl": "tsx --import ./src/meta/dev/register.ts", @@ -18,6 +21,9 @@ "abode-web": "tsx --import ./src/meta/dev/register.ts --import ./src/meta/dev/webhot.ts src/bin/abode-web.ts", "abode-tui": "tsx --import ./src/meta/dev/register.ts --import ./src/meta/dev/silenthot.ts src/bin/abode-tui.ts", "abode-sources": "tsx --import ./src/meta/dev/register.ts src/bin/abode-sources.ts", + "abode-export": "tsx --import ./src/meta/dev/register.ts src/bin/abode-export.ts", + "abode-import": "tsx --import ./src/meta/dev/register.ts src/bin/abode-import.ts", + "abode-inspect": "tsx --import ./src/meta/dev/register.ts src/bin/abode-inspect.ts", "build": "NODE_ENV=production npm run build:impl", "build:impl": "rm -rf dist && tsx node_modules/.bin/webpack && chmod +x dist/bin/* && chmod -x dist/bin/*.*", "test": "node --import tsx/esm --import ./src/meta/dev/register.ts --test $(find test -name '*.test.ts' | sort)", diff --git a/src/bin/abode-export.ts b/src/bin/abode-export.ts new file mode 100644 index 0000000..ba38e4d --- /dev/null +++ b/src/bin/abode-export.ts @@ -0,0 +1,97 @@ +import { createWriteStream } from "node:fs"; +import { pipeline } from "node:stream/promises"; +import { getDbInterface } from "../db/index.js"; +import { isExportable, type ExportFilter } from "../db/types/ExportImport.js"; +import { isExportKind } from "../db/export/filter.js"; + +const args = process.argv.slice(2); + +function printUsage(err: boolean | string = false): never { + const log = (err ? console.error : console.log).bind(console); + if (typeof err === "string") { + log(`Error: ${err}`); + log(""); + } + log("Usage:"); + log("\tabode-export --help"); + log( + "\tabode-export [--kinds=user,abode,...] [--exclude-kinds=...] \\", + ); + log("\t [--abodes=aid,...] [--users=uid,...] [--out=file|-]"); + process.exit(err ? 1 : 0); +} + +if (["-h", "--help", "help"].some((x) => args.includes(x))) printUsage(); + +let url: string | undefined; +const flags = new Map(); +for (const arg of args) { + if (arg.startsWith("--")) { + const eq = arg.indexOf("="); + if (eq === -1) printUsage(`invalid flag: ${arg}`); + flags.set(arg.slice(2, eq), arg.slice(eq + 1)); + } else if (url === undefined) { + url = arg; + } else { + printUsage("too many arguments"); + } +} +if (!url) printUsage("missing "); + +const knownFlags = ["kinds", "exclude-kinds", "abodes", "users", "out"]; +for (const key of flags.keys()) { + if (!knownFlags.includes(key)) printUsage(`unknown flag: --${key}`); +} + +function parseList(value: string | undefined): string[] | undefined { + if (value === undefined) return undefined; + return value.split(",").filter(Boolean); +} +function parseKinds(value: string | undefined) { + const list = parseList(value); + if (!list) return undefined; + const bad = list.filter((k) => !isExportKind(k)); + if (bad.length) printUsage(`invalid kind(s): ${bad.join(", ")}`); + return list.filter(isExportKind); +} + +const filter: ExportFilter = {}; +const kinds = parseKinds(flags.get("kinds")); +if (kinds) filter.kinds = kinds; +const excludeKinds = parseKinds(flags.get("exclude-kinds")); +if (excludeKinds) filter.excludeKinds = excludeKinds; +const abodes = parseList(flags.get("abodes")); +if (abodes) filter.abodes = abodes; +const users = parseList(flags.get("users")); +if (users) filter.users = users; + +const db = await getDbInterface(url); +if (!isExportable(db)) { + console.error(`Error: backend '${db.name}' does not support export`); + process.exit(1); +} + +const out = flags.get("out") ?? "-"; +const dest = + out === "-" ? process.stdout : createWriteStream(out, { encoding: "utf8" }); + +const ac = new AbortController(); +const onSignal = () => ac.abort(); +process.on("SIGINT", onSignal); +process.on("SIGTERM", onSignal); + +try { + await pipeline(db.export({ filter, signal: ac.signal }), dest); +} catch (e) { + if (ac.signal.aborted) { + console.error("Export aborted"); + process.exit(130); + } + throw e; +} finally { + process.off("SIGINT", onSignal); + process.off("SIGTERM", onSignal); + await db.close().catch(() => {}); +} + +process.exit(0); diff --git a/src/bin/abode-import.ts b/src/bin/abode-import.ts new file mode 100644 index 0000000..d8e0c2e --- /dev/null +++ b/src/bin/abode-import.ts @@ -0,0 +1,106 @@ +import { createReadStream } from "node:fs"; +import { getWrappedDb } from "../db/sqlite/impl/index.js"; +import { SqliteInterface } from "../db/sqlite/SqliteInterface.js"; +import { parseSqliteUrl } from "../db/sqlite/url.js"; +import { isExportKind } from "../db/export/filter.js"; +import type { ExportFilter } from "../db/types/ExportImport.js"; + +const args = process.argv.slice(2); + +function printUsage(err: boolean | string = false): never { + const log = (err ? console.error : console.log).bind(console); + if (typeof err === "string") { + log(`Error: ${err}`); + log(""); + } + log("Usage:"); + log("\tabode-import --help"); + log("\tabode-import [--kinds=...] \\"); + log( + "\t [--exclude-kinds=...] [--abodes=aid,...] [--users=uid,...]", + ); + log(""); + log( + "The target database must already be migrated (run abode-migrate first).", + ); + process.exit(err ? 1 : 0); +} + +if (["-h", "--help", "help"].some((x) => args.includes(x))) printUsage(); + +const positional: string[] = []; +const flags = new Map(); +for (const arg of args) { + if (arg.startsWith("--")) { + const eq = arg.indexOf("="); + if (eq === -1) printUsage(`invalid flag: ${arg}`); + flags.set(arg.slice(2, eq), arg.slice(eq + 1)); + } else { + positional.push(arg); + } +} + +const url = positional[0]; +const input = positional[1]; +if (!url) printUsage("missing "); +if (!input) printUsage("missing "); +if (positional.length > 2) printUsage("too many arguments"); + +const knownFlags = ["kinds", "exclude-kinds", "abodes", "users"]; +for (const key of flags.keys()) { + if (!knownFlags.includes(key)) printUsage(`unknown flag: --${key}`); +} + +function parseList(value: string | undefined): string[] | undefined { + if (value === undefined) return undefined; + return value.split(",").filter(Boolean); +} +function parseKinds(value: string | undefined) { + const list = parseList(value); + if (!list) return undefined; + const bad = list.filter((k) => !isExportKind(k)); + if (bad.length) printUsage(`invalid kind(s): ${bad.join(", ")}`); + return list.filter(isExportKind); +} + +const filter: ExportFilter = {}; +const kinds = parseKinds(flags.get("kinds")); +if (kinds) filter.kinds = kinds; +const excludeKinds = parseKinds(flags.get("exclude-kinds")); +if (excludeKinds) filter.excludeKinds = excludeKinds; +const abodes = parseList(flags.get("abodes")); +if (abodes) filter.abodes = abodes; +const users = parseList(flags.get("users")); +if (users) filter.users = users; + +// Import is sqlite-only: construct the backend directly rather than resolving +// generically, so it can never be pointed at a remote (api) target. +const db = new SqliteInterface(getWrappedDb(...parseSqliteUrl(url))); + +const source = input === "-" ? process.stdin : createReadStream(input); + +const ac = new AbortController(); +const onSignal = () => ac.abort(); +process.on("SIGINT", onSignal); +process.on("SIGTERM", onSignal); + +try { + const result = await db.import(source, { filter, signal: ac.signal }); + const total = Object.values(result.counts).reduce((a, b) => a + b, 0); + console.log(`Imported ${total} record(s):`); + for (const [kind, count] of Object.entries(result.counts)) { + console.log(`- ${kind}: ${count}`); + } +} catch (e) { + if (ac.signal.aborted) { + console.error("Import aborted; no changes committed"); + process.exit(130); + } + throw e; +} finally { + process.off("SIGINT", onSignal); + process.off("SIGTERM", onSignal); + await db.close().catch(() => {}); +} + +process.exit(0); diff --git a/src/bin/abode-inspect.ts b/src/bin/abode-inspect.ts new file mode 100644 index 0000000..cdd5536 --- /dev/null +++ b/src/bin/abode-inspect.ts @@ -0,0 +1,68 @@ +import { createReadStream } from "node:fs"; +import { inspectExportStream } from "../db/export/inspect.js"; +import { isExportKind } from "../db/export/filter.js"; +import type { ExportKind } from "../db/types/ExportImport.js"; + +const args = process.argv.slice(2); + +function printUsage(err: boolean | string = false): never { + const log = (err ? console.error : console.log).bind(console); + if (typeof err === "string") { + log(`Error: ${err}`); + log(""); + } + log("Usage:"); + log("\tabode-inspect --help"); + log("\tabode-inspect [--stop-after=note,...]"); + process.exit(err ? 1 : 0); +} + +if (["-h", "--help", "help"].some((x) => args.includes(x))) printUsage(); + +let input: string | undefined; +const flags = new Map(); +for (const arg of args) { + if (arg.startsWith("--")) { + const eq = arg.indexOf("="); + if (eq === -1) printUsage(`invalid flag: ${arg}`); + flags.set(arg.slice(2, eq), arg.slice(eq + 1)); + } else if (input === undefined) { + input = arg; + } else { + printUsage("too many arguments"); + } +} +if (!input) printUsage("missing "); + +for (const key of flags.keys()) { + if (key !== "stop-after") printUsage(`unknown flag: --${key}`); +} + +let stopAfterKinds: ExportKind[] | undefined; +const rawStop = flags.get("stop-after"); +if (rawStop !== undefined) { + const list = rawStop.split(",").filter(Boolean); + const bad = list.filter((k) => !isExportKind(k)); + if (bad.length) printUsage(`invalid kind(s): ${bad.join(", ")}`); + stopAfterKinds = list.filter(isExportKind); +} + +const source = input === "-" ? process.stdin : createReadStream(input); + +const { counts, meta } = await inspectExportStream(source, { stopAfterKinds }); + +if (meta) { + console.log("Meta:"); + console.log(`- format version: ${meta.v ?? "?"}`); + console.log(`- exported at: ${meta.exportedAt ?? "?"}`); + console.log(`- source backend: ${meta.source ?? "?"}`); + if (meta.filter && Object.keys(meta.filter).length) { + console.log(`- effective filter: ${JSON.stringify(meta.filter)}`); + } +} +console.log("Counts:"); +const entries = Object.entries(counts); +if (!entries.length) console.log("(none)"); +for (const [kind, count] of entries) console.log(`- ${kind}: ${count}`); + +process.exit(0); diff --git a/src/db/api/ApiInterface.ts b/src/db/api/ApiInterface.ts index f376280..3886ddc 100644 --- a/src/db/api/ApiInterface.ts +++ b/src/db/api/ApiInterface.ts @@ -25,8 +25,11 @@ import type { CreateUser, UpdateUser, } from "../types/User.js"; +import { Readable } from "node:stream"; +import type { ReadableStream as WebReadableStream } from "node:stream/web"; +import type { Exportable, ExportOptions } from "../types/ExportImport.js"; -export class ApiInterface implements DbInterface { +export class ApiInterface implements DbInterface, Exportable { #root: string; #headers: Record; #readonly: boolean; @@ -296,4 +299,45 @@ export class ApiInterface implements DbInterface { async listNotesByUserId(uid: string): Promise { return this.#call("GET", "/users/:uid/notes", { params: { uid } }); } + + export(options: ExportOptions = {}): NodeJS.ReadableStream { + // Streaming NDJSON bypasses the JSON-only `#call` helper: the raw + // `fetch` gets the caller's `signal` directly, so aborting cancels the + // underlying HTTP request itself. + const { filter, signal } = options; + const sp = new URLSearchParams(); + if (filter?.kinds) sp.set("kinds", filter.kinds.join(",")); + if (filter?.excludeKinds) + sp.set("excludeKinds", filter.excludeKinds.join(",")); + if (filter?.abodes) sp.set("abodes", filter.abodes.join(",")); + if (filter?.users) sp.set("users", filter.users.join(",")); + const query = sp.toString(); + const url = this.#root + "/export" + (query ? `?${query}` : ""); + const headers = { ...this.#headers }; + + async function* generate(): AsyncGenerator { + const res = await fetch(url, { method: "GET", headers, signal }); + if (!res.ok) { + const text = await res.text(); + switch (res.status) { + case 400: + throw new InvalidAbodeError(); + case 401: + throw new NotAuthorizedAbodeError(); + case 403: + throw new ReadonlyAbodeError(); + case 404: + throw new NotFoundAbodeError(); + case 409: + throw new ConflictAbodeError(); + default: + throw new Error(`${res.status} ${res.statusText} ${text}`); + } + } + if (!res.body) return; + yield* Readable.fromWeb(res.body as WebReadableStream); + } + + return Readable.from(generate()); + } } diff --git a/src/db/export/filter.ts b/src/db/export/filter.ts new file mode 100644 index 0000000..9f78890 --- /dev/null +++ b/src/db/export/filter.ts @@ -0,0 +1,79 @@ +import type { ExportFilter, ExportKind } from "../types/ExportImport.js"; + +const EXPORT_KINDS = new Set([ + "user", + "abode", + "resident", + "apikey", + "note", +]); + +export function isExportKind(x: unknown): x is ExportKind { + return typeof x === "string" && EXPORT_KINDS.has(x as ExportKind); +} + +/** Whether a `kind` survives a filter's `kinds`/`excludeKinds` rules. */ +export function kindAllowed( + filter: ExportFilter | undefined, + kind: ExportKind, +): boolean { + if (!filter) return true; + if (filter.kinds && !filter.kinds.includes(kind)) return false; + if (filter.excludeKinds && filter.excludeKinds.includes(kind)) return false; + return true; +} + +/** + * Whether an individual record passes a filter's `abodes`/`users` allowlists. + * `abodes` scopes abode/resident/note (by aid); `users` scopes user/apikey + * (by uid). An absent allowlist means "unrestricted". + */ +export function recordAllowed( + filter: ExportFilter | undefined, + kind: ExportKind, + record: { uid?: string; aid?: string }, +): boolean { + if (!filter) return true; + switch (kind) { + case "user": + case "apikey": + return !filter.users || filter.users.includes(record.uid as string); + case "abode": + case "resident": + case "note": + return !filter.abodes || filter.abodes.includes(record.aid as string); + } +} + +function intersectList(a?: T[], b?: T[]): T[] | undefined { + if (!a) return b; + if (!b) return a; + const bs = new Set(b); + return a.filter((x) => bs.has(x)); +} + +function unionList(a?: T[], b?: T[]): T[] | undefined { + if (!a) return b; + if (!b) return a; + return [...new Set([...a, ...b])]; +} + +/** + * Combine two filters as a hard intersection: the result can never permit more + * than either input. An absent allowlist is treated as "unrestricted", so + * intersecting it with a present one yields the present one. `excludeKinds` + * are unioned (either exclusion still excludes). + */ +export function intersectExportFilters( + a: ExportFilter | null | undefined, + b: ExportFilter | null | undefined, +): ExportFilter { + if (!a) return b ?? {}; + if (!b) return a; + return { + kinds: intersectList(a.kinds, b.kinds), + excludeKinds: unionList(a.excludeKinds, b.excludeKinds), + abodes: intersectList(a.abodes, b.abodes), + users: intersectList(a.users, b.users), + }; +} diff --git a/src/db/export/inspect.ts b/src/db/export/inspect.ts new file mode 100644 index 0000000..becab57 --- /dev/null +++ b/src/db/export/inspect.ts @@ -0,0 +1,60 @@ +import readline from "node:readline"; +import type { + ExportFilter, + ExportKind, + ExportMeta, +} from "../types/ExportImport.js"; +import { isExportKind } from "./filter.js"; + +export type InspectResult = { + counts: Partial>; + meta?: Partial & { filter?: ExportFilter }; +}; + +/** + * Tally the kinds/counts contained in an NDJSON export stream without ever + * touching a database. Works on any stream — a file, {@link ApiInterface}'s + * export, or a pipe straight from {@link SqliteInterface}'s export. + * + * If `stopAfterKinds` is given, reading stops as soon as at least one record of + * every requested kind has been seen, rather than draining to EOF — useful for + * probing large dumps ("does this contain notes at all?"). + */ +export async function inspectExportStream( + source: NodeJS.ReadableStream, + options: { signal?: AbortSignal; stopAfterKinds?: ExportKind[] } = {}, +): Promise { + const { signal, stopAfterKinds } = options; + const counts: Partial> = {}; + let meta: InspectResult["meta"]; + + const rl = readline.createInterface({ + input: source, + crlfDelay: Infinity, + signal, + }); + try { + for await (const raw of rl) { + signal?.throwIfAborted(); + const line = raw.trim(); + if (!line) continue; + let parsed: { kind?: unknown; data?: unknown }; + try { + parsed = JSON.parse(line); + } catch { + continue; + } + if (parsed.kind === "meta") { + meta = parsed.data as InspectResult["meta"]; + continue; + } + if (!isExportKind(parsed.kind)) continue; + counts[parsed.kind] = (counts[parsed.kind] ?? 0) + 1; + if (stopAfterKinds && stopAfterKinds.every((k) => counts[k])) break; + } + } finally { + rl.close(); + } + + return { counts, meta }; +} diff --git a/src/db/sqlite/SqliteInterface.ts b/src/db/sqlite/SqliteInterface.ts index e5a0102..03c96fa 100644 --- a/src/db/sqlite/SqliteInterface.ts +++ b/src/db/sqlite/SqliteInterface.ts @@ -34,6 +34,7 @@ import { selectClientUser, selectClientUsers, selectNote, + selectNotes, selectPartialNotes, selectResident, selectResidents, @@ -45,8 +46,21 @@ import type { UpdateNote, } from "../types/Note.js"; import type { WrappedDb } from "./impl/types.js"; +import { Readable } from "node:stream"; +import readline from "node:readline"; +import type { + Exportable, + ExportKind, + ExportOptions, + Importable, + ImportOptions, + ImportResult, +} from "../types/ExportImport.js"; +import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js"; -export class SqliteInterface implements BackendDbInterface { +export class SqliteInterface + implements BackendDbInterface, Exportable, Importable +{ #db: WrappedDb; constructor(db: WrappedDb) { @@ -534,4 +548,219 @@ export class SqliteInterface implements BackendDbInterface { async listNotesByUserId(uid: string): Promise { return selectPartialNotes(this.#db, sql`n."created_by" = ${{ uuid: uid }}`); } + + export(options: ExportOptions = {}): NodeJS.ReadableStream { + const { filter, signal } = options; + const db = this.#db; + const source = this.name; + + // Per-table full materialization + JS-side per-record yielding (each + // `load()` is exactly one `WrappedDb.all()`). Kept lazy so the first query + // only fires once the destination starts pulling, and skipped entirely + // once the signal is aborted — no further reads after the destination + // goes away. + const tables: [ExportKind, () => { uid?: string; aid?: string }[]][] = [ + ["user", () => selectClientUsers(db)], + ["abode", () => selectAbodes(db)], + ["resident", () => selectResidents(db)], + ["apikey", () => selectClientApikeys(db)], + ["note", () => selectNotes(db)], + ]; + + async function* generate(): AsyncGenerator { + if (signal?.aborted) return; + yield JSON.stringify({ + kind: "meta", + data: { + v: 1, + exportedAt: new Date().toISOString(), + source, + filter: filter ?? {}, + }, + }) + "\n"; + try { + for (const [kind, load] of tables) { + if (signal?.aborted) return; + if (!kindAllowed(filter, kind)) continue; + for (const row of load()) { + if (signal?.aborted) return; + if (recordAllowed(filter, kind, row)) { + yield JSON.stringify({ kind, data: row }) + "\n"; + } + } + } + } catch (e) { + // Aborts unwind via early `return`, never here; a genuine mid-stream + // failure is surfaced as a trailing sentinel line (HTTP 200 headers + // are already flushed, so `convertError` can no longer apply). + if (signal?.aborted) return; + yield JSON.stringify({ + kind: "error", + data: { + message: e instanceof Error ? e.message : String(e), + code: e instanceof Error ? e.name : undefined, + }, + }) + "\n"; + } + } + + return Readable.from(generate()); + } + + async import( + source: NodeJS.ReadableStream, + options: ImportOptions = {}, + ): Promise { + this.#checkReadonly(); + const { filter, signal } = options; + const db = this.#db; + const counts: Partial> = {}; + + const rl = readline.createInterface({ + input: source, + crlfDelay: Infinity, + signal, + }); + + // Bulk restore trusts the export's referential integrity, and a filtered + // dump may legitimately reference `created_by`/`updated_by` users outside + // its scope. Suppress FK enforcement for the duration (can only be toggled + // outside a transaction) and restore it in `finally`. + db.run(sql`PRAGMA foreign_keys = OFF`); + db.run(sql`BEGIN`); + try { + for await (const raw of rl) { + signal?.throwIfAborted(); + const line = raw.trim(); + if (!line) continue; + let parsed: { kind?: unknown; data?: unknown }; + try { + parsed = JSON.parse(line); + } catch { + throw new InvalidAbodeError(); + } + if (parsed.kind === "meta") continue; + if (parsed.kind === "error") { + throw new Error( + `export stream reported an error: ${ + (parsed.data as { message?: string })?.message ?? "unknown" + }`, + ); + } + if (!isExportKind(parsed.kind)) continue; + if (!kindAllowed(filter, parsed.kind)) continue; + const data = parsed.data as { uid?: string; aid?: string }; + if (!recordAllowed(filter, parsed.kind, data)) continue; + this.#importRecord(parsed.kind, parsed.data); + counts[parsed.kind] = (counts[parsed.kind] ?? 0) + 1; + } + // An abort while blocked on the source closes readline without throwing, + // so re-check before committing to guarantee no partial commit. + signal?.throwIfAborted(); + db.run(sql`COMMIT`); + } catch (e) { + try { + db.run(sql`ROLLBACK`); + } catch { + /* already rolled back */ + } + throw e; + } finally { + rl.close(); + db.run(sql`PRAGMA foreign_keys = ON`); + } + + return { counts }; + } + + #importRecord(kind: ExportKind, data: unknown): void { + const db = this.#db; + switch (kind) { + case "user": { + const u = data as ClientUser; + // `password` is never exported; imported users land on the schema + // default ('#unset') and must reset before they can log in. + db.run(sql` + INSERT INTO "users"("uid", "email", "name", "flags", "created_at", "updated_at") + VALUES( + ${{ uuid: u.uid }}, + ${{ text: u.email }}, + ${{ text: u.name }}, + ${{ jsonb: u.flags }}, + ${{ date: u.created_at }}, + ${{ date: u.updated_at }} + ) + `); + break; + } + case "abode": { + const a = data as Abode; + db.run(sql` + INSERT INTO "abodes"("aid", "name", "created_at", "created_by", "updated_at", "updated_by") + VALUES( + ${{ uuid: a.aid }}, + ${{ text: a.name }}, + ${{ date: a.created_at }}, + ${a.created_by ? { uuid: a.created_by } : { null: true }}, + ${{ date: a.updated_at }}, + ${a.updated_by ? { uuid: a.updated_by } : { null: true }} + ) + `); + break; + } + case "resident": { + const r = data as Resident; + db.run(sql` + INSERT INTO "residents"("uid", "aid", "flags", "created_at", "created_by", "updated_at", "updated_by") + VALUES( + ${{ uuid: r.uid }}, + ${{ uuid: r.aid }}, + ${{ jsonb: r.flags }}, + ${{ date: r.created_at }}, + ${r.created_by ? { uuid: r.created_by } : { null: true }}, + ${{ date: r.updated_at }}, + ${r.updated_by ? { uuid: r.updated_by } : { null: true }} + ) + `); + break; + } + case "apikey": { + const k = data as ClientApikey; + // `token` is never exported; mint a fresh unique one so the record's + // metadata (kid/permissions/expiry) survives even though the original + // secret cannot. + db.run(sql` + INSERT INTO "apikeys"("uid", "kid", "token", "name", "permissions", "created_at", "expires_at") + VALUES( + ${{ uuid: k.uid }}, + ${{ uuid: k.kid }}, + ${{ text: createApikeyToken() }}, + ${{ text: k.name }}, + ${{ jsonb: k.permissions }}, + ${{ date: k.created_at }}, + ${k.expires_at ? { date: k.expires_at } : { null: true }} + ) + `); + break; + } + case "note": { + const n = data as Note; + db.run(sql` + INSERT INTO "notes"("nid", "aid", "name", "content", "properties", "created_at", "created_by", "updated_at", "updated_by") + VALUES( + ${{ uuid: n.nid }}, + ${{ uuid: n.aid }}, + ${{ text: n.name }}, + ${{ text: n.content ?? "" }}, + ${{ jsonb: n.properties }}, + ${{ date: n.created_at }}, + ${n.created_by ? { uuid: n.created_by } : { null: true }}, + ${{ date: n.updated_at }}, + ${n.updated_by ? { uuid: n.updated_by } : { null: true }} + ) + `); + break; + } + } + } } diff --git a/src/db/sqlite/query.ts b/src/db/sqlite/query.ts index 81b31e9..0c8898f 100644 --- a/src/db/sqlite/query.ts +++ b/src/db/sqlite/query.ts @@ -118,10 +118,10 @@ export function selectClientApikey( } export function selectClientApikeys( db: WrappedDb, - where: SqlCode, + where?: SqlCode, ): ClientApikey[] { const rawApikeys = db.all( - sql`${sqlClientApikey} WHERE ${where}`, + where ? sql`${sqlClientApikey} WHERE ${where}` : sqlClientApikey, ); return rawApikeys.map(sqliteToClientApikey); } diff --git a/src/db/types/ExportImport.ts b/src/db/types/ExportImport.ts new file mode 100644 index 0000000..9c600b2 --- /dev/null +++ b/src/db/types/ExportImport.ts @@ -0,0 +1,68 @@ +import type { DbInterface } from "./DbInterface.js"; + +/** + * The kinds of records that can travel through an export/import stream. + * + * `Session` is intentionally excluded: it has no CRUD/list surface in + * {@link DbInterface} and is ephemeral/non-portable between instances. + */ +export type ExportKind = "user" | "abode" | "resident" | "apikey" | "note"; + +export type ExportFilter = { + /** Include only these kinds; omit = all kinds. */ + kinds?: ExportKind[]; + /** Excluded after `kinds` is applied. */ + excludeKinds?: ExportKind[]; + /** aid allowlist — scopes abode/resident/note. */ + abodes?: string[]; + /** uid allowlist — scopes user/apikey. */ + users?: string[]; +}; + +export interface ExportOptions { + filter?: ExportFilter; + signal?: AbortSignal; +} + +export interface Exportable { + /** Produce a stream of NDJSON lines (one JSON envelope per line). */ + export(options?: ExportOptions): NodeJS.ReadableStream; +} + +export interface ImportOptions { + filter?: ExportFilter; + signal?: AbortSignal; +} + +export interface ImportResult { + counts: Partial>; +} + +export interface Importable { + import( + source: NodeJS.ReadableStream, + options?: ImportOptions, + ): Promise; +} + +/** + * The NDJSON envelope written/read for every line. The leading line is a + * `meta` record; a trailing `error` record may appear if the source failed + * after streaming had already begun. + */ +export type ExportEnvelope = + | { kind: "meta"; data: ExportMeta } + | { kind: ExportKind; data: unknown } + | { kind: "error"; data: { message: string; code?: string } }; + +export type ExportMeta = { + v: number; + exportedAt: string; + source: string; + /** The *effective* filter actually applied (may be narrower than requested). */ + filter: ExportFilter; +}; + +export function isExportable(db: DbInterface): db is DbInterface & Exportable { + return typeof (db as Partial).export === "function"; +} diff --git a/src/webapi/apirouter.ts b/src/webapi/apirouter.ts index 8ff5f0d..ccf55ac 100644 --- a/src/webapi/apirouter.ts +++ b/src/webapi/apirouter.ts @@ -16,6 +16,29 @@ import { } from "../schema/validators.js"; import { authenticate } from "./middleware/authenticate.js"; import { InvalidAbodeError, NotFoundAbodeError } from "../db/types/errors.js"; +import { isExportable } from "../db/types/ExportImport.js"; +import type { ExportFilter, ExportKind } from "../db/types/ExportImport.js"; +import { intersectExportFilters, isExportKind } from "../db/export/filter.js"; +import { computeForcedExportFilter } from "./exportScope.js"; + +function parseExportFilter(query: Record): ExportFilter { + const list = (v: unknown): string[] | undefined => { + if (typeof v !== "string" || !v) return undefined; + return v.split(",").filter(Boolean); + }; + const kinds = (v: unknown): ExportKind[] | undefined => + list(v)?.filter(isExportKind); + const filter: ExportFilter = {}; + const k = kinds(query.kinds); + if (k) filter.kinds = k; + const ek = kinds(query.excludeKinds); + if (ek) filter.excludeKinds = ek; + const abodes = list(query.abodes); + if (abodes) filter.abodes = abodes; + const users = list(query.users); + if (users) filter.users = users; + return filter; +} export function apirouter(db: BackendDbInterface): KoaRouter { const router = new KoaRouter(); @@ -42,6 +65,31 @@ export function apirouter(db: BackendDbInterface): KoaRouter { ctx.status = 204; }); + router.get("/export", authenticate(db), async (ctx) => { + // Everything that can throw a domain error runs *before* any byte is + // written, so `convertError` still applies. Once `ctx.body` is a stream, + // a mid-stream failure surfaces as a trailing `error` NDJSON line instead. + const forced = await computeForcedExportFilter(db, { + user: ctx.user!, + session: ctx.session!, + }); + const effective = intersectExportFilters( + parseExportFilter(ctx.query), + forced, + ); + if (!isExportable(db)) { + ctx.status = 501; + ctx.body = { ok: false, error: "export_unsupported" }; + return; + } + const ac = new AbortController(); + ctx.res.on("close", () => { + if (!ctx.res.writableEnded) ac.abort(); + }); + ctx.type = "application/x-ndjson"; + ctx.body = db.export({ filter: effective, signal: ac.signal }); + }); + router.use("/users", authenticate(db)); router.get("/users", async (ctx) => { ctx.body = await db.listUsers(); diff --git a/src/webapi/exportScope.ts b/src/webapi/exportScope.ts new file mode 100644 index 0000000..aa79bc8 --- /dev/null +++ b/src/webapi/exportScope.ts @@ -0,0 +1,58 @@ +import type { Context } from "koa"; +import type { BackendDbInterface } from "../db/types/DbInterface.js"; +import type { ClientUser } from "../db/types/User.js"; +import type { ExportFilter } from "../db/types/ExportImport.js"; + +/** + * Compute the export scope that must be *forced* on a caller, independent of + * anything they requested. Returns `null` when the caller is unrestricted (a + * global admin whose credential imposes no narrowing) — their own filter, if + * any, is then honored verbatim as a voluntary narrowing. + * + * Otherwise returns `{ abodes, users }`: the abodes the caller resides in, and + * the users needed to keep that data referentially whole (the caller plus every + * co-resident of those abodes). This is the maximum a non-admin may export; the + * route intersects it with any caller-supplied filter (never a union). + */ +export async function computeForcedExportFilter( + db: BackendDbInterface, + ctx: { user: ClientUser; session: NonNullable }, +): Promise { + const { user, session } = ctx; + + if (user.flags.admin) { + if (session.source !== "apikey") return null; + const p = session.key.permissions; + const unrestricted = + !!p.admin && + !!p.all && + !p.restrict_users?.length && + !p.restrict_abodes?.length; + if (unrestricted) return null; + } + + const residencies = await db.listResidentsByUserId(user.uid); + const abodeSet = new Set(residencies.map((r) => r.aid)); + const userSet = new Set([user.uid]); + for (const aid of abodeSet) { + for (const u of await db.listUsersByAbodeId(aid)) userSet.add(u.uid); + } + + let abodes = [...abodeSet]; + let users = [...userSet]; + + // An apikey can only narrow what its owning user could otherwise export. + if (session.source === "apikey") { + const p = session.key.permissions; + if (p.restrict_abodes?.length) { + const allow = new Set(p.restrict_abodes); + abodes = abodes.filter((a) => allow.has(a)); + } + if (p.restrict_users?.length) { + const allow = new Set(p.restrict_users); + users = users.filter((u) => allow.has(u)); + } + } + + return { abodes, users }; +} diff --git a/test/tools/export-import.test.ts b/test/tools/export-import.test.ts new file mode 100644 index 0000000..359c529 --- /dev/null +++ b/test/tools/export-import.test.ts @@ -0,0 +1,715 @@ +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import readline from "node:readline"; +import { Readable } from "node:stream"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import Koa from "koa"; +import { createTestDb, type TestDb } from "../helpers/sqlite.js"; +import { apirouter } from "../../src/webapi/apirouter.js"; +import { computeForcedExportFilter } from "../../src/webapi/exportScope.js"; +import { + intersectExportFilters, + recordAllowed, + kindAllowed, +} from "../../src/db/export/filter.js"; +import { inspectExportStream } from "../../src/db/export/inspect.js"; +import { hashPassword } from "../../src/util/hash.js"; +import type { ClientUser } from "../../src/db/types/User.js"; + +// --------------------------------------------------------------------------- +// helpers +// --------------------------------------------------------------------------- + +async function streamToString(s: NodeJS.ReadableStream): Promise { + let out = ""; + for await (const chunk of s) out += chunk; + return out; +} + +function parseLines(ndjson: string): { kind: string; data: any }[] { + return ndjson + .split("\n") + .filter(Boolean) + .map((l) => JSON.parse(l)); +} + +/** Normalise `*_at` fields to epoch ms so formatting never breaks equality. */ +function norm(obj: Record): Record { + const out: Record = {}; + for (const [k, v] of Object.entries(obj)) { + out[k] = k.endsWith("_at") + ? v == null + ? null + : new Date(v as string).getTime() + : v; + } + return out; +} +function normSorted( + arr: Record[], + key: (x: any) => string, +): Record[] { + return arr.map(norm).sort((a, b) => key(a).localeCompare(key(b))); +} + +interface Seed { + admin: ClientUser; + normal: ClientUser; + co: ClientUser; + aid1: string; + aid2: string; + nid1: string; + nid2: string; +} + +async function seed(db: TestDb["db"]): Promise { + const pw = await hashPassword("password"); + const admin = await db.createUser({ + email: "admin@test.example", + name: "Admin", + password: pw, + flags: { admin: true }, + }); + const normal = await db.createUser({ + email: "normal@test.example", + name: "Normal", + password: pw, + flags: {}, + }); + const co = await db.createUser({ + email: "co@test.example", + name: "Co Resident", + password: pw, + flags: {}, + }); + const abode1 = await db.createAbode( + { name: "Abode One" }, + { uid: admin.uid }, + ); + const abode2 = await db.createAbode( + { name: "Abode Two" }, + { uid: admin.uid }, + ); + await db.createResident( + { uid: normal.uid, aid: abode1.aid, flags: {} }, + { uid: admin.uid }, + ); + await db.createResident( + { uid: co.uid, aid: abode1.aid, flags: {} }, + { uid: admin.uid }, + ); + await db.createResident( + { uid: admin.uid, aid: abode2.aid, flags: { admin: true } }, + { uid: admin.uid }, + ); + await db.createApikey({ + uid: normal.uid, + name: "normal key", + permissions: {}, + expires_at: null, + }); + const note1 = await db.createNote( + { + aid: abode1.aid, + name: "Note One", + content: "hello", + properties: { type: "note" }, + }, + { uid: normal.uid }, + ); + const note2 = await db.createNote( + { + aid: abode2.aid, + name: "Note Two", + content: "world", + properties: { type: "note" }, + }, + { uid: admin.uid }, + ); + return { + admin, + normal, + co, + aid1: abode1.aid, + aid2: abode2.aid, + nid1: note1.nid, + nid2: note2.nid, + }; +} + +// --------------------------------------------------------------------------- +// 1. round-trip +// --------------------------------------------------------------------------- + +describe("export/import: sqlite -> sqlite round-trip", () => { + it("reproduces users, abodes, residents, apikeys, notes", async () => { + const src = await createTestDb(); + const dst = await createTestDb(); + try { + const s = await seed(src.db); + + const result = await dst.db.import(src.db.export()); + assert.ok(result.counts.user && result.counts.user >= 3); + assert.ok(result.counts.note && result.counts.note >= 2); + + assert.deepEqual( + normSorted(await dst.db.listUsers(), (u) => u.uid), + normSorted(await src.db.listUsers(), (u) => u.uid), + ); + assert.deepEqual( + normSorted(await dst.db.listAbodes(), (a) => a.aid), + normSorted(await src.db.listAbodes(), (a) => a.aid), + ); + assert.deepEqual( + normSorted(await dst.db.listResidents(), (r) => r.uid + r.aid), + normSorted(await src.db.listResidents(), (r) => r.uid + r.aid), + ); + // apikeys: token is regenerated on import, so the ClientApikey view + // (which omits token) must still match exactly. + assert.deepEqual( + normSorted(await dst.db.listApikeysByUser(s.normal.uid), (k) => k.kid), + normSorted(await src.db.listApikeysByUser(s.normal.uid), (k) => k.kid), + ); + // notes (full, with content) + const srcNote = await src.db.getNoteById(s.nid1); + const dstNote = await dst.db.getNoteById(s.nid1); + assert.deepEqual(norm(dstNote), norm(srcNote)); + } finally { + src.close(); + dst.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 2. filter narrowing +// --------------------------------------------------------------------------- + +describe("export/import: filter narrowing", () => { + it("scopes abode/resident/note to the requested abodes", async () => { + const src = await createTestDb(); + try { + const s = await seed(src.db); + const ndjson = await streamToString( + src.db.export({ filter: { abodes: [s.aid1] } }), + ); + const lines = parseLines(ndjson); + + const abodeAids = lines + .filter((l) => l.kind === "abode") + .map((l) => l.data.aid); + assert.deepEqual(abodeAids, [s.aid1]); + + const noteAids = new Set( + lines.filter((l) => l.kind === "note").map((l) => l.data.aid), + ); + assert.ok(noteAids.has(s.aid1)); + assert.ok(!noteAids.has(s.aid2)); + + const residentAids = new Set( + lines.filter((l) => l.kind === "resident").map((l) => l.data.aid), + ); + assert.ok(!residentAids.has(s.aid2)); + } finally { + src.close(); + } + }); + + it("imports the narrowed dump into a fresh db with the same scope", async () => { + const src = await createTestDb(); + const dst = await createTestDb(); + try { + const s = await seed(src.db); + await dst.db.import(src.db.export({ filter: { abodes: [s.aid1] } })); + + const abodes = await dst.db.listAbodes(); + assert.deepEqual( + abodes.map((a) => a.aid), + [s.aid1], + ); + const notes = await dst.db.listNotesByAbodeId(s.aid1); + assert.equal(notes.length, 1); + assert.equal((await dst.db.listNotesByAbodeId(s.aid2)).length, 0); + } finally { + src.close(); + dst.close(); + } + }); + + it("kinds filter selects only the requested kinds", async () => { + const src = await createTestDb(); + try { + await seed(src.db); + const ndjson = await streamToString( + src.db.export({ filter: { kinds: ["abode"] } }), + ); + const kinds = new Set(parseLines(ndjson).map((l) => l.kind)); + assert.ok(kinds.has("abode")); + assert.ok(!kinds.has("user")); + assert.ok(!kinds.has("note")); + } finally { + src.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 3. forced-filter enforcement +// --------------------------------------------------------------------------- + +describe("exportScope: computeForcedExportFilter", () => { + it("returns null for a global admin on a basic/session credential", async () => { + const t = await createTestDb(); + try { + const s = await seed(t.db); + const forced = await computeForcedExportFilter(t.db, { + user: s.admin, + session: { source: "basic" }, + }); + assert.equal(forced, null); + } finally { + t.close(); + } + }); + + it("returns null for a global admin with an unrestricted apikey", async () => { + const t = await createTestDb(); + try { + const s = await seed(t.db); + const forced = await computeForcedExportFilter(t.db, { + user: s.admin, + session: { + source: "apikey", + key: { + kid: "k", + uid: s.admin.uid, + name: "k", + permissions: { admin: true, all: true }, + created_at: new Date().toISOString(), + expires_at: null, + }, + }, + }); + assert.equal(forced, null); + } finally { + t.close(); + } + }); + + it("forces a non-admin to their abodes + co-residents", async () => { + const t = await createTestDb(); + try { + const s = await seed(t.db); + const forced = await computeForcedExportFilter(t.db, { + user: s.normal, + session: { source: "basic" }, + }); + assert.ok(forced); + assert.deepEqual(forced!.abodes, [s.aid1]); + assert.deepEqual( + new Set(forced!.users), + new Set([s.normal.uid, s.co.uid]), + ); + assert.ok(!forced!.users!.includes(s.admin.uid)); + + // A caller requesting a wider abode never gets it: intersection, not union. + const effective = intersectExportFilters( + { abodes: [s.aid1, s.aid2] }, + forced, + ); + assert.deepEqual(effective.abodes, [s.aid1]); + assert.ok(!effective.abodes!.includes(s.aid2)); + } finally { + t.close(); + } + }); + + it("intersects a non-admin apikey with restrict_abodes", async () => { + const t = await createTestDb(); + try { + const s = await seed(t.db); + const forced = await computeForcedExportFilter(t.db, { + user: s.normal, + session: { + source: "apikey", + key: { + kid: "k", + uid: s.normal.uid, + name: "k", + permissions: { restrict_abodes: [s.aid2] }, + created_at: new Date().toISOString(), + expires_at: null, + }, + }, + }); + // normal resides only in aid1; restrict to aid2 => empty intersection. + assert.deepEqual(forced!.abodes, []); + } finally { + t.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 4. cancellation, export side +// --------------------------------------------------------------------------- + +describe("export cancellation", () => { + it("stops querying tables after the destination aborts", async () => { + const t = await createTestDb(); + try { + const pw = await hashPassword("password"); + // Enough users that the first table can't be buffered in one go, so the + // generator backpressures mid-`user` and never reaches later tables. + for (let i = 0; i < 400; i++) { + await t.db.createUser({ + email: `bulk-${i}@test.example`, + name: `Bulk ${i}`, + password: pw, + flags: {}, + }); + } + + let allCalls = 0; + const orig = t.wrapped.all.bind(t.wrapped); + (t.wrapped as { all: unknown }).all = (stmt: never) => { + allCalls++; + return orig(stmt); + }; + + const ac = new AbortController(); + const stream = t.db.export({ signal: ac.signal }); + const rl = readline.createInterface({ input: stream }); + + let lines = 0; + let callsAtAbort = -1; + for await (const _ of rl) { + lines++; + if (lines === 3) { + ac.abort(); + callsAtAbort = allCalls; + } + } + + assert.ok(callsAtAbort >= 1, "at least the first table was queried"); + assert.equal(allCalls, callsAtAbort, "no further queries after abort"); + assert.ok(allCalls < 5, "did not materialise all five tables"); + } finally { + t.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 5. cancellation, import side +// --------------------------------------------------------------------------- + +describe("import cancellation", () => { + it("rolls back and leaves no dangling transaction when the source errors", async () => { + const dst = await createTestDb(); + try { + function line(kind: string, data: unknown): string { + return JSON.stringify({ kind, data }) + "\n"; + } + const now = new Date().toISOString(); + const mkUser = (i: number) => ({ + uid: crypto.randomUUID(), + email: `imp-${i}@test.example`, + name: `Imp ${i}`, + flags: {}, + created_at: now, + updated_at: now, + }); + + // Emits a couple of valid user records, then throws mid-stream. + const source = Readable.from( + (async function* () { + yield line("meta", { v: 1 }); + yield line("user", mkUser(1)); + yield line("user", mkUser(2)); + throw new Error("source exploded"); + })(), + ); + + await assert.rejects(() => dst.db.import(source), /source exploded/); + + // ROLLBACK happened: nothing persisted. + assert.equal((await dst.db.listUsers()).length, 0); + + // No lingering open transaction: a follow-up write succeeds immediately. + const pw = await hashPassword("password"); + const created = await dst.db.createUser({ + email: "after@test.example", + name: "After", + password: pw, + flags: {}, + }); + assert.ok(created.uid); + } finally { + dst.close(); + } + }); + + it("aborts cleanly via signal and commits nothing", async () => { + const dst = await createTestDb(); + try { + const ac = new AbortController(); + const now = new Date().toISOString(); + const source = Readable.from( + (async function* () { + yield JSON.stringify({ kind: "meta", data: { v: 1 } }) + "\n"; + yield JSON.stringify({ + kind: "user", + data: { + uid: crypto.randomUUID(), + email: "abort@test.example", + name: "Abort", + flags: {}, + created_at: now, + updated_at: now, + }, + }) + "\n"; + ac.abort(); + // Keep the stream alive so abort — not EOF — ends the import. + await new Promise((r) => setTimeout(r, 1000)); + })(), + ); + + await assert.rejects(() => dst.db.import(source, { signal: ac.signal })); + assert.equal((await dst.db.listUsers()).length, 0); + } finally { + dst.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 6. api export endpoint +// --------------------------------------------------------------------------- + +describe("GET /export endpoint", () => { + let t: TestDb; + let s: Seed; + let url: string; + let close: () => Promise; + + const basic = (email: string) => + "Basic " + Buffer.from(`${email}:password`).toString("base64"); + + before(async () => { + t = await createTestDb(); + s = await seed(t.db); + const app = new Koa(); + const router = apirouter(t.db); + app.use(router.routes()); + app.use(router.allowedMethods()); + const server = createServer(app.callback()); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const { port } = server.address() as AddressInfo; + url = `http://127.0.0.1:${port}`; + close = () => + new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + }); + + after(async () => { + await close(); + t.close(); + }); + + it("global admin exports everything", async () => { + const res = await fetch(`${url}/export`, { + headers: { Authorization: basic(s.admin.email) }, + }); + assert.equal(res.status, 200); + const lines = parseLines(await res.text()); + const abodeAids = new Set( + lines.filter((l) => l.kind === "abode").map((l) => l.data.aid), + ); + assert.ok(abodeAids.has(s.aid1)); + assert.ok(abodeAids.has(s.aid2)); + const userUids = new Set( + lines.filter((l) => l.kind === "user").map((l) => l.data.uid), + ); + assert.ok(userUids.has(s.admin.uid)); + assert.ok(userUids.has(s.normal.uid)); + }); + + it("non-admin is force-scoped even when requesting wider abodes", async () => { + const res = await fetch(`${url}/export?abodes=${s.aid1},${s.aid2}`, { + headers: { Authorization: basic(s.normal.email) }, + }); + assert.equal(res.status, 200); + const lines = parseLines(await res.text()); + + const abodeAids = new Set( + lines.filter((l) => l.kind === "abode").map((l) => l.data.aid), + ); + assert.ok(abodeAids.has(s.aid1)); + assert.ok(!abodeAids.has(s.aid2), "aid2 forced out of scope"); + + const userUids = new Set( + lines.filter((l) => l.kind === "user").map((l) => l.data.uid), + ); + assert.ok(userUids.has(s.normal.uid)); + assert.ok(userUids.has(s.co.uid)); + assert.ok(!userUids.has(s.admin.uid), "admin not a co-resident of aid1"); + + // The meta line records the *effective* (narrowed) filter. + const meta = lines.find((l) => l.kind === "meta"); + assert.ok(meta); + assert.deepEqual(meta!.data.filter.abodes, [s.aid1]); + }); + + it("stops server-side querying shortly after the client aborts", async () => { + // Dedicated db/server. Rows are deliberately large (and inserted in bulk so + // the seed stays fast) so the users table alone can't fit in the socket + // buffer — the export generator backpressures mid-`user` and never reaches + // later tables while the client is still holding the connection open. + const big = await createTestDb(); + let server: ReturnType | undefined; + try { + const pw = await hashPassword("password"); + await big.db.createUser({ + email: "a@test.example", + name: "A", + password: pw, + flags: { admin: true }, + }); + const { sql, db: raw } = big.db._; + // ~32MB of user rows — far more than any socket/kernel buffer can hold, + // so the generator is guaranteed to still be suspended mid-`user` (never + // having queried a later table) when the client aborts. + const bigName = "x".repeat(16000); + raw.multi(() => { + for (let i = 0; i < 2000; i++) { + raw.run(sql` + INSERT INTO "users"("uid", "email", "name", "flags") + VALUES( + ${{ uuid: crypto.randomUUID() }}, + ${{ text: `b-${i}@test.example` }}, + ${{ text: bigName }}, + ${{ jsonb: {} }} + ) + `); + } + }); + + let allCalls = 0; + const orig = big.wrapped.all.bind(big.wrapped); + (big.wrapped as { all: unknown }).all = (stmt: never) => { + allCalls++; + return orig(stmt); + }; + + const app = new Koa(); + app.on("error", () => {}); // swallow the expected ECONNRESET on abort + const router = apirouter(big.db); + app.use(router.routes()); + app.use(router.allowedMethods()); + server = createServer(app.callback()); + await new Promise((r) => server!.listen(0, "127.0.0.1", r)); + const { port } = server.address() as AddressInfo; + + const ac = new AbortController(); + const res = await fetch(`http://127.0.0.1:${port}/export`, { + headers: { Authorization: basic("a@test.example") }, + signal: ac.signal, + }); + const reader = res.body!.getReader(); + await reader.read(); // first chunk — server has begun streaming users + const callsWhileStreaming = allCalls; + assert.ok(callsWhileStreaming >= 1, "server queried the first table"); + ac.abort(); + await reader.cancel().catch(() => {}); + + await new Promise((r) => setTimeout(r, 250)); + const settled = allCalls; + await new Promise((r) => setTimeout(r, 250)); + assert.equal(allCalls, settled, "no further queries after abort"); + assert.ok(allCalls < 5, "did not materialise all five tables"); + } finally { + if (server) await new Promise((r) => server!.close(() => r())); + big.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// 7. inspect utility +// --------------------------------------------------------------------------- + +describe("inspectExportStream", () => { + it("tallies counts and reads meta without a db", async () => { + const src = await createTestDb(); + try { + await seed(src.db); + const ndjson = await streamToString(src.db.export()); + const { counts, meta } = await inspectExportStream( + Readable.from([ndjson]), + ); + assert.equal(meta?.v, 1); + assert.equal(meta?.source, "sqlite"); + assert.ok((counts.user ?? 0) >= 3); + assert.ok((counts.abode ?? 0) >= 2); + assert.ok((counts.note ?? 0) >= 2); + } finally { + src.close(); + } + }); + + it("stopAfterKinds short-circuits once every requested kind is seen", async () => { + const src = await createTestDb(); + try { + await seed(src.db); + const ndjson = await streamToString(src.db.export()); + const { counts } = await inspectExportStream(Readable.from([ndjson]), { + stopAfterKinds: ["user"], + }); + assert.ok((counts.user ?? 0) >= 1); + // stopped as soon as the first user was seen, before later kinds. + assert.equal(counts.note ?? 0, 0); + } finally { + src.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// filter unit checks +// --------------------------------------------------------------------------- + +describe("filter helpers", () => { + it("kindAllowed respects kinds/excludeKinds", () => { + assert.equal(kindAllowed({ kinds: ["abode"] }, "abode"), true); + assert.equal(kindAllowed({ kinds: ["abode"] }, "user"), false); + assert.equal(kindAllowed({ excludeKinds: ["note"] }, "note"), false); + assert.equal(kindAllowed(undefined, "note"), true); + }); + + it("recordAllowed scopes by aid/uid per kind", () => { + assert.equal( + recordAllowed({ abodes: ["a1"] }, "abode", { aid: "a1" }), + true, + ); + assert.equal( + recordAllowed({ abodes: ["a1"] }, "abode", { aid: "a2" }), + false, + ); + assert.equal( + recordAllowed({ users: ["u1"] }, "apikey", { uid: "u1" }), + true, + ); + assert.equal( + recordAllowed({ users: ["u1"] }, "apikey", { uid: "u2" }), + false, + ); + // abodes allowlist does not constrain user records + assert.equal( + recordAllowed({ abodes: ["a1"] }, "user", { uid: "u9" }), + true, + ); + }); +}); -- 2.54.0 From aadc950e2405e8e6204c37e86e73a7bbdddb3c87 Mon Sep 17 00:00:00 2001 From: Codinget Date: Wed, 22 Jul 2026 23:16:31 +0000 Subject: [PATCH 2/4] fix: scope apikey export to the caller for non-admins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A non-admin's forced export filter includes co-resident *user* records so abode/resident data isn't left with dangling references, but the same `users` allowlist was also governing `apikey` records — leaking co-residents' apikey metadata (name/permissions/expiry, though not the secret token). Add a dedicated `apikeys` uid-allowlist to ExportFilter that scopes apikey records specifically, falling back to `users` when absent (so existing unfiltered/voluntary-narrowing behaviour and the round-trip are unchanged). computeForcedExportFilter now sets it to the caller alone (intersected with an apikey credential's restrict_users), so a non-admin can only ever export their own keys. Global admins (forced filter null) are unaffected. Co-Authored-By: Claude Opus 4.8 --- src/db/export/filter.ts | 13 +++++++++---- src/db/types/ExportImport.ts | 10 +++++++++- src/webapi/exportScope.ts | 16 +++++++++++----- test/tools/export-import.test.ts | 20 ++++++++++++++++++++ 4 files changed, 49 insertions(+), 10 deletions(-) diff --git a/src/db/export/filter.ts b/src/db/export/filter.ts index 9f78890..23876bd 100644 --- a/src/db/export/filter.ts +++ b/src/db/export/filter.ts @@ -24,9 +24,10 @@ export function kindAllowed( } /** - * Whether an individual record passes a filter's `abodes`/`users` allowlists. - * `abodes` scopes abode/resident/note (by aid); `users` scopes user/apikey - * (by uid). An absent allowlist means "unrestricted". + * Whether an individual record passes a filter's `abodes`/`users`/`apikeys` + * allowlists. `abodes` scopes abode/resident/note (by aid); `users` scopes user + * (by uid); `apikey` records are scoped by `apikeys` when present, else by + * `users`. An absent allowlist means "unrestricted". */ export function recordAllowed( filter: ExportFilter | undefined, @@ -36,8 +37,11 @@ export function recordAllowed( if (!filter) return true; switch (kind) { case "user": - case "apikey": return !filter.users || filter.users.includes(record.uid as string); + case "apikey": { + const allow = filter.apikeys ?? filter.users; + return !allow || allow.includes(record.uid as string); + } case "abode": case "resident": case "note": @@ -75,5 +79,6 @@ export function intersectExportFilters( excludeKinds: unionList(a.excludeKinds, b.excludeKinds), abodes: intersectList(a.abodes, b.abodes), users: intersectList(a.users, b.users), + apikeys: intersectList(a.apikeys, b.apikeys), }; } diff --git a/src/db/types/ExportImport.ts b/src/db/types/ExportImport.ts index 9c600b2..e1afa39 100644 --- a/src/db/types/ExportImport.ts +++ b/src/db/types/ExportImport.ts @@ -15,8 +15,16 @@ export type ExportFilter = { excludeKinds?: ExportKind[]; /** aid allowlist — scopes abode/resident/note. */ abodes?: string[]; - /** uid allowlist — scopes user/apikey. */ + /** uid allowlist — scopes user (and apikey, unless `apikeys` is set). */ users?: string[]; + /** + * uid allowlist scoping apikey records specifically. When set it takes + * precedence over `users` for the `apikey` kind — used to export a + * non-admin's own keys while still exporting co-residents' *user* records + * for referential integrity, without leaking their apikey metadata. Absent = + * fall back to `users`. + */ + apikeys?: string[]; }; export interface ExportOptions { diff --git a/src/webapi/exportScope.ts b/src/webapi/exportScope.ts index aa79bc8..12e0b97 100644 --- a/src/webapi/exportScope.ts +++ b/src/webapi/exportScope.ts @@ -9,10 +9,13 @@ import type { ExportFilter } from "../db/types/ExportImport.js"; * global admin whose credential imposes no narrowing) — their own filter, if * any, is then honored verbatim as a voluntary narrowing. * - * Otherwise returns `{ abodes, users }`: the abodes the caller resides in, and - * the users needed to keep that data referentially whole (the caller plus every - * co-resident of those abodes). This is the maximum a non-admin may export; the - * route intersects it with any caller-supplied filter (never a union). + * Otherwise returns `{ abodes, users, apikeys }`: the abodes the caller resides + * in, the users needed to keep that data referentially whole (the caller plus + * every co-resident of those abodes), and — scoped tighter than `users` — + * apikeys limited to the caller alone, so a non-admin never exports another + * user's apikey metadata even though that user's record is included. This is + * the maximum a non-admin may export; the route intersects it with any + * caller-supplied filter (never a union). */ export async function computeForcedExportFilter( db: BackendDbInterface, @@ -40,6 +43,8 @@ export async function computeForcedExportFilter( let abodes = [...abodeSet]; let users = [...userSet]; + // apikeys are self-only for non-admins, regardless of co-residency. + let apikeys = [user.uid]; // An apikey can only narrow what its owning user could otherwise export. if (session.source === "apikey") { @@ -51,8 +56,9 @@ export async function computeForcedExportFilter( if (p.restrict_users?.length) { const allow = new Set(p.restrict_users); users = users.filter((u) => allow.has(u)); + apikeys = apikeys.filter((u) => allow.has(u)); } } - return { abodes, users }; + return { abodes, users, apikeys }; } diff --git a/test/tools/export-import.test.ts b/test/tools/export-import.test.ts index 359c529..c8bdaca 100644 --- a/test/tools/export-import.test.ts +++ b/test/tools/export-import.test.ts @@ -109,6 +109,12 @@ async function seed(db: TestDb["db"]): Promise { permissions: {}, expires_at: null, }); + await db.createApikey({ + uid: co.uid, + name: "co key", + permissions: {}, + expires_at: null, + }); const note1 = await db.createNote( { aid: abode1.aid, @@ -312,6 +318,9 @@ describe("exportScope: computeForcedExportFilter", () => { new Set([s.normal.uid, s.co.uid]), ); assert.ok(!forced!.users!.includes(s.admin.uid)); + // apikeys are self-only, even though co is a co-resident whose user + // record is exported for referential integrity. + assert.deepEqual(forced!.apikeys, [s.normal.uid]); // A caller requesting a wider abode never gets it: intersection, not union. const effective = intersectExportFilters( @@ -557,6 +566,17 @@ describe("GET /export endpoint", () => { assert.ok(userUids.has(s.co.uid)); assert.ok(!userUids.has(s.admin.uid), "admin not a co-resident of aid1"); + // apikeys are self-only: the caller's own key is exported, but a + // co-resident's key metadata is NOT, even though their user record is. + const apikeyUids = lines + .filter((l) => l.kind === "apikey") + .map((l) => l.data.uid); + assert.deepEqual(new Set(apikeyUids), new Set([s.normal.uid])); + assert.ok( + !apikeyUids.includes(s.co.uid), + "co-resident apikey metadata must not leak", + ); + // The meta line records the *effective* (narrowed) filter. const meta = lines.find((l) => l.kind === "meta"); assert.ok(meta); -- 2.54.0 From 3a56dcd9e598ca861dbfa14d31703f0669ece357 Mon Sep 17 00:00:00 2001 From: Codinget Date: Wed, 22 Jul 2026 23:36:21 +0000 Subject: [PATCH 3/4] feat: add export/import to the postgres backend; import auto-detects backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The export/import plan predated the postgres backend. Bring it up to parity: - PostgresInterface implements Exportable + Importable, mirroring the sqlite backend. Export is a signal-checked async generator (one query per table); import drives a transaction via `WrappedPool.multi`, which rolls back on any error/abort and commits only after the whole stream is consumed cleanly. - The `note` kind is skipped on postgres (its note CRUD is still unimplemented, so a pg database holds none) — a full dump from sqlite imports its user/abode/resident/apikey records and drops notes. - Unlike sqlite (PRAGMA foreign_keys=off), postgres keeps FK enforcement; the FK-safe insertion order keeps a full dump valid, and truly-dangling partial dumps will (correctly) fail. - abode-import now resolves the backend via getDbInterface instead of constructing SqliteInterface directly; isImportable keeps it from ever running over the remote (api) interface, which has no import. - Add isImportable(); make postgres selectClientApikeys' where optional. - Tests: exercise the real PostgresInterface export/import paths against an in-memory fake WrappedPgClient (no pg service in CI) — NDJSON shape, meta.source, filtering, note-skip, insert dispatch, and error/abort rollback. Co-Authored-By: Claude Opus 4.8 --- src/bin/abode-import.ts | 25 +- src/db/postgres/PostgresInterface.ts | 208 ++++++++++++- src/db/postgres/query.ts | 4 +- src/db/types/ExportImport.ts | 4 + test/backends/postgres/export-import.test.ts | 296 +++++++++++++++++++ 5 files changed, 524 insertions(+), 13 deletions(-) create mode 100644 test/backends/postgres/export-import.test.ts diff --git a/src/bin/abode-import.ts b/src/bin/abode-import.ts index d8e0c2e..501892b 100644 --- a/src/bin/abode-import.ts +++ b/src/bin/abode-import.ts @@ -1,9 +1,7 @@ import { createReadStream } from "node:fs"; -import { getWrappedDb } from "../db/sqlite/impl/index.js"; -import { SqliteInterface } from "../db/sqlite/SqliteInterface.js"; -import { parseSqliteUrl } from "../db/sqlite/url.js"; +import { getDbInterface } from "../db/index.js"; import { isExportKind } from "../db/export/filter.js"; -import type { ExportFilter } from "../db/types/ExportImport.js"; +import { isImportable, type ExportFilter } from "../db/types/ExportImport.js"; const args = process.argv.slice(2); @@ -15,14 +13,15 @@ function printUsage(err: boolean | string = false): never { } log("Usage:"); log("\tabode-import --help"); - log("\tabode-import [--kinds=...] \\"); + log("\tabode-import [--kinds=...] \\"); log( "\t [--exclude-kinds=...] [--abodes=aid,...] [--users=uid,...]", ); log(""); log( - "The target database must already be migrated (run abode-migrate first).", + "The target must be a local backend (sqlite or postgres), already migrated", ); + log("(run abode-migrate first). Remote (api) targets are not importable."); process.exit(err ? 1 : 0); } @@ -42,7 +41,7 @@ for (const arg of args) { const url = positional[0]; const input = positional[1]; -if (!url) printUsage("missing "); +if (!url) printUsage("missing "); if (!input) printUsage("missing "); if (positional.length > 2) printUsage("too many arguments"); @@ -73,9 +72,15 @@ if (abodes) filter.abodes = abodes; const users = parseList(flags.get("users")); if (users) filter.users = users; -// Import is sqlite-only: construct the backend directly rather than resolving -// generically, so it can never be pointed at a remote (api) target. -const db = new SqliteInterface(getWrappedDb(...parseSqliteUrl(url))); +// Resolve the backend generically. Import lives on the local backends (sqlite, +// postgres); the remote (api) interface has no `import`, so `isImportable` +// keeps it from ever running over HTTP. +const db = await getDbInterface(url); +if (!isImportable(db)) { + console.error(`Error: backend '${db.name}' does not support import`); + await db.close().catch(() => {}); + process.exit(1); +} const source = input === "-" ? process.stdin : createReadStream(input); diff --git a/src/db/postgres/PostgresInterface.ts b/src/db/postgres/PostgresInterface.ts index a92aa21..3be560c 100644 --- a/src/db/postgres/PostgresInterface.ts +++ b/src/db/postgres/PostgresInterface.ts @@ -43,8 +43,21 @@ import type { UpdateNote, } from "../types/Note.js"; import type { WrappedPgClient } from "./pool.js"; +import { Readable } from "node:stream"; +import readline from "node:readline"; +import type { + Exportable, + ExportKind, + ExportOptions, + Importable, + ImportOptions, + ImportResult, +} from "../types/ExportImport.js"; +import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js"; -export class PostgresInterface implements BackendDbInterface { +export class PostgresInterface + implements BackendDbInterface, Exportable, Importable +{ #db: WrappedPgClient; constructor(db: WrappedPgClient) { @@ -508,4 +521,197 @@ export class PostgresInterface implements BackendDbInterface { async listNotesByUserId(_uid: string): Promise { throw new Error("Unimplemented"); } + + export(options: ExportOptions = {}): NodeJS.ReadableStream { + const { filter, signal } = options; + const db = this.#db; + const source = this.name; + + // `note` is omitted: the postgres backend has no note CRUD yet, so a pg + // database can hold none. Each `load()` is a single query, run lazily and + // skipped once the destination aborts. + const tables: [ + ExportKind, + () => Promise<{ uid?: string; aid?: string }[]>, + ][] = [ + ["user", () => selectClientUsers(db)], + ["abode", () => selectAbodes(db)], + ["resident", () => selectResidents(db)], + ["apikey", () => selectClientApikeys(db)], + ]; + + async function* generate(): AsyncGenerator { + if (signal?.aborted) return; + yield JSON.stringify({ + kind: "meta", + data: { + v: 1, + exportedAt: new Date().toISOString(), + source, + filter: filter ?? {}, + }, + }) + "\n"; + try { + for (const [kind, load] of tables) { + if (signal?.aborted) return; + if (!kindAllowed(filter, kind)) continue; + for (const row of await load()) { + if (signal?.aborted) return; + if (recordAllowed(filter, kind, row)) { + yield JSON.stringify({ kind, data: row }) + "\n"; + } + } + } + } catch (e) { + if (signal?.aborted) return; + yield JSON.stringify({ + kind: "error", + data: { + message: e instanceof Error ? e.message : String(e), + code: e instanceof Error ? e.name : undefined, + }, + }) + "\n"; + } + } + + return Readable.from(generate()); + } + + async import( + source: NodeJS.ReadableStream, + options: ImportOptions = {}, + ): Promise { + this.#checkReadonly(); + const { filter, signal } = options; + const counts: Partial> = {}; + + const rl = readline.createInterface({ + input: source, + crlfDelay: Infinity, + signal, + }); + + // Postgres holds FK enforcement (unlike sqlite, which we toggle off): the + // FK-safe insertion order — user, abode, resident, apikey — keeps a full + // dump valid. `notes` are skipped entirely (unsupported on this backend). + // The transaction commits only if the whole stream is consumed cleanly; an + // error/abort rolls it back via `multi`. + try { + await this.#db.rethrow(() => + this.#db.multi(async (tx) => { + for await (const raw of rl) { + signal?.throwIfAborted(); + const line = raw.trim(); + if (!line) continue; + let parsed: { kind?: unknown; data?: unknown }; + try { + parsed = JSON.parse(line); + } catch { + throw new InvalidAbodeError(); + } + if (parsed.kind === "meta") continue; + if (parsed.kind === "error") { + throw new Error( + `export stream reported an error: ${ + (parsed.data as { message?: string })?.message ?? "unknown" + }`, + ); + } + if (!isExportKind(parsed.kind)) continue; + if (parsed.kind === "note") continue; // unsupported on postgres + if (!kindAllowed(filter, parsed.kind)) continue; + const data = parsed.data as { uid?: string; aid?: string }; + if (!recordAllowed(filter, parsed.kind, data)) continue; + await this.#importRecord(tx, parsed.kind, parsed.data); + counts[parsed.kind] = (counts[parsed.kind] ?? 0) + 1; + } + // An abort while blocked on the source closes readline without + // throwing, so re-check before the transaction commits. + signal?.throwIfAborted(); + }), + ); + } finally { + rl.close(); + } + + return { counts }; + } + + async #importRecord( + tx: WrappedPgClient, + kind: ExportKind, + data: unknown, + ): Promise { + switch (kind) { + case "user": { + const u = data as ClientUser; + // `password` is never exported; imported users land on the schema + // default ('#unset') and must reset before they can log in. + await tx.run(sql` + INSERT INTO "users"("uid", "email", "name", "flags", "created_at", "updated_at") + VALUES( + ${{ uuid: u.uid }}, + ${{ text: u.email }}, + ${{ text: u.name }}, + ${{ jsonb: u.flags }}, + ${{ date: u.created_at }}, + ${{ date: u.updated_at }} + ) + `); + break; + } + case "abode": { + const a = data as Abode; + await tx.run(sql` + INSERT INTO "abodes"("aid", "name", "created_at", "created_by", "updated_at", "updated_by") + VALUES( + ${{ uuid: a.aid }}, + ${{ text: a.name }}, + ${{ date: a.created_at }}, + ${a.created_by ? { uuid: a.created_by } : { null: true }}, + ${{ date: a.updated_at }}, + ${a.updated_by ? { uuid: a.updated_by } : { null: true }} + ) + `); + break; + } + case "resident": { + const r = data as Resident; + await tx.run(sql` + INSERT INTO "residents"("uid", "aid", "flags", "created_at", "created_by", "updated_at", "updated_by") + VALUES( + ${{ uuid: r.uid }}, + ${{ uuid: r.aid }}, + ${{ jsonb: r.flags }}, + ${{ date: r.created_at }}, + ${r.created_by ? { uuid: r.created_by } : { null: true }}, + ${{ date: r.updated_at }}, + ${r.updated_by ? { uuid: r.updated_by } : { null: true }} + ) + `); + break; + } + case "apikey": { + const k = data as ClientApikey; + // `token` is never exported; mint a fresh unique one so the record's + // metadata (kid/permissions/expiry) survives even though the original + // secret cannot. + await tx.run(sql` + INSERT INTO "apikeys"("uid", "kid", "token", "name", "permissions", "created_at", "expires_at") + VALUES( + ${{ uuid: k.uid }}, + ${{ uuid: k.kid }}, + ${{ text: createApikeyToken() }}, + ${{ text: k.name }}, + ${{ jsonb: k.permissions }}, + ${{ date: k.created_at }}, + ${k.expires_at ? { date: k.expires_at } : { null: true }} + ) + `); + break; + } + case "note": + break; // unsupported on postgres; skipped before reaching here + } + } } diff --git a/src/db/postgres/query.ts b/src/db/postgres/query.ts index 766bb04..d74f4d8 100644 --- a/src/db/postgres/query.ts +++ b/src/db/postgres/query.ts @@ -130,10 +130,10 @@ export async function selectClientApikey( } export async function selectClientApikeys( db: WrappedPgClient, - where: SqlCode, + where?: SqlCode, ): Promise { const rows = await db.all( - sql`${sqlClientApikey} WHERE ${where}`, + where ? sql`${sqlClientApikey} WHERE ${where}` : sqlClientApikey, ); return rows.map(pgToClientApikey); } diff --git a/src/db/types/ExportImport.ts b/src/db/types/ExportImport.ts index e1afa39..812defe 100644 --- a/src/db/types/ExportImport.ts +++ b/src/db/types/ExportImport.ts @@ -74,3 +74,7 @@ export type ExportMeta = { export function isExportable(db: DbInterface): db is DbInterface & Exportable { return typeof (db as Partial).export === "function"; } + +export function isImportable(db: DbInterface): db is DbInterface & Importable { + return typeof (db as Partial).import === "function"; +} diff --git a/test/backends/postgres/export-import.test.ts b/test/backends/postgres/export-import.test.ts new file mode 100644 index 0000000..3d1b789 --- /dev/null +++ b/test/backends/postgres/export-import.test.ts @@ -0,0 +1,296 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { Readable } from "node:stream"; +import { PostgresInterface } from "../../../src/db/postgres/PostgresInterface.js"; +import type { WrappedPgClient } from "../../../src/db/postgres/pool.js"; +import type { SqlCode } from "../../../src/db/postgres/sql.js"; + +// The postgres backend has no CI database, so these tests drive the real +// PostgresInterface.export/import code paths against an in-memory fake client. +// They verify control flow (NDJSON shape, meta.source, filtering, note +// skipping, counts, insert dispatch, abort -> rollback); the SQL-arg forms are +// the same {uuid}/{text}/{jsonb}/{date} patterns the pg backend's own CRUD +// already exercises against real Postgres. + +interface Rows { + users?: Record[]; + abodes?: Record[]; + residents?: Record[]; + apikeys?: Record[]; +} + +class FakePg implements WrappedPgClient { + readonly = false; + inserts: { table: string; vars: unknown[] }[] = []; + committed = false; + rolledBack = false; + #rows: Rows; + + constructor(rows: Rows = {}) { + this.#rows = rows; + } + + async destroy(): Promise {} + + async all(stmt: SqlCode): Promise { + const s = stmt._sql; + if (s.includes('FROM "users"')) return (this.#rows.users ?? []) as R[]; + if (s.includes('FROM "abodes"')) return (this.#rows.abodes ?? []) as R[]; + if (s.includes('FROM "residents"')) + return (this.#rows.residents ?? []) as R[]; + if (s.includes('FROM "apikeys"')) return (this.#rows.apikeys ?? []) as R[]; + return [] as R[]; + } + + async get(stmt: SqlCode): Promise { + const rows = await this.all(stmt); + return rows[0] ?? null; + } + + async run(stmt: SqlCode): Promise<{ changes: number }> { + const table = stmt._sql.match(/INSERT INTO "(\w+)"/)?.[1] ?? "?"; + this.inserts.push({ table, vars: stmt._vars }); + return { changes: 1 }; + } + + async multi(fn: (tx: WrappedPgClient) => Promise): Promise { + try { + const r = await fn(this); + this.committed = true; + return r; + } catch (e) { + this.rolledBack = true; + throw e; + } + } + + async rethrow(fn: () => Promise): Promise { + return fn(); + } +} + +const iso = "2026-01-02T03:04:05.000Z"; + +function seededRows(): Rows { + return { + users: [ + { + uid: "11111111-1111-1111-1111-111111111111", + email: "u1@test.example", + name: "User One", + flags: {}, + created_at: new Date(iso), + updated_at: new Date(iso), + }, + ], + abodes: [ + { + aid: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1", + name: "Abode One", + created_at: new Date(iso), + created_by: "11111111-1111-1111-1111-111111111111", + updated_at: new Date(iso), + updated_by: null, + }, + { + aid: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa2", + name: "Abode Two", + created_at: new Date(iso), + created_by: null, + updated_at: new Date(iso), + updated_by: null, + }, + ], + residents: [ + { + uid: "11111111-1111-1111-1111-111111111111", + aid: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1", + flags: {}, + created_at: new Date(iso), + created_by: null, + updated_at: new Date(iso), + updated_by: null, + }, + ], + apikeys: [ + { + uid: "11111111-1111-1111-1111-111111111111", + kid: "kkkkkkkk-kkkk-kkkk-kkkk-kkkkkkkkkkk1", + name: "key one", + permissions: {}, + created_at: new Date(iso), + expires_at: null, + }, + ], + }; +} + +async function streamToString(s: NodeJS.ReadableStream): Promise { + let out = ""; + for await (const chunk of s) out += chunk; + return out; +} +function parseLines(ndjson: string): { kind: string; data: any }[] { + return ndjson + .split("\n") + .filter(Boolean) + .map((l) => JSON.parse(l)); +} +function line(kind: string, data: unknown): string { + return JSON.stringify({ kind, data }) + "\n"; +} + +describe("postgres export", () => { + it("streams meta + the four supported kinds, never note", async () => { + const db = new PostgresInterface(new FakePg(seededRows())); + const lines = parseLines(await streamToString(db.export())); + + const meta = lines.find((l) => l.kind === "meta"); + assert.ok(meta); + assert.equal(meta!.data.source, "postgres"); + assert.equal(meta!.data.v, 1); + + const kinds = new Set(lines.map((l) => l.kind)); + assert.ok(kinds.has("user")); + assert.ok(kinds.has("abode")); + assert.ok(kinds.has("resident")); + assert.ok(kinds.has("apikey")); + assert.ok(!kinds.has("note"), "note is unsupported on postgres"); + }); + + it("applies the kinds filter", async () => { + const db = new PostgresInterface(new FakePg(seededRows())); + const lines = parseLines( + await streamToString(db.export({ filter: { kinds: ["abode"] } })), + ); + const kinds = new Set( + lines.filter((l) => l.kind !== "meta").map((l) => l.kind), + ); + assert.deepEqual(kinds, new Set(["abode"])); + }); + + it("applies the abodes allowlist to abode/resident records", async () => { + const db = new PostgresInterface(new FakePg(seededRows())); + const lines = parseLines( + await streamToString( + db.export({ + filter: { abodes: ["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1"] }, + }), + ), + ); + const abodeAids = lines + .filter((l) => l.kind === "abode") + .map((l) => l.data.aid); + assert.deepEqual(abodeAids, ["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1"]); + }); +}); + +describe("postgres import", () => { + it("dispatches inserts per kind, skips note, and commits", async () => { + const fake = new FakePg(); + const db = new PostgresInterface(fake); + const source = Readable.from([ + line("meta", { v: 1 }), + line("user", { + uid: "11111111-1111-1111-1111-111111111111", + email: "u1@test.example", + name: "User One", + flags: {}, + created_at: iso, + updated_at: iso, + }), + line("abode", { + aid: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1", + name: "Abode One", + created_at: iso, + created_by: "11111111-1111-1111-1111-111111111111", + updated_at: iso, + updated_by: null, + }), + line("apikey", { + uid: "11111111-1111-1111-1111-111111111111", + kid: "kkkkkkkk-kkkk-kkkk-kkkk-kkkkkkkkkkk1", + name: "key one", + permissions: {}, + created_at: iso, + expires_at: null, + }), + // note lines are silently skipped on postgres + line("note", { + nid: "nnnnnnnn-nnnn-nnnn-nnnn-nnnnnnnnnnn1", + aid: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1", + name: "Note", + content: "x", + properties: {}, + created_at: iso, + created_by: null, + updated_at: iso, + updated_by: null, + }), + ]); + + const result = await db.import(source); + assert.deepEqual(result.counts, { user: 1, abode: 1, apikey: 1 }); + assert.ok(fake.committed); + assert.deepEqual(fake.inserts.map((i) => i.table).sort(), [ + "abodes", + "apikeys", + "users", + ]); + assert.ok(!fake.inserts.some((i) => i.table === "notes"), "no note insert"); + // apikey gets a freshly-minted token (never exported) + const apikeyInsert = fake.inserts.find((i) => i.table === "apikeys")!; + assert.ok( + apikeyInsert.vars.some( + (v) => typeof v === "string" && v.startsWith("at_"), + ), + "apikey insert carries a fresh token", + ); + }); + + it("rejects and rolls back when the stream carries an error sentinel", async () => { + const fake = new FakePg(); + const db = new PostgresInterface(fake); + const source = Readable.from([ + line("meta", { v: 1 }), + line("user", { + uid: "11111111-1111-1111-1111-111111111111", + email: "u1@test.example", + name: "User One", + flags: {}, + created_at: iso, + updated_at: iso, + }), + line("error", { message: "boom" }), + ]); + + await assert.rejects(() => db.import(source), /boom/); + assert.ok(fake.rolledBack); + assert.ok(!fake.committed); + }); + + it("rejects and rolls back on signal abort", async () => { + const fake = new FakePg(); + const db = new PostgresInterface(fake); + const ac = new AbortController(); + const source = Readable.from( + (async function* () { + yield line("meta", { v: 1 }); + yield line("user", { + uid: "11111111-1111-1111-1111-111111111111", + email: "u1@test.example", + name: "User One", + flags: {}, + created_at: iso, + updated_at: iso, + }); + ac.abort(); + await new Promise((r) => setTimeout(r, 1000)); + })(), + ); + + await assert.rejects(() => db.import(source, { signal: ac.signal })); + assert.ok(fake.rolledBack); + assert.ok(!fake.committed); + }); +}); -- 2.54.0 From d7e31dfce9bbad8f96f263f1ebd6e9e976d9963c Mon Sep 17 00:00:00 2001 From: Codinget Date: Thu, 23 Jul 2026 00:14:15 +0000 Subject: [PATCH 4/4] docs: codify the FK-safe stream ordering as a wire-format invariant The postgres importer inserts records sequentially with FK enforcement live, so it depends on records arriving in dependency order. That requirement was implicit in each backend's export table list; make it explicit and enforced. - Add EXPORT_KIND_ORDER (user, abode, resident, apikey, note) as a documented single source of truth, with the FK dependency chain spelled out on its doc comment, and note the ordering guarantee on the ExportEnvelope wire-format doc. Derive the isExportKind set from it. - Both backends' export() now iterate EXPORT_KIND_ORDER via a loader map (postgres omits note by leaving it out of the map), so emission order is tied to the constant and can't drift. - Tests: a change-detector on EXPORT_KIND_ORDER, plus assertions that both the sqlite and postgres (fake) exports emit record kinds grouped in FK-safe order (kind rank non-decreasing down the stream, after the leading meta). Co-Authored-By: Claude Opus 4.8 --- src/db/export/filter.ts | 14 ++--- src/db/postgres/PostgresInterface.ts | 44 ++++++++------- src/db/sqlite/SqliteInterface.ts | 37 ++++++------ src/db/types/ExportImport.ts | 34 ++++++++++- test/backends/postgres/export-import.test.ts | 18 ++++++ test/tools/export-import.test.ts | 59 ++++++++++++++++++++ 6 files changed, 159 insertions(+), 47 deletions(-) diff --git a/src/db/export/filter.ts b/src/db/export/filter.ts index 23876bd..45bb10d 100644 --- a/src/db/export/filter.ts +++ b/src/db/export/filter.ts @@ -1,12 +1,10 @@ -import type { ExportFilter, ExportKind } from "../types/ExportImport.js"; +import { + EXPORT_KIND_ORDER, + type ExportFilter, + type ExportKind, +} from "../types/ExportImport.js"; -const EXPORT_KINDS = new Set([ - "user", - "abode", - "resident", - "apikey", - "note", -]); +const EXPORT_KINDS = new Set(EXPORT_KIND_ORDER); export function isExportKind(x: unknown): x is ExportKind { return typeof x === "string" && EXPORT_KINDS.has(x as ExportKind); diff --git a/src/db/postgres/PostgresInterface.ts b/src/db/postgres/PostgresInterface.ts index 3be560c..3ed448f 100644 --- a/src/db/postgres/PostgresInterface.ts +++ b/src/db/postgres/PostgresInterface.ts @@ -45,13 +45,14 @@ import type { import type { WrappedPgClient } from "./pool.js"; import { Readable } from "node:stream"; import readline from "node:readline"; -import type { - Exportable, - ExportKind, - ExportOptions, - Importable, - ImportOptions, - ImportResult, +import { + EXPORT_KIND_ORDER, + type Exportable, + type ExportKind, + type ExportOptions, + type Importable, + type ImportOptions, + type ImportResult, } from "../types/ExportImport.js"; import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js"; @@ -527,18 +528,19 @@ export class PostgresInterface const db = this.#db; const source = this.name; - // `note` is omitted: the postgres backend has no note CRUD yet, so a pg - // database can hold none. Each `load()` is a single query, run lazily and - // skipped once the destination aborts. - const tables: [ - ExportKind, - () => Promise<{ uid?: string; aid?: string }[]>, - ][] = [ - ["user", () => selectClientUsers(db)], - ["abode", () => selectAbodes(db)], - ["resident", () => selectResidents(db)], - ["apikey", () => selectClientApikeys(db)], - ]; + // Emission follows the FK-safe EXPORT_KIND_ORDER (part of the wire + // contract; see ExportImport.ts). `note` has no loader: the postgres + // backend has no note CRUD yet, so a pg database can hold none. Each + // `load()` is a single query, run lazily and skipped once the destination + // aborts. + const loaders: Partial< + Record Promise<{ uid?: string; aid?: string }[]>> + > = { + user: () => selectClientUsers(db), + abode: () => selectAbodes(db), + resident: () => selectResidents(db), + apikey: () => selectClientApikeys(db), + }; async function* generate(): AsyncGenerator { if (signal?.aborted) return; @@ -552,8 +554,10 @@ export class PostgresInterface }, }) + "\n"; try { - for (const [kind, load] of tables) { + for (const kind of EXPORT_KIND_ORDER) { if (signal?.aborted) return; + const load = loaders[kind]; + if (!load) continue; if (!kindAllowed(filter, kind)) continue; for (const row of await load()) { if (signal?.aborted) return; diff --git a/src/db/sqlite/SqliteInterface.ts b/src/db/sqlite/SqliteInterface.ts index 03c96fa..c827a04 100644 --- a/src/db/sqlite/SqliteInterface.ts +++ b/src/db/sqlite/SqliteInterface.ts @@ -48,13 +48,14 @@ import type { import type { WrappedDb } from "./impl/types.js"; import { Readable } from "node:stream"; import readline from "node:readline"; -import type { - Exportable, - ExportKind, - ExportOptions, - Importable, - ImportOptions, - ImportResult, +import { + EXPORT_KIND_ORDER, + type Exportable, + type ExportKind, + type ExportOptions, + type Importable, + type ImportOptions, + type ImportResult, } from "../types/ExportImport.js"; import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js"; @@ -558,14 +559,16 @@ export class SqliteInterface // `load()` is exactly one `WrappedDb.all()`). Kept lazy so the first query // only fires once the destination starts pulling, and skipped entirely // once the signal is aborted — no further reads after the destination - // goes away. - const tables: [ExportKind, () => { uid?: string; aid?: string }[]][] = [ - ["user", () => selectClientUsers(db)], - ["abode", () => selectAbodes(db)], - ["resident", () => selectResidents(db)], - ["apikey", () => selectClientApikeys(db)], - ["note", () => selectNotes(db)], - ]; + // goes away. Emission follows the FK-safe EXPORT_KIND_ORDER (part of the + // wire contract; see ExportImport.ts). + const loaders: Record { uid?: string; aid?: string }[]> = + { + user: () => selectClientUsers(db), + abode: () => selectAbodes(db), + resident: () => selectResidents(db), + apikey: () => selectClientApikeys(db), + note: () => selectNotes(db), + }; async function* generate(): AsyncGenerator { if (signal?.aborted) return; @@ -579,10 +582,10 @@ export class SqliteInterface }, }) + "\n"; try { - for (const [kind, load] of tables) { + for (const kind of EXPORT_KIND_ORDER) { if (signal?.aborted) return; if (!kindAllowed(filter, kind)) continue; - for (const row of load()) { + for (const row of loaders[kind]()) { if (signal?.aborted) return; if (recordAllowed(filter, kind, row)) { yield JSON.stringify({ kind, data: row }) + "\n"; diff --git a/src/db/types/ExportImport.ts b/src/db/types/ExportImport.ts index 812defe..dbee43a 100644 --- a/src/db/types/ExportImport.ts +++ b/src/db/types/ExportImport.ts @@ -8,6 +8,34 @@ import type { DbInterface } from "./DbInterface.js"; */ export type ExportKind = "user" | "abode" | "resident" | "apikey" | "note"; +/** + * Canonical order in which record kinds are emitted into an export stream, and + * the order in which an importer may safely apply them. + * + * This ordering is **part of the wire contract**, not an implementation + * detail. It is FK-safe: every foreign key points only at a kind that appears + * earlier (or at the same kind, earlier in the stream), so an importer that + * inserts records one-by-one with referential integrity enforced never + * forward-references a row it hasn't inserted yet: + * + * - `abode.created_by`/`updated_by` → `user` + * - `resident.uid` → `user`, `resident.aid` → `abode` + * - `apikey.uid` → `user` + * - `note.aid` → `abode`, `note.created_by`/`updated_by` → `user` + * + * The sqlite backend can afford to relax this (it suspends FK enforcement for + * the load), but the postgres backend relies on it: it inserts sequentially + * with constraints live. Every {@link Exportable} MUST emit in this order, and + * reordering it is a breaking change to the format. + */ +export const EXPORT_KIND_ORDER = [ + "user", + "abode", + "resident", + "apikey", + "note", +] as const satisfies readonly ExportKind[]; + export type ExportFilter = { /** Include only these kinds; omit = all kinds. */ kinds?: ExportKind[]; @@ -55,8 +83,10 @@ export interface Importable { /** * The NDJSON envelope written/read for every line. The leading line is a - * `meta` record; a trailing `error` record may appear if the source failed - * after streaming had already begun. + * `meta` record; the record lines that follow are grouped by kind in + * {@link EXPORT_KIND_ORDER} (an FK-safe order importers may rely on); a + * trailing `error` record may appear if the source failed after streaming had + * already begun. */ export type ExportEnvelope = | { kind: "meta"; data: ExportMeta } diff --git a/test/backends/postgres/export-import.test.ts b/test/backends/postgres/export-import.test.ts index 3d1b789..214386f 100644 --- a/test/backends/postgres/export-import.test.ts +++ b/test/backends/postgres/export-import.test.ts @@ -4,6 +4,10 @@ import { Readable } from "node:stream"; import { PostgresInterface } from "../../../src/db/postgres/PostgresInterface.js"; import type { WrappedPgClient } from "../../../src/db/postgres/pool.js"; import type { SqlCode } from "../../../src/db/postgres/sql.js"; +import { + EXPORT_KIND_ORDER, + type ExportKind, +} from "../../../src/db/types/ExportImport.js"; // The postgres backend has no CI database, so these tests drive the real // PostgresInterface.export/import code paths against an in-memory fake client. @@ -158,6 +162,20 @@ describe("postgres export", () => { assert.ok(!kinds.has("note"), "note is unsupported on postgres"); }); + it("emits record kinds grouped in FK-safe EXPORT_KIND_ORDER", async () => { + const db = new PostgresInterface(new FakePg(seededRows())); + const lines = parseLines(await streamToString(db.export())); + assert.equal(lines[0]?.kind, "meta", "first line is meta"); + const rank = (k: string) => EXPORT_KIND_ORDER.indexOf(k as ExportKind); + let last = -1; + for (const { kind } of lines.slice(1)) { + const r = rank(kind); + assert.notEqual(r, -1, `unexpected kind ${kind}`); + assert.ok(r >= last, `kind ${kind} out of FK-safe order`); + last = r; + } + }); + it("applies the kinds filter", async () => { const db = new PostgresInterface(new FakePg(seededRows())); const lines = parseLines( diff --git a/test/tools/export-import.test.ts b/test/tools/export-import.test.ts index c8bdaca..aee0b91 100644 --- a/test/tools/export-import.test.ts +++ b/test/tools/export-import.test.ts @@ -16,6 +16,31 @@ import { import { inspectExportStream } from "../../src/db/export/inspect.js"; import { hashPassword } from "../../src/util/hash.js"; import type { ClientUser } from "../../src/db/types/User.js"; +import { + EXPORT_KIND_ORDER, + type ExportKind, +} from "../../src/db/types/ExportImport.js"; + +/** + * Assert the record lines of a parsed export are grouped in FK-safe + * EXPORT_KIND_ORDER: the leading line is `meta`, and every record kind's + * position in the order is non-decreasing down the stream. + */ +function assertFkSafeOrder(lines: { kind: string }[]): void { + assert.equal(lines[0]?.kind, "meta", "first line is meta"); + const rank = (k: string) => EXPORT_KIND_ORDER.indexOf(k as ExportKind); + let last = -1; + for (const { kind } of lines.slice(1)) { + if (kind === "error") continue; + const r = rank(kind); + assert.notEqual(r, -1, `unexpected kind ${kind}`); + assert.ok( + r >= last, + `kind ${kind} (order ${r}) appears after a later kind (order ${last})`, + ); + last = r; + } +} // --------------------------------------------------------------------------- // helpers @@ -144,6 +169,40 @@ async function seed(db: TestDb["db"]): Promise { }; } +// --------------------------------------------------------------------------- +// 0. wire-format ordering invariant +// --------------------------------------------------------------------------- + +describe("export ordering (FK-safe wire contract)", () => { + it("EXPORT_KIND_ORDER is the FK-safe dependency order", () => { + // Change-detector: reordering is a breaking change to the format and must + // keep every kind after the kinds it references (see ExportImport.ts). + assert.deepEqual(EXPORT_KIND_ORDER, [ + "user", + "abode", + "resident", + "apikey", + "note", + ]); + }); + + it("sqlite export emits record kinds grouped in EXPORT_KIND_ORDER", async () => { + const src = await createTestDb(); + try { + await seed(src.db); + const lines = parseLines(await streamToString(src.db.export())); + // every kind must be present so the ordering is actually exercised + const kinds = new Set(lines.map((l) => l.kind)); + for (const k of EXPORT_KIND_ORDER) { + assert.ok(kinds.has(k), `stream contains ${k}`); + } + assertFkSafeOrder(lines); + } finally { + src.close(); + } + }); +}); + // --------------------------------------------------------------------------- // 1. round-trip // --------------------------------------------------------------------------- -- 2.54.0