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 <noreply@anthropic.com>
This commit is contained in:
+7
-1
@@ -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)",
|
||||
|
||||
@@ -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 <database-url> [--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<string, string>();
|
||||
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 <database-url>");
|
||||
|
||||
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);
|
||||
@@ -0,0 +1,105 @@
|
||||
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 <sqlite-database-url> <input-file|-> [--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<string, string>();
|
||||
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 <sqlite-database-url>");
|
||||
if (!input) printUsage("missing <input-file|->");
|
||||
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);
|
||||
@@ -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 <input-file|-> [--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<string, string>();
|
||||
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 <input-file|->");
|
||||
|
||||
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);
|
||||
@@ -25,8 +25,14 @@ 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<string, string>;
|
||||
#readonly: boolean;
|
||||
@@ -296,4 +302,44 @@ export class ApiInterface implements DbInterface {
|
||||
async listNotesByUserId(uid: string): Promise<PartialNote[]> {
|
||||
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<Buffer | Uint8Array> {
|
||||
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<Uint8Array>);
|
||||
}
|
||||
|
||||
return Readable.from(generate());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import type { ExportFilter, ExportKind } from "../types/ExportImport.js";
|
||||
|
||||
const EXPORT_KINDS = new Set<ExportKind>([
|
||||
"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<T extends string>(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<T extends string>(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),
|
||||
};
|
||||
}
|
||||
@@ -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<Record<ExportKind, number>>;
|
||||
meta?: Partial<ExportMeta> & { 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<InspectResult> {
|
||||
const { signal, stopAfterKinds } = options;
|
||||
const counts: Partial<Record<ExportKind, number>> = {};
|
||||
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 };
|
||||
}
|
||||
@@ -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,223 @@ export class SqliteInterface implements BackendDbInterface {
|
||||
async listNotesByUserId(uid: string): Promise<PartialNote[]> {
|
||||
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<string> {
|
||||
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<ImportResult> {
|
||||
this.#checkReadonly();
|
||||
const { filter, signal } = options;
|
||||
const db = this.#db;
|
||||
const counts: Partial<Record<ExportKind, number>> = {};
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,10 +118,10 @@ export function selectClientApikey(
|
||||
}
|
||||
export function selectClientApikeys(
|
||||
db: WrappedDb,
|
||||
where: SqlCode
|
||||
where?: SqlCode
|
||||
): ClientApikey[] {
|
||||
const rawApikeys = db.all<RawClientApikey>(
|
||||
sql`${sqlClientApikey} WHERE ${where}`
|
||||
where ? sql`${sqlClientApikey} WHERE ${where}` : sqlClientApikey
|
||||
);
|
||||
return rawApikeys.map(sqliteToClientApikey);
|
||||
}
|
||||
|
||||
@@ -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<Record<ExportKind, number>>;
|
||||
}
|
||||
|
||||
export interface Importable {
|
||||
import(
|
||||
source: NodeJS.ReadableStream,
|
||||
options?: ImportOptions
|
||||
): Promise<ImportResult>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<Exportable>).export === "function";
|
||||
}
|
||||
@@ -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<string, unknown>): 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();
|
||||
|
||||
@@ -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<Context["session"]> }
|
||||
): Promise<ExportFilter | null> {
|
||||
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<string>([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 };
|
||||
}
|
||||
@@ -0,0 +1,672 @@
|
||||
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<string> {
|
||||
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<string, unknown>): Record<string, unknown> {
|
||||
const out: Record<string, unknown> = {};
|
||||
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<string, unknown>[],
|
||||
key: (x: any) => string
|
||||
): Record<string, unknown>[] {
|
||||
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<Seed> {
|
||||
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<void>;
|
||||
|
||||
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<void>((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<void>((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<typeof createServer> | 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<void>((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<void>((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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user