add export/import streaming to the pluggable backends #13

Merged
codinget merged 4 commits from feat/export-import-streaming into master 2026-07-23 20:43:15 +02:00
6 changed files with 159 additions and 47 deletions
Showing only changes of commit d7e31dfce9 - Show all commits
+6 -8
View File
@@ -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<ExportKind>([ const EXPORT_KINDS = new Set<ExportKind>(EXPORT_KIND_ORDER);
"user",
"abode",
"resident",
"apikey",
"note",
]);
export function isExportKind(x: unknown): x is ExportKind { export function isExportKind(x: unknown): x is ExportKind {
return typeof x === "string" && EXPORT_KINDS.has(x as ExportKind); return typeof x === "string" && EXPORT_KINDS.has(x as ExportKind);
+24 -20
View File
@@ -45,13 +45,14 @@ import type {
import type { WrappedPgClient } from "./pool.js"; import type { WrappedPgClient } from "./pool.js";
import { Readable } from "node:stream"; import { Readable } from "node:stream";
import readline from "node:readline"; import readline from "node:readline";
import type { import {
Exportable, EXPORT_KIND_ORDER,
ExportKind, type Exportable,
ExportOptions, type ExportKind,
Importable, type ExportOptions,
ImportOptions, type Importable,
ImportResult, type ImportOptions,
type ImportResult,
} from "../types/ExportImport.js"; } from "../types/ExportImport.js";
import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js"; import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.js";
@@ -527,18 +528,19 @@ export class PostgresInterface
const db = this.#db; const db = this.#db;
const source = this.name; const source = this.name;
// `note` is omitted: the postgres backend has no note CRUD yet, so a pg // Emission follows the FK-safe EXPORT_KIND_ORDER (part of the wire
// database can hold none. Each `load()` is a single query, run lazily and // contract; see ExportImport.ts). `note` has no loader: the postgres
// skipped once the destination aborts. // backend has no note CRUD yet, so a pg database can hold none. Each
const tables: [ // `load()` is a single query, run lazily and skipped once the destination
ExportKind, // aborts.
() => Promise<{ uid?: string; aid?: string }[]>, const loaders: Partial<
][] = [ Record<ExportKind, () => Promise<{ uid?: string; aid?: string }[]>>
["user", () => selectClientUsers(db)], > = {
["abode", () => selectAbodes(db)], user: () => selectClientUsers(db),
["resident", () => selectResidents(db)], abode: () => selectAbodes(db),
["apikey", () => selectClientApikeys(db)], resident: () => selectResidents(db),
]; apikey: () => selectClientApikeys(db),
};
async function* generate(): AsyncGenerator<string> { async function* generate(): AsyncGenerator<string> {
if (signal?.aborted) return; if (signal?.aborted) return;
@@ -552,8 +554,10 @@ export class PostgresInterface
}, },
}) + "\n"; }) + "\n";
try { try {
for (const [kind, load] of tables) { for (const kind of EXPORT_KIND_ORDER) {
if (signal?.aborted) return; if (signal?.aborted) return;
const load = loaders[kind];
if (!load) continue;
if (!kindAllowed(filter, kind)) continue; if (!kindAllowed(filter, kind)) continue;
for (const row of await load()) { for (const row of await load()) {
if (signal?.aborted) return; if (signal?.aborted) return;
+20 -17
View File
@@ -48,13 +48,14 @@ import type {
import type { WrappedDb } from "./impl/types.js"; import type { WrappedDb } from "./impl/types.js";
import { Readable } from "node:stream"; import { Readable } from "node:stream";
import readline from "node:readline"; import readline from "node:readline";
import type { import {
Exportable, EXPORT_KIND_ORDER,
ExportKind, type Exportable,
ExportOptions, type ExportKind,
Importable, type ExportOptions,
ImportOptions, type Importable,
ImportResult, type ImportOptions,
type ImportResult,
} from "../types/ExportImport.js"; } from "../types/ExportImport.js";
import { isExportKind, kindAllowed, recordAllowed } from "../export/filter.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 // `load()` is exactly one `WrappedDb.all()`). Kept lazy so the first query
// only fires once the destination starts pulling, and skipped entirely // only fires once the destination starts pulling, and skipped entirely
// once the signal is aborted — no further reads after the destination // once the signal is aborted — no further reads after the destination
// goes away. // goes away. Emission follows the FK-safe EXPORT_KIND_ORDER (part of the
const tables: [ExportKind, () => { uid?: string; aid?: string }[]][] = [ // wire contract; see ExportImport.ts).
["user", () => selectClientUsers(db)], const loaders: Record<ExportKind, () => { uid?: string; aid?: string }[]> =
["abode", () => selectAbodes(db)], {
["resident", () => selectResidents(db)], user: () => selectClientUsers(db),
["apikey", () => selectClientApikeys(db)], abode: () => selectAbodes(db),
["note", () => selectNotes(db)], resident: () => selectResidents(db),
]; apikey: () => selectClientApikeys(db),
note: () => selectNotes(db),
};
async function* generate(): AsyncGenerator<string> { async function* generate(): AsyncGenerator<string> {
if (signal?.aborted) return; if (signal?.aborted) return;
@@ -579,10 +582,10 @@ export class SqliteInterface
}, },
}) + "\n"; }) + "\n";
try { try {
for (const [kind, load] of tables) { for (const kind of EXPORT_KIND_ORDER) {
if (signal?.aborted) return; if (signal?.aborted) return;
if (!kindAllowed(filter, kind)) continue; if (!kindAllowed(filter, kind)) continue;
for (const row of load()) { for (const row of loaders[kind]()) {
if (signal?.aborted) return; if (signal?.aborted) return;
if (recordAllowed(filter, kind, row)) { if (recordAllowed(filter, kind, row)) {
yield JSON.stringify({ kind, data: row }) + "\n"; yield JSON.stringify({ kind, data: row }) + "\n";
+32 -2
View File
@@ -8,6 +8,34 @@ import type { DbInterface } from "./DbInterface.js";
*/ */
export type ExportKind = "user" | "abode" | "resident" | "apikey" | "note"; 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 = { export type ExportFilter = {
/** Include only these kinds; omit = all kinds. */ /** Include only these kinds; omit = all kinds. */
kinds?: ExportKind[]; kinds?: ExportKind[];
@@ -55,8 +83,10 @@ export interface Importable {
/** /**
* The NDJSON envelope written/read for every line. The leading line is a * 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 * `meta` record; the record lines that follow are grouped by kind in
* after streaming had already begun. * {@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 = export type ExportEnvelope =
| { kind: "meta"; data: ExportMeta } | { kind: "meta"; data: ExportMeta }
@@ -4,6 +4,10 @@ import { Readable } from "node:stream";
import { PostgresInterface } from "../../../src/db/postgres/PostgresInterface.js"; import { PostgresInterface } from "../../../src/db/postgres/PostgresInterface.js";
import type { WrappedPgClient } from "../../../src/db/postgres/pool.js"; import type { WrappedPgClient } from "../../../src/db/postgres/pool.js";
import type { SqlCode } from "../../../src/db/postgres/sql.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 // The postgres backend has no CI database, so these tests drive the real
// PostgresInterface.export/import code paths against an in-memory fake client. // 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"); 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 () => { it("applies the kinds filter", async () => {
const db = new PostgresInterface(new FakePg(seededRows())); const db = new PostgresInterface(new FakePg(seededRows()));
const lines = parseLines( const lines = parseLines(
+59
View File
@@ -16,6 +16,31 @@ import {
import { inspectExportStream } from "../../src/db/export/inspect.js"; import { inspectExportStream } from "../../src/db/export/inspect.js";
import { hashPassword } from "../../src/util/hash.js"; import { hashPassword } from "../../src/util/hash.js";
import type { ClientUser } from "../../src/db/types/User.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 // helpers
@@ -144,6 +169,40 @@ async function seed(db: TestDb["db"]): Promise<Seed> {
}; };
} }
// ---------------------------------------------------------------------------
// 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 // 1. round-trip
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------