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>

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<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;
+      }
+    }
+  }
 }