#!/usr/bin/env node
/**
* orgTransfer.js — Export / Import / Netzwerk-Kopie genau EINER Organisation
* aus der members-MariaDB.
*
* Problem: Die members-Datenbank ist mandantenfähig, aber nicht über eine
* OrgUID-Spalte getrennt, sondern als Objekt-Graph. Eine Organisation ist ein
* `group`-Objekt mit `Data.root = true`, an dem alle ihre Objekte über
* "Propagations-Links" hängen (member, memberA, memberS, memberG, memberSys,
* member0, memberGA). Ein mysqldump der ganzen DB ist für einen Umzug einer
* einzelnen Organisation also unbrauchbar.
*
* Dieses Skript berechnet den Objekt-Scope einer Organisation rekursiv,
* exportiert die Kern-Tabellen als einspielbares SQL und kann das Ergebnis
* wahlweise in eine Datei schreiben, direkt in eine andere Datenbank
* streamen (Netzwerk-Kopie) oder aus einer Datei wieder einspielen.
*
* node src/scripts/orgTransfer.js inspect --org <UUID>
* node src/scripts/orgTransfer.js export --org <UUID> [Optionen]
* node src/scripts/orgTransfer.js import --in <DATEI> [Optionen]
* node src/scripts/orgTransfer.js copy --org <UUID> --to-db <DB> [Optionen]
*
* Als `--org` darf jede Objekt-UID übergeben werden: ist es nicht selbst die
* Organisation, wird die zugehörige Organisation über die Propagations-Links
* aufgelöst (z.B. eine Person -> ihre Organisation).
*
* UUIDs werden über die DB-Funktionen `UUID2BIN()` / `BIN2UUID()` konvertiert
* (dieselben, die auch das Backend benutzt). Die Ziel-Datenbank muss diese
* Funktionen besitzen — sie kommen aus `initFunctions.sql`.
*
* Siehe `--help` für alle Optionen.
*
* @module scripts/orgTransfer
*/
import fs from 'node:fs';
import fsp from 'node:fs/promises';
import path from 'node:path';
import zlib from 'node:zlib';
import { pipeline } from 'node:stream/promises';
import { loadSecretsFromVault } from '@commtool/vault-secrets';
import { getConnection } from '@commtool/sql-query';
// ---------------------------------------------------------------------------
// 1. Scope-Definition
// ---------------------------------------------------------------------------
/**
* Link-Typen, über die Zugehörigkeit zur Organisation propagiert wird.
*
* Dies sind exakt die Typen aus `getOrganizationForObject()` im Backend
* (`src/utils/organizationUtils.js`) — dort wird dieselbe Semantik genutzt,
* um zu einem Objekt seine Organisation zu finden. Umgekehrt aufgezogen
* ergeben sie den vollständigen Objekt-Bestand einer Organisation.
*/
const PROPAGATION_LINK_TYPES = [
'member',
'memberSys',
'memberA',
'memberS',
'memberG',
'member0',
'memberGA',
];
/**
* Zusätzliche, explizit anzufordernde Scope-Erweiterungen.
*
* Der Propagations-Scope enthält nur die "eigenen" Objekte der Organisation.
* Bestimmte Objekte werden von ihnen nur *referenziert* und fehlen dadurch —
* ohne sie funktioniert aber z.B. eine dynamische Liste nicht:
*
* - `filters` : `include`/`exclude`/`intersect`-Objekte einer `dlist`.
* Sie sind das ZIEL eines `dynamic`-Links, werden also nur
* über eine Vorwärts-Verfolgung gefunden. Im Test fehlten
* 129 Filter-Objekte bei ~36k Referenzen.
* - `templates` : Vorlagen/Kataloge (`achievementT`, `eventT`, `actionT`,
* `function`, `eventJobT`, ...) — in beide Richtungen
* verlinkt.
*
* `outgoing` folgt den Links von Scope-Objekten weg (referenzierte Ziele),
* `incoming` folgt Links in den Scope hinein. Beides läuft bis zum Fixpunkt,
* Fremd-Organisationen werden dabei nicht hineingezogen.
*/
const EXTRA_SCOPES = {
filters: { outgoing: ['dynamic'] },
templates: {
incoming: ['achievement', 'event', 'action', 'function', 'eventJob', 'eventFunction'],
outgoing: ['achievement', 'event', 'action', 'function', 'eventJob', 'eventFunction'],
},
};
/**
* Zugehörigkeit über `UIDBelongsTo` wird **transitiv** verfolgt.
*
* `UIDBelongsTo` ist laut Schema Inheritance/Ownership — die Aussage „gehört
* diesem Objekt". Sie ist damit ein Graph-Rand wie ein Link, nur dass der
* Export ihm bisher gar nicht folgte. Zwei Fälle fallen dadurch heraus:
*
* 1. **Eltern-Objekte statt Organisation.** `include`/`exclude`/`intersect`
* hängen an ihrem `list`/`group`-Elternobjekt, nicht an der Organisation
* (gemessen: 155 an `list`, 142 an `group`). `action`/`eventJobT` hängen an
* `eventT`. Ein Prädikat `UIDBelongsTo = <org>` sieht davon nichts.
* 2. **Tiefer als eine Ebene.** Gemessen in der Dev-DB: 79 Objekte auf Tiefe 2
* (67 `eventJobT` + 12 `action`, alle an einem `eventT`), verteilt auf eine
* Organisation. `--include filters,templates` hilft dort nicht, weil der
* `templates`-Scope *Links* folgt und diese Objekte gar keine tragen.
*
* Deshalb wird bis zum Fixpunkt wiederholt, nicht einmalig. Ein einzelner
* `INSERT ... JOIN __scope` pro Runde, Abbruch bei `affectedRows = 0`; die
* Obergrenze ist nur ein Sicherheitsnetz gegen einen Zyklus, der durch
* `INSERT IGNORE` ohnehin nicht entstehen kann.
*
* **Guard:** Sub-Organisationen (`Data.root = true`) werden nicht hineingezogen —
* die sind eigenständige Organisationen mit eigenem Export. Dieselbe Regel
* benutzt die Link-Closure unten.
*/
const ORG_OWNED_MAX_ROUNDS = 12;
/**
* Tabellen, die exportiert werden können.
*
* `scope`:
* 'org' — über die Objekt-UID (`__scope`)
* 'member' — Member-Zeilen, die von Objekten des Scopes benutzt werden
* 'accounting' — über `Transactions.UIDOrganization`
* 'events' — über `eventLog.UIDorga`
*
* `always`: gehört zum Kern-Export (Default-Tabellensatz).
*/
const TABLE_SPECS = [
{
name: 'Member',
scope: 'member',
always: true,
order: 10,
// Member ist die "auslagerbare Personendaten"-Tabelle: `ObjectBase.UIDBelongsTo`
// zeigt auf `Member.UID` (bei Personen ist das die eigene UID, sonst die
// gemeinsame Zeile mehrerer Objekte). Gebraucht werden daher beide Mengen —
// sie liegen im Scope-Aufbau in `__member_scope`.
},
{
name: 'ObjectBase',
scope: 'org',
always: true,
order: 20,
},
{
name: 'Links',
scope: 'org',
always: true,
order: 30,
},
{
name: 'Visible',
scope: 'org',
always: true,
order: 40,
},
{
name: 'SearchIndex',
scope: 'org',
order: 50,
},
{
name: 'Transactions',
scope: 'accounting',
order: 60,
// Nicht über den Objekt-Graph erreichbar, sondern über die Org-Spalte.
where: (ctx) => `UIDOrganization = UUID2BIN(${ctx.q(ctx.org.uid)})`,
},
{
name: 'TransactionLines',
scope: 'accounting',
order: 70,
where: (ctx) =>
`UIDTransaction IN (SELECT UID FROM Transactions WHERE UIDOrganization = UUID2BIN(${ctx.q(ctx.org.uid)}))`,
},
{
name: 'eventLog',
scope: 'events',
order: 80,
where: (ctx) => `UIDorga = ${ctx.q('UUID-' + ctx.org.uid)}`,
},
];
/** Optionale Scope-Flags -> Tabellen, die sie mitbringen. */
const SCOPE_TABLES = {
accounting: ['Transactions', 'TransactionLines'],
events: ['eventLog'],
};
/** Datentypen, die als nackte Zahl ausgegeben werden. */
const NUMERIC_TYPES = new Set([
'tinyint', 'smallint', 'mediumint', 'int', 'integer', 'bigint',
'decimal', 'numeric', 'float', 'double', 'real', 'year',
]);
/** Datentypen, die als String-Literal ausgegeben werden. */
const STRING_TYPES = new Set([
'char', 'varchar', 'tinytext', 'text', 'mediumtext', 'longtext',
'enum', 'set', 'date', 'datetime', 'timestamp', 'time', 'json',
]);
// ---------------------------------------------------------------------------
// 2. Kleine Helfer
// ---------------------------------------------------------------------------
const log = (...a) => process.stderr.write(a.join(' ') + '\n');
const warn = (...a) => process.stderr.write('! ' + a.join(' ') + '\n');
/** Fortschrittszeilen nur auf einem echten Terminal ausgeben (\r in Pipes ist Müll). */
const wantProgress = (opts) => !opts.quiet && process.stderr.isTTY === true;
/** Sekunden mit einer Nachkommastelle. */
const secs = (ms) => (ms / 1000).toFixed(1) + 's';
/** Bytes menschenlesbar. */
function human(bytes) {
const u = ['B', 'KiB', 'MiB', 'GiB'];
let i = 0;
let v = Number(bytes);
while (v >= 1024 && i < u.length - 1) { v /= 1024; i++; }
return v.toFixed(i ? 1 : 0) + ' ' + u[i];
}
const UUID_RE = /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/;
/** Nur echte Nutzdaten durchlassen (Schutz gegen kaputte Treiber-Werte). */
const NUMBER_RE = /^-?\d+(\.\d+)?([eE][-+]?\d+)?$/;
/** SQL-String-Literal mit MySQL-Backslash-Escapes (wie mysqldump). */
function quote(value) {
const s = String(value);
let out = "'";
for (let i = 0; i < s.length; i++) {
const c = s[i];
switch (c) {
case '\\': out += '\\\\'; break;
case "'": out += "\\'"; break;
case '\n': out += '\\n'; break;
case '\r': out += '\\r'; break;
case '\0': out += '\\0'; break;
case '\x1a': out += '\\Z'; break;
default: out += c;
}
}
return out + "'";
}
/** Bezeichner quoten. */
const id = (name) => '`' + String(name).replace(/`/g, '``') + '`';
// ---------------------------------------------------------------------------
// 3. CLI
// ---------------------------------------------------------------------------
const USAGE = `
orgTransfer.js — eine Organisation aus der members-DB exportieren / importieren
inspect --org <UUID> Scope nur analysieren, nichts schreiben
export --org <UUID> [Opt.] Scope als einspielbares SQL schreiben
import --in <DATEI> [Opt.] SQL-Dump in eine Datenbank einspielen
copy --org <UUID> --to-db <DB> [Opt.] direkt DB -> DB streamen
Quelle (export/copy) — normalerweise aus Vault/ENV:
--from-db <DB> --from-host <H> --from-user <U> --from-pass <P>
Ziel (import/copy) — sonst wie Quelle:
--to-db <DB> --to-host <H> --to-user <U> --to-pass <P>
Credentials ohne Kommandozeilen-Flags (ENV hat Vorrang vor Vault):
DB_HOST/DB_USER/DB_PASS/DB_DATABASE gilt für BEIDE Verbindungen
DB_TO_HOST/DB_TO_USER/DB_TO_PASS/DB_TO_DATABASE nur fürs Ziel
Zwei-Host-Kopie, Quelle komplett aus Vault:
docker exec -e DB_TO_HOST=ziel-h -e DB_TO_USER=root \\
-e DB_TO_PASS=geheim -e DB_TO_DATABASE=member <ctr> \\
node src/scripts/orgTransfer.js copy --org <UUID>
Optionen export/copy:
--out <DATEI> Zieldatei (Default: org-<uid8>-<zeit>.sql)
--gzip Datei gzip-komprimiert schreiben (.sql.gz)
--include <LISTE> Zusatz-Scopes: filters,templates,accounting,events
--tables <LISTE> expliziter Tabellensatz (überschreibt --include)
--include-history System-Versionierung mitnehmen (FOR SYSTEM_TIME ALL)
--batch <N> Zeilen pro INSERT (Default 200)
--max-statement <KiB> INSERT-Batch spätestens ab dieser Größe flushen (Default 1024 = 1 MiB)
--limit <N> nur N Zeilen pro Tabelle (für Tests)
--no-batch eine INSERT-Zeile pro Datensatz (sehr groß, nur für Diffs)
Optionen import:
--mode <M> abort | skip | replace | overwrite (Default: abort)
--dry-run Datei nur parsen und zählen, nichts schreiben
--in - von stdin lesen
abort Abbruch, wenn die Organisation im Ziel schon existiert
skip nichts tun, wenn sie schon existiert
replace vorhandene Organisation (z. B. der Stub, den das Backend beim
Start aus den Config-Dateien anlegt) vorher entfernen, dann
einspielen — der Weg für ein Ziel, das nicht leer ist
overwrite trotzdem einspielen (neue Versionen; scheitert an PKs wie
Member.UID / Visible.UID+UIDUser)
Allgemein:
--dry-run export/copy: nur Plan + Scope-Report ausgeben
--quiet weniger Fortschritt
-h, --help diese Hilfe
Der Objekt-Scope umfasst zwei Arten von Zugehörigkeit:
1. Propagations-Links (member, memberSys, memberA, memberS, memberG,
member0, memberGA) — der mitgliedschaftliche Graph.
2. UIDBelongsTo — transitiv bis zum Fixpunkt. Das sind die org-eigenen
Objekte OHNE Link (changeable-Filter, actionT/eventJobT-Vorlagen,
function, achievementT) sowie Objekte, die an einem Elternobjekt des
Scopes hängen (include/exclude/intersect an list/group, eventJobT und
action an eventT).
NICHT enthalten sind Objekte, die nur über sonstige Link-Typen referenziert
werden (family, familyFees, dynamic, ...) — \`inspect\` listet sie als
"Links, die den Scope verlassen" auf.
`;
/** Minimaler Argument-Parser (`--key wert`, `--flag`, `--key=wert`). */
function parseArgs(argv) {
const out = { _: [] };
for (let i = 0; i < argv.length; i++) {
const a = argv[i];
if (a === '-h') { out.help = true; continue; }
if (!a.startsWith('--')) { out._.push(a); continue; }
const eq = a.indexOf('=');
if (eq !== -1) {
out[a.slice(2, eq)] = a.slice(eq + 1);
continue;
}
const key = a.slice(2);
const next = argv[i + 1];
if (next === undefined || next.startsWith('--')) {
out[key] = true;
} else {
out[key] = next;
i++;
}
}
return out;
}
const toInt = (v, dflt) => {
if (v === undefined || v === true) return dflt;
const n = parseInt(String(v), 10);
return Number.isFinite(n) ? n : dflt;
};
// ---------------------------------------------------------------------------
// 4. Datenbank-Zugriff
// ---------------------------------------------------------------------------
/**
* Verbindung öffnen.
*
* `dateStrings: true` ist hier wichtig: ohne das castet der mariadb-Treiber
* DATETIME/TIMESTAMP-Werte in JS-`Date`-Objekte und wendet dabei die lokale
* Zeitzone an. Für einen verlustfreien Dump brauchen wir den Roh-String des
* Servers ('2026-06-29 13:08:26.234419').
*
* @param {object} [opts] host/user/password/database-Overrides
* @returns {Promise<any>} mariadb-Verbindung (nicht gepoolt)
*/
async function openConnection(opts = {}) {
const o = {};
if (opts.host) o.host = opts.host;
if (opts.user) o.user = opts.user;
if (opts.password !== undefined) o.password = opts.password;
if (opts.database) o.database = opts.database;
return getConnection({
...o,
dateStrings: true,
multipleStatements: false,
connectTimeout: toInt(process.env.DB_CONNECTTIMEOUT, 10000),
queryTimeout: 0,
});
}
/** Auflösung der Org-UID aus CLI/Env-Overrides. */
function sourceOptions(args) {
// Reihenfolge: --from-* > DB_* (aus Vault). Fehlt beides, greift der Default
// von `getConnection()` — dieselben DB_*-Variablen.
return {
host: pick(args['from-host'], 'DB_HOST'),
user: pick(args['from-user'], 'DB_USER'),
password: pick(args['from-pass'], 'DB_PASS'),
database: pick(args['from-db'], 'DB_DATABASE'),
};
}
/** CLI-Wert, falls gesetzt — sonst der ENV-Fallback. */
const pick = (cliValue, envName) => {
if (cliValue && cliValue !== true) return String(cliValue);
const env = process.env[envName];
return env !== undefined && env !== '' ? env : undefined;
};
function targetOptions(args) {
// Reihenfolge: --to-* > DB_TO_* > (getConnection-Default: DB_* der Quelle).
//
// Der eigene Namensraum DB_TO_* ist wichtig für Zwei-Host-Läufe: die
// Variablen `DB_*` gelten für BEIDE Verbindungen, ein `-e DB_PASS=<ziel>`
// würde also auch das Quell-Passwort ersetzen. Mit `DB_TO_PASS` bleiben die
// Quell-Credentials aus Vault unangetastet.
return {
host: pick(args['to-host'], 'DB_TO_HOST'),
user: pick(args['to-user'], 'DB_TO_USER'),
password: pick(args['to-pass'], 'DB_TO_PASS'),
database: pick(args['to-db'], 'DB_TO_DATABASE'),
};
}
// ---------------------------------------------------------------------------
// 5. Scope-Aufbau
// ---------------------------------------------------------------------------
/**
* Organisation zu einer beliebigen Objekt-UID bestimmen.
*
* Ist die UID selbst eine Organisation (`Data.root = true`), wird sie direkt
* verwendet. Andernfalls wird über die Propagations-Links die zugehörige
* Organisation gesucht — dieselbe Query wie `getOrganizationForObject()` im
* Backend.
*
* @returns {Promise<{uid: string, title: string|null, resolvedFrom?: object}>}
*/
async function resolveOrganization(conn, uid, orgQ) {
if (!UUID_RE.test(uid)) {
throw new Error(`--org muss eine UUID sein, bekommen: ${uid}`);
}
const [obj] = await conn.query(
`SELECT BIN2UUID(UID) AS UID, Type, Title, JSON_VALUE(Data,'$.root') AS rootVal
FROM ObjectBase WHERE UID = UUID2BIN(?) LIMIT 1`,
[uid],
);
if (!obj) throw new Error(`Kein Objekt mit UID ${uid} gefunden`);
if (isRoot(obj.rootVal)) {
return { uid, title: obj.Title, type: obj.Type };
}
const list = PROPAGATION_LINK_TYPES.map((t) => orgQ(t)).join(',');
const [org] = await conn.query(
`SELECT BIN2UUID(org.UID) AS UID, org.Title AS Title
FROM ObjectBase obj
INNER JOIN Links l ON l.UID = obj.UID AND l.Type IN (${list})
INNER JOIN ObjectBase org ON org.UID = l.UIDTarget AND JSON_VALUE(org.Data,'$.root') = true
WHERE obj.UID = UUID2BIN(?)
LIMIT 1`,
[uid],
);
if (!org) {
throw new Error(
`Zu ${uid} (${obj.Type} "${obj.Title ?? ''}") wurde keine Organisation gefunden.\n` +
` Das Objekt hängt über keinen Propagations-Link an einer Organisation.`,
);
}
return {
uid: org.UID,
title: org.Title,
resolvedFrom: { uid, type: obj.Type, title: obj.Title },
};
}
/** `Data.root` kommt je nach Zugriffsweg als Zahl, String oder boolean. */
const isRoot = (v) => v === 1 || v === '1' || v === true || v === 'true';
/**
* Objekt-Scope einer Organisation berechnen und in `__scope` materialisieren.
*
* Der Scope wird als TEMPORARY TABLE gehalten: er wird von jeder Tabellen-
* Query gebraucht und ist mit ~250k Zeilen zu groß für IN-Listen.
*
* Der rekursive Arm expandiert nur von Objekten aus, die entweder die
* Start-Organisation selbst sind oder keine Organisation sind. Dadurch kann
* die Traversierung nicht in eine andere Organisation "überlaufen", falls
* deren Root über einen Propagations-Link erreichbar wäre.
*/
async function buildScope(conn, org, opts) {
const incoming = [...PROPAGATION_LINK_TYPES];
const outgoing = [];
for (const scope of opts.include) {
const def = EXTRA_SCOPES[scope] ?? {};
for (const t of def.incoming ?? []) incoming.push(t);
for (const t of def.outgoing ?? []) outgoing.push(t);
}
const inList = [...new Set(incoming)].map((t) => quote(t)).join(',');
const outList = [...new Set(outgoing)].map((t) => quote(t)).join(',');
try {
await conn.query(
`CREATE TEMPORARY TABLE __scope (UID binary(16) NOT NULL, PRIMARY KEY (UID)) ENGINE=InnoDB`,
);
} catch (e) {
throw new Error(
`Temporäre Tabelle konnte nicht angelegt werden (${e.message}).\n` +
` Der DB-Benutzer braucht das Recht CREATE TEMPORARY TABLES.`,
);
}
// (1) Mitglieds-Closure: alles, was über die Propagations-Links an der
// Organisation hängt. Ein rekursiver Durchlauf.
await conn.query(
`INSERT INTO __scope (UID)
WITH RECURSIVE scope(UID) AS (
SELECT UUID2BIN(?) AS UID
UNION
SELECT l.UID
FROM Links l
JOIN scope s ON l.UIDTarget = s.UID
JOIN ObjectBase so ON so.UID = s.UID
WHERE l.Type IN (${inList})
AND (s.UID = UUID2BIN(?) OR COALESCE(JSON_VALUE(so.Data,'$.root') = true, false) = false)
)
SELECT UID FROM scope`,
[org.uid, org.uid],
);
// (2) Zugehörigkeits-Closure über `UIDBelongsTo` — transitiv, bis zum Fixpunkt.
// Deckt org-eigene Objekte ohne Link (changeable, actionT, function), Objekte
// an einem Elternobjekt (Filter an list/group, eventJobT und action an eventT)
// und tiefere Ebenen ab. Muss vor der Member-Materialisierung laufen, damit
// deren Member-Zeilen mitkommen.
const belongsToGuard = `
AND (o.UID = UUID2BIN(?) OR COALESCE(JSON_VALUE(o.Data,'$.root') = true, false) = false)`;
for (let round = 1; round <= ORG_OWNED_MAX_ROUNDS; round++) {
const res = await conn.query(
`INSERT IGNORE INTO __scope (UID)
SELECT DISTINCT o.UID
FROM ObjectBase o JOIN __scope s ON o.UIDBelongsTo = s.UID
WHERE o.UID <> s.UID ${belongsToGuard}`,
[org.uid],
);
if (!Number(res.affectedRows ?? 0)) break;
}
// (3) Referenz-Closure (nur wenn Zusatz-Scopes angefordert wurden): den
// angegebenen Link-Typen in beide Richtungen bis zum Fixpunkt folgen.
// Läuft nach der Zugehörigkeits-Closure, damit deren Objekte als Startpunkte
// mitzählen.
if (opts.include.length) {
const guard = `
AND NOT EXISTS (
SELECT 1 FROM ObjectBase t
WHERE t.UID = l.UIDTarget AND t.UID <> UUID2BIN(?)
AND COALESCE(JSON_VALUE(t.Data,'$.root') = true, false) = true
)`;
for (let round = 1; round <= 6; round++) {
let added = 0;
if (outList) {
const res = await conn.query(
`INSERT IGNORE INTO __scope (UID)
SELECT DISTINCT l.UIDTarget
FROM Links l JOIN __scope s ON s.UID = l.UID
WHERE l.Type IN (${outList}) ${guard}`,
[org.uid],
);
added += Number(res.affectedRows ?? 0);
}
const res = await conn.query(
`INSERT IGNORE INTO __scope (UID)
SELECT DISTINCT l.UID
FROM Links l JOIN __scope s ON s.UID = l.UIDTarget
JOIN ObjectBase so ON so.UID = s.UID
WHERE l.Type IN (${inList})
AND (s.UID = UUID2BIN(?) OR COALESCE(JSON_VALUE(so.Data,'$.root') = true, false) = false)`,
[org.uid],
);
added += Number(res.affectedRows ?? 0);
if (!added) break;
}
}
// Member-Zeilen separat materialisieren (siehe TABLE_SPECS.Member).
// Der Filter ist die Vereinigung aus "Objekt-UID ist selbst eine Member-Zeile"
// und "ein Objekt des Scopes zeigt per UIDBelongsTo darauf"; danach wird auf
// tatsächlich existierende Member-Zeilen eingeschränkt.
await conn.query(
`CREATE TEMPORARY TABLE __member_scope (UID binary(16) NOT NULL, PRIMARY KEY (UID)) ENGINE=InnoDB`,
);
await conn.query(
`INSERT IGNORE INTO __member_scope (UID)
SELECT UID FROM Member WHERE UID IN (
SELECT UID FROM __scope
UNION
SELECT o.UIDBelongsTo FROM ObjectBase o JOIN __scope s ON s.UID = o.UID
WHERE o.UIDBelongsTo IS NOT NULL
)`,
);
}
/** Kurzreport über den Inhalt des Scopes (für inspect und export). */
async function scopeReport(conn, opts) {
const [totals] = await conn.query(
`SELECT (SELECT COUNT(*) FROM __scope) AS objects,
(SELECT COUNT(*) FROM __scope s JOIN ObjectBase o ON o.UID = s.UID) AS objectsExisting,
(SELECT COUNT(*) FROM __member_scope) AS memberRows,
(SELECT COUNT(*) FROM Links l JOIN __scope s ON s.UID = l.UID) AS linksInScope`,
);
const types = await conn.query(
`SELECT o.Type, COUNT(*) AS n
FROM __scope s JOIN ObjectBase o ON o.UID = s.UID
GROUP BY o.Type ORDER BY n DESC`,
);
const leaving = await conn.query(
`SELECT l.Type AS linkType, COALESCE(t.Type, 'FEHLT-ZIEL') AS targetType, COUNT(*) AS n
FROM Links l
JOIN __scope s ON s.UID = l.UID
LEFT JOIN __scope s2 ON s2.UID = l.UIDTarget
LEFT JOIN ObjectBase t ON t.UID = l.UIDTarget
WHERE s2.UID IS NULL
GROUP BY l.Type, targetType ORDER BY n DESC`,
);
const roots = await conn.query(
`SELECT BIN2UUID(o.UID) AS UID, o.Title
FROM __scope s JOIN ObjectBase o ON o.UID = s.UID
WHERE o.Type = 'group' AND o.UID = o.UIDBelongsTo AND JSON_VALUE(o.Data,'$.root') = true`,
);
return { totals, types, leaving, roots };
}
function printScopeReport(report, opts) {
const t = report.totals;
log('');
log('── Scope ────────────────────────────────────────────────');
log(` Objekt-UIDs im Scope ${t.objects}`);
if (Number(t.objectsExisting) !== Number(t.objects)) {
log(` davon in ObjectBase ${t.objectsExisting} (${Number(t.objects) - Number(t.objectsExisting)} UIDs ohne aktuelles Objekt)`);
}
log(` Member-Zeilen ${t.memberRows}`);
log(` Links im Scope ${t.linksInScope}`);
log(` Organisationen ${report.roots.length} (${report.roots.map((r) => r.Title).join(', ')})`);
log('');
log(' Objekte nach Typ:');
for (const r of report.types) log(` ${String(r.n).padStart(8)} ${r.Type}`);
if (report.leaving.length) {
log('');
log(' Links, die den Scope verlassen (Ziel nicht im Export):');
for (const r of report.leaving) {
log(` ${String(r.n).padStart(8)} ${r.linkType} -> ${r.targetType}`);
}
log('');
log(' Diese Objekte gehören nicht zur Organisation. Falls sie für einen');
log(' funktionierenden Restore nötig sind: --include filters,templates');
}
if (report.roots.length > 1) {
warn(`Scope enthält ${report.roots.length} Organisationen — das sollte nicht passieren!`);
}
}
// ---------------------------------------------------------------------------
// 6. Spalten-Metadaten
// ---------------------------------------------------------------------------
/**
* Alle Spalten der gewünschten Tabellen aus `information_schema` holen und
* klassifizieren.
*
* Damit bleibt das Skript schema-agnostisch: neue Spalten werden automatisch
* mitgenommen, generierte Spalten automatisch ausgelassen.
*/
async function loadColumns(conn, tableNames) {
const list = tableNames.map((n) => quote(n)).join(',');
const rows = await conn.query(
`SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, IS_NULLABLE, EXTRA,
CHARACTER_MAXIMUM_LENGTH, ORDINAL_POSITION
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN (${list})
ORDER BY TABLE_NAME, ORDINAL_POSITION`,
);
const map = new Map();
for (const r of rows) {
const dataType = String(r.DATA_TYPE).toLowerCase();
const col = {
name: r.COLUMN_NAME,
dataType,
columnType: String(r.COLUMN_TYPE).toLowerCase(),
nullable: r.IS_NULLABLE === 'YES',
extra: String(r.EXTRA ?? ''),
charLen: Number(r.CHARACTER_MAXIMUM_LENGTH ?? 0),
// `DEFAULT_GENERATED` (Spalten mit Default-Ausdruck, z.B. Member.Geo)
// darf NICHT als generiert gelten — nur echte GENERATED ALWAYS AS.
generated: /VIRTUAL GENERATED|STORED GENERATED/i.test(String(r.EXTRA ?? '')),
isUuid: dataType === 'binary' && Number(r.CHARACTER_MAXIMUM_LENGTH) === 16,
isBinary: ['binary', 'varbinary', 'tinyblob', 'blob', 'mediumblob', 'longblob'].includes(dataType),
isGeometry: [
'point', 'linestring', 'polygon', 'multipoint', 'multilinestring',
'multipolygon', 'geometry', 'geometrycollection',
].includes(dataType),
};
(map.get(r.TABLE_NAME) ?? map.set(r.TABLE_NAME, []).get(r.TABLE_NAME)).push(col);
}
return map;
}
/** Kurzform einer UUID für Dateinamen. */
const shortUuid = (u) => String(u).slice(0, 8);
/**
* Prüfen, ob eine Tabelle existiert und welcher Art sie ist.
*
* Nur BASE TABLE / SYSTEM VERSIONED werden exportiert — die Views `Objects`,
* `ObjectTargets` und `HasTarget` leiten sich aus den Basistabellen ab und
* dürfen nicht eingespielt werden.
*/
async function loadTableTypes(conn, names) {
const list = names.map((n) => quote(n)).join(',');
const rows = await conn.query(
`SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN (${list})`,
);
return new Map(rows.map((r) => [r.TABLE_NAME, String(r.TABLE_TYPE).toUpperCase()]));
}
function writableColumns(cols, opts) {
return cols.filter((c) => {
if (!c.generated) return true;
// Die Periodenspalten sind die Ausnahme: im History-Modus werden sie
// explizit geschrieben (mit `system_versioning_insert_history=1`).
return opts.history && (c.name === 'ValidFrom' || c.name === 'ValidUntil');
});
}
/**
* SELECT-Ausdrücke bauen.
*
* Die Konvertierung passiert in SQL, nicht in JS: `BIN2UUID()` liefert den
* lesbaren String, `HEX()` die Binärdaten, `ST_AsWKB()`/`ST_SRID()` die
* Geometrie. Damit gibt es keine Zeitzonen- oder Encoding-Überraschungen
* durch den Node-Treiber.
*/
function buildSelect(cols) {
return cols.map((c) => {
if (c.isUuid) return `BIN2UUID(${id(c.name)}) AS ${id(c.name)}`;
if (c.isGeometry) {
return `HEX(ST_AsWKB(${id(c.name)})) AS ${id(c.name)}, ` +
`ST_SRID(${id(c.name)}) AS ${id('__srid_' + c.name)}`;
}
if (c.isBinary) return `HEX(${id(c.name)}) AS ${id(c.name)}`;
return id(c.name);
}).join(', ');
}
/** Buffer (bit) -> Dezimalstring. */
function bufferToBitString(buf) {
let v = 0n;
for (const b of buf) v = (v << 8n) | BigInt(b);
return v.toString();
}
/**
* Einen Treiber-Wert in ein SQL-Literal umwandeln.
*
* UUIDs gehen über `UUID2BIN()` — die Gegenrichtung zu `BIN2UUID()` im
* SELECT und die vom Projekt vorgegebene Konvertierung (siehe
* `backend-uuid-approach`-Regel).
*/
function encodeValue(col, value, row) {
if (value === null || value === undefined) return 'NULL';
if (col.isUuid) return `UUID2BIN(${quote(value)})`;
if (col.isGeometry) {
const wkb = Buffer.isBuffer(value) ? value.toString('hex') : String(value);
const srid = Number(row['__srid_' + col.name] ?? 0);
return `ST_GeomFromWKB(x'${wkb}', ${Number.isFinite(srid) ? srid : 0})`;
}
if (col.isBinary) {
return `x'${Buffer.isBuffer(value) ? value.toString('hex') : String(value)}'`;
}
if (col.dataType === 'bit') {
return Buffer.isBuffer(value) ? bufferToBitString(value) : String(value);
}
// MariaDB kennt keinen eigenen JSON-Typ: `longtext ... CHECK (json_valid(col))`
// (z.B. `eventLog.Data`) meldet der Server über das Wire-Protokoll als JSON,
// der Treiber liefert dann bereits ein geparstes Objekt. Wieder
// serialisieren, sonst landet "[object Object]" im Dump.
if (typeof value === 'object' && value !== null && !Buffer.isBuffer(value)) {
if (value instanceof Date) return quote(formatDateLocal(value));
return quote(JSON.stringify(value));
}
if (NUMERIC_TYPES.has(col.dataType)) {
const s = String(value);
// bigint kommt als String (bigNumberStrings) — nur weiterreichen, wenn es
// wirklich eine Zahl ist, sonst wäre es eine Injection-Fläche.
return NUMBER_RE.test(s) ? s : quote(s);
}
if (typeof value === 'number' && Number.isFinite(value)) return String(value);
return quote(Buffer.isBuffer(value) ? value.toString('utf8') : String(value));
}
/**
* Notfall-Formatierung für einen `Date`, falls der Treiber doch castet.
*
* Normalerweise kommt das nicht vor (`dateStrings: true` in
* {@link openConnection}) — aber ein stillschweigend verschobener Zeitstempel
* wäre schlimmer als eine lokale Formatierung.
*/
function formatDateLocal(d) {
const p = (n, w = 2) => String(n).padStart(w, '0');
return `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())} ` +
`${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}.${p(d.getMilliseconds(), 3)}000`;
}
// ---------------------------------------------------------------------------
// 7. Zeilen-Streaming und INSERT-Batches
// ---------------------------------------------------------------------------
/**
* Ergebnis einer Query als Stream durchreichen.
*
* Wichtig für `Visible` (~800k Zeilen) und `Links`: die Treiber-Query würde
* sonst alle Zeilen gleichzeitig im Speicher halten. `queryStream` respektiert
* Backpressure.
*/
async function* streamRows(conn, sql, params) {
const stream = conn.queryStream(sql, params);
for await (const row of stream) yield row;
}
/**
* Sammelt Wertetupel und schreibt sie als mehrzeilige INSERT-Statements.
*
* Flush-Kriterium ist entweder die Zeilenzahl (`--batch`) oder die
* Statement-Größe (`--max-statement`) — Objekt-`Data`-Felder sind sehr
* unterschiedlich groß, deshalb beides.
*/
class InsertBatcher {
/**
* @param {(sql: string) => Promise<void>} sink
* @param {{batch: number, maxBytes: number, batchMode: boolean}} opts
*/
constructor(sink, opts) {
this.sink = sink;
this.batchSize = opts.batch;
this.maxBytes = opts.maxBytes;
this.batchMode = opts.batchMode;
this.table = null;
this.columns = null;
this.tuples = [];
this.size = 0;
this.statements = 0;
}
/** Neue Tabelle beginnen (flusht die vorherige). */
async table_(name, columns) {
await this.flush();
this.table = name;
this.columns = columns;
}
/** Eine Zeile hinzufügen. */
async add(tuple) {
if (!this.batchMode) {
this.tuples.push(tuple);
await this.flush();
return;
}
this.tuples.push(tuple);
this.size += tuple.length + 2;
if (this.tuples.length >= this.batchSize || this.size >= this.maxBytes) {
await this.flush();
}
}
/** Aktuellen Batch schreiben. */
async flush() {
if (!this.tuples.length) return;
const head = `INSERT INTO ${id(this.table)} (${this.columns.map(id).join(', ')}) VALUES\n`;
const stmt = head + this.tuples.join(',\n') + ';\n';
this.tuples = [];
this.size = 0;
this.statements++;
await this.sink(stmt);
}
}
// ---------------------------------------------------------------------------
// 8. Export-Pipeline
// ---------------------------------------------------------------------------
/** WHERE-Klausel einer Tabelle. */
function tableFilter(spec, ctx) {
if (typeof spec.where === 'function') return spec.where(ctx);
switch (spec.scope) {
case 'member': return `UID IN (SELECT UID FROM __member_scope)`;
case 'org': return `UID IN (SELECT UID FROM __scope)`;
default:
throw new Error(`Unbekannter Scope "${spec.scope}" für Tabelle ${spec.name}`);
}
}
/** Tabellen für diesen Lauf bestimmen. */
function selectedTables(opts) {
if (opts.tables.length) {
const wanted = new Set(opts.tables.map((t) => t.toLowerCase()));
const found = TABLE_SPECS.filter((s) => wanted.has(s.name.toLowerCase()));
const missing = opts.tables.filter((t) => !TABLE_SPECS.some((s) => s.name.toLowerCase() === t.toLowerCase()));
if (missing.length) warn(`Unbekannte Tabellen (ignoriert): ${missing.join(', ')}`);
return found.sort((a, b) => a.order - b.order);
}
const names = new Set(TABLE_SPECS.filter((s) => s.always).map((s) => s.name));
for (const flag of opts.include) {
for (const t of SCOPE_TABLES[flag] ?? []) names.add(t);
}
return TABLE_SPECS.filter((s) => names.has(s.name)).sort((a, b) => a.order - b.order);
}
/** Dump-Kopf. */
function headerSql(ctx, opts, plan) {
const tables = plan.map((p) => p.spec.name).join(', ');
const extra = opts.include.length ? opts.include.join(',') : '–';
// Maschinenlesbares Manifest der geschriebenen Spalten. Das Einspiel-Skript
// prüft damit vorab, ob das Ziel-Schema zum Dump passt (z.B. wenn die
// Zieldatenbank eine ältere Migration-Stufe hat).
const manifest = plan.map((p) => `${p.spec.name}:${p.cols.map((c) => c.name).join(',')}`).join('|');
return `
-- ===========================================================================
-- CommTool organizations transfer dump
-- org: ${ctx.org.uid}${ctx.org.title ? ` (${ctx.org.title})` : ''}
${ctx.org.resolvedFrom
? `-- resolved: ${ctx.org.resolvedFrom.uid} (${ctx.org.resolvedFrom.type}${
ctx.org.resolvedFrom.title ? ` "${ctx.org.resolvedFrom.title}"` : ''})\n`
: ''}-- generated: ${new Date().toISOString()}
-- source db: ${process.env.DB_DATABASE ?? '?'} @ ${process.env.DB_HOST ?? '?'}
-- tables: ${tables}
-- include: ${extra}
-- history: ${opts.history ? 'YES (FOR SYSTEM_TIME ALL)' : 'no'}
-- objects: ${ctx.report.totals.objects} member rows: ${ctx.report.totals.memberRows} links: ${ctx.report.totals.linksInScope}
-- tool: members-back/src/scripts/orgTransfer.js
--
-- Einspielen:
-- mysql -h HOST -u USER -p DB < ${path.basename(opts.outFile || 'dump.sql')}
-- oder:
-- node src/scripts/orgTransfer.js import --in ${path.basename(opts.outFile || 'dump.sql')} --to-db DB
--
-- Voraussetzung: die Zieldatenbank besitzt die Funktionen UUID2BIN()/BIN2UUID()
-- (deployt über initFunctions.sql) und dieselben Spalten wie die Quelle.
-- ===========================================================================
SET NAMES utf8mb4;
SET SESSION sql_mode = REPLACE(@@SESSION.sql_mode, 'NO_BACKSLASH_ESCAPES', '');
SET SESSION foreign_key_checks = 0;
SET SESSION autocommit = 1;
${opts.history ? `-- System-Versionierung: Historie wird explizit geschrieben
SET SESSION system_versioning_insert_history = 1;
SET SESSION system_versioning_alter_history = KEEP;
` : ''}
-- @manifest ${manifest}
`;
}
/** Dump-Fuß. */
function footerSql(opts) {
return `
-- ===========================================================================
-- ${opts.stats.rows} Datenzeilen in ${opts.stats.statements} INSERT-Statements
-- ===========================================================================
`;
}
/**
* Kompletten Export in einen `sink` schreiben.
*
* @param {any} conn Quell-Verbindung (Scope muss bereits gebaut sein)
* @param {object} ctx
* @param {object} opts
* @param {(sql: string) => Promise<void>} sink
*/
/**
* Auflösen, welche Tabellen mit welchen Spalten tatsächlich geschrieben werden.
*
* Das Ergebnis steht sowohl im Dump-Manifest als auch im Insert-Loop — beides
* muss dieselbe Sicht haben.
*/
function planTables(ctx, opts) {
const plan = [];
for (const spec of selectedTables(opts)) {
const tableType = ctx.tableTypes.get(spec.name);
if (!tableType) { warn(`Tabelle ${spec.name} existiert nicht — übersprungen`); continue; }
if (!tableType.startsWith('BASE TABLE') && !tableType.startsWith('SYSTEM VERSIONED')) {
warn(`Tabelle ${spec.name} ist ${tableType} — übersprungen`);
continue;
}
// `FOR SYSTEM_TIME ALL` gibt es nur auf system-versionierten Tabellen
// (Member ist z.B. keine). Deshalb pro Tabelle entscheiden.
const versioned = tableType.startsWith('SYSTEM VERSIONED');
const history = Boolean(opts.history) && versioned;
if (opts.history && !versioned) {
log(` (${spec.name} ist nicht system-versioned — nur aktuelle Zeilen)`);
}
const cols = writableColumns(ctx.columns.get(spec.name) ?? [], { ...opts, history });
if (!cols.length) { warn(`Tabelle ${spec.name}: keine schreibbaren Spalten — übersprungen`); continue; }
plan.push({ spec, cols, history });
}
return plan;
}
async function exportTo(conn, ctx, opts, sink) {
const progress = wantProgress(opts);
const batcher = new InsertBatcher(sink, {
batch: opts.batch,
maxBytes: opts.maxBytes,
batchMode: opts.batchMode,
});
const stats = { rows: 0, statements: 0, tables: {} };
// Erst alle Tabellen auflösen und planen, damit das Manifest im Kopf
// vollständig ist.
const plan = planTables(ctx, opts);
await sink(headerSql(ctx, opts, plan));
for (const { spec, cols, history } of plan) {
const from = `${id(spec.name)}${history ? ' FOR SYSTEM_TIME ALL' : ''}`;
const orderBy = history ? ` ORDER BY ${id('UID')}, ${id('ValidFrom')}` : '';
const limit = opts.limit ? ` LIMIT ${opts.limit}` : '';
const sql = `SELECT ${buildSelect(cols)} FROM ${from} WHERE ${tableFilter(spec, ctx)}${orderBy}${limit}`;
log(` → ${spec.name} ...`);
const t0 = Date.now();
await batcher.table_(spec.name, cols.map((c) => c.name));
let n = 0;
for await (const row of streamRows(conn, sql)) {
const tuple = '(' + cols.map((c) => encodeValue(c, row[c.name], row)).join(', ') + ')';
await batcher.add(tuple);
n++;
if (progress && n % 25000 === 0) {
process.stderr.write(`\r ${spec.name}: ${n} Zeilen ...`);
}
}
await batcher.flush();
stats.rows += n;
stats.tables[spec.name] = n;
if (progress) {
process.stderr.write(`\r ${spec.name}: ${n} Zeilen (${secs(Date.now() - t0)}) \n`);
}
}
stats.statements = batcher.statements;
await sink(footerSql({ ...opts, stats }));
return stats;
}
// ---------------------------------------------------------------------------
// 9. Dump einspielen
// ---------------------------------------------------------------------------
/** Ist das Statement nur ein Kommentar/Whitespace? */
function isEffectivelyEmpty(stmt) {
let s = stmt;
// Kommentare entfernen, dabei Strings in Ruhe lassen.
s = s.replace(/\/\*[\s\S]*?\*\//g, ' ');
s = s.replace(/^\s*(--|#)[^\n]*$/gm, ' ');
s = s.replace(/--[^\n]*/g, ' ');
return s.trim().length === 0;
}
/**
* SQL-Text in Statements zerlegen (push-basiert).
*
* Ein naives `split(';')` ist nicht möglich: die INSERT-Statements enthalten
* JSON-Payloads. Deshalb ein kleiner Zustandsautomat, der String-Literale,
* Backslash-Escapes, Bezeichner-Quotes und Kommentare kennt. `--` und `#`
* gelten nur außerhalb von Strings als Kommentar.
*
* Bewusst push-basiert und nicht als Stream-Generator: dieselbe Klasse
* bedient sowohl `import` (Datei -> Ziel-DB) als auch `copy` (Export-Sink ->
* Ziel-DB), ohne den Dump zwischendurch zu puffern.
*/
class StatementSplitter {
/** @param {(stmt: string) => Promise<void>} onStatement */
constructor(onStatement) {
this.onStatement = onStatement;
this.state = 'plain'; // plain | sq | dq | bq | line | block
this.escaped = false;
this.buf = '';
this.count = 0;
this.bytes = 0;
}
async push(text) {
for (let i = 0; i < text.length; i++) {
const ch = text[i];
const nx = text[i + 1];
if (this.escaped) { this.buf += ch; this.escaped = false; continue; }
if (this.state === 'line') {
this.buf += ch;
if (ch === '\n') this.state = 'plain';
continue;
}
if (this.state === 'block') {
this.buf += ch;
if (ch === '*' && nx === '/') { this.buf += nx; i++; this.state = 'plain'; }
continue;
}
if (this.state === 'sq' || this.state === 'dq' || this.state === 'bq') {
if (ch === '\\') { this.buf += ch; this.escaped = true; continue; }
const closer = this.state === 'sq' ? "'" : this.state === 'dq' ? '"' : '`';
if (ch === closer && nx === closer) { this.buf += ch + nx; i++; continue; }
this.buf += ch;
if (ch === closer) this.state = 'plain';
continue;
}
// plain
if (ch === "'") { this.state = 'sq'; this.buf += ch; continue; }
if (ch === '"') { this.state = 'dq'; this.buf += ch; continue; }
if (ch === '`') { this.state = 'bq'; this.buf += ch; continue; }
if (ch === '-' && nx === '-') { this.state = 'line'; this.buf += ch; continue; }
if (ch === '#') { this.state = 'line'; this.buf += ch; continue; }
if (ch === '/' && nx === '*') { this.state = 'block'; this.buf += ch + nx; i++; continue; }
if (ch === ';') { await this.emit(); continue; }
this.buf += ch;
}
}
/** Rest am Dateiende ausgeben. */
async end() { await this.emit(); }
async emit() {
const stmt = this.buf;
this.buf = '';
if (!stmt.trim() || isEffectivelyEmpty(stmt)) return;
this.count++;
this.bytes += stmt.length;
await this.onStatement(stmt);
}
}
/** Lesbaren Stream zeilenweise durch den Splitter schicken. */
async function runDump(readable, onStatement) {
readable.setEncoding('utf8');
const splitter = new StatementSplitter(onStatement);
for await (const chunk of readable) await splitter.push(chunk);
await splitter.end();
return { count: splitter.count, bytes: splitter.bytes };
}
/** Org-UID aus dem Dump-Kopf lesen (für den Vorab-Check). */
function parseDumpOrg(headerText) {
const m = /^--\s*org:\s*([0-9a-fA-F-]{36})/m.exec(headerText);
return m ? m[1] : null;
}
/**
* Spalten-Manifest aus dem Dump-Kopf lesen.
* Format: `-- @manifest Member:UID,Display|ObjectBase:UID,Type,...`
*
* @returns {Map<string, string[]>}
*/
function parseDumpManifest(headerText) {
const m = /^--\s*@manifest\s+(.+)$/m.exec(headerText);
const out = new Map();
if (!m) return out;
for (const entry of m[1].trim().split('|')) {
const idx = entry.indexOf(':');
if (idx === -1) continue;
const table = entry.slice(0, idx).trim();
const cols = entry.slice(idx + 1).split(',').map((c) => c.trim()).filter(Boolean);
if (table && cols.length) out.set(table, cols);
}
return out;
}
/**
* Prüfen, ob die Zieldatenbank die im Dump verwendeten Tabellen und Spalten
* besitzt.
*
* Der Export liest die Spalten aus `information_schema` der Quelle. Ist das
* Ziel eine ältere Migrationsstufe, fehlen dort Spalten — ohne diesen Check
* bricht das Einspielen mitten im Dump mit einem kryptischen
* "Unknown column" ab.
*/
async function checkTargetSchema(conn, manifest) {
if (!manifest?.size) return;
const names = [...manifest.keys()];
const types = await loadTableTypes(conn, names);
const columns = await loadColumns(conn, names);
const problems = [];
const notes = [];
for (const [table, wanted] of manifest) {
const type = types.get(table);
if (!type) { problems.push(`Tabelle \`${table}\` fehlt im Ziel`); continue; }
const have = new Set((columns.get(table) ?? []).map((c) => c.name.toLowerCase()));
const missing = wanted.filter((c) => !have.has(c.toLowerCase()));
if (missing.length) {
problems.push(`\`${table}\`: fehlende Spalten -> ${missing.join(', ')}`);
}
const extra = [...have].filter((c) => !wanted.some((w) => w.toLowerCase() === c));
// Generierte Spalten (TUID, ValidFrom, ...) meldet das Ziel zwangsläufig
// als "zusätzlich" — das ist erwartet und kein Problem.
const generated = new Set(
(columns.get(table) ?? []).filter((c) => c.generated).map((c) => c.name.toLowerCase()),
);
const extraReal = extra.filter((c) => !generated.has(c));
if (extraReal.length) notes.push(`\`${table}\`: Ziel hat zusätzliche Spalten (auf Defaults) -> ${extraReal.join(', ')}`);
}
for (const n of notes) warn(n);
if (problems.length) {
throw new Error(
'Ziel-Schema passt nicht zum Dump:\n ' + problems.join('\n ') +
'\n Der Dump stammt aus einem Schema mit anderen Spalten. Entweder das Ziel\n' +
' migrieren oder den Dump aus einer passenden Quelle erzeugen.',
);
}
}
/** Voraussetzungen der Zieldatenbank prüfen. */
async function assertTargetReady(conn) {
const rows = await conn.query(
`SELECT ROUTINE_NAME FROM information_schema.ROUTINES
WHERE ROUTINE_SCHEMA = DATABASE() AND ROUTINE_NAME IN ('UUID2BIN','BIN2UUID')`,
);
const have = new Set(rows.map((r) => r.ROUTINE_NAME));
const missing = ['UUID2BIN', 'BIN2UUID'].filter((f) => !have.has(f));
if (missing.length) {
throw new Error(
`Der Zieldatenbank fehlen die Funktionen: ${missing.join(', ')}.\n` +
` Sie werden über initFunctions.sql deployt und sind Voraussetzung,\n` +
` weil der Dump UUIDs mit UUID2BIN('...') schreibt.`,
);
}
}
/** Prüfen, ob die Organisation im Ziel schon existiert. */
async function targetHasOrg(conn, orgUid) {
if (!orgUid) return null;
const [row] = await conn.query(
`SELECT COUNT(*) AS n FROM ObjectBase WHERE UID = UUID2BIN(?)`,
[orgUid],
);
return Number(row?.n ?? 0) > 0;
}
/**
* Vorhandene Organisation im Ziel entfernen — der „tolerante" Import-Pfad.
*
* **Warum das überhaupt nötig ist.** `checkOrgaExists()` (config/initOrga.js)
* läuft beim Backend-Start für **jede** Organisation, die eine Config-Datei
* (`config/admin/UUID-<uuid>.yaml`) hat, und legt sie an, falls sie fehlt:
* Org-Objekt, BOT-User (`extern`), `@@SuperAdmin`-Job und Superuser-Filter —
* mit Defaults, also einem Platzhalter-Namen, nicht mit den echten Daten.
*
* Ein Dump-Import trifft dieses Ziel also praktisch **nie leer** an. Und ein
* „drüberbügeln" kann nicht funktionieren, weil der Stub seine System-Objekte
* bei jeder Erzeugung mit **neuen Zufalls-UIDs** anlegt (`SELECT UIDV1() AS
* UIDExtern, UIDV1() AS UIDJob, UIDV1() AS UIDFilter` in initOrga.js). Diese
* UIDs können nie mit den gleichnamigen Objekten aus dem Dump zusammenfallen:
* es blieben doppelte BOT-User, zwei `@@SuperAdmin`-Jobs und zwei
* Superuser-Filter zurück. Vorher scheitert der Import schon am
* Primärschlüssel — `ObjectBase` ist auf `(UID, ValidUntil)` geschlüsselt, und
* die offene Zeile (`ValidUntil` = 2106-02-07) ist bereits belegt; `Member`
* (`UID`) und `Visible` (`UID, UIDUser`) ebenso.
*
* **Was gelöscht wird.** Die Closure wird im Ziel mit denselben Regeln
* aufgebaut wie beim Export (`buildScope`), dann Zeile für Zeile entfernt —
* und nur für die Tabellen, die der Dump auch wieder einspielt. Was der Dump
* nicht mitbringt, wird nicht angefasst.
*
* Bei system-versionierten Tabellen (`ObjectBase`, `Links`) beendet das
* `DELETE` nur die offene Version (`ValidUntil` = jetzt); die Historie bleibt
* erhalten. Damit ist der Primärschlüssel frei und der anschließende
* `INSERT` des Dumps passt wieder.
*
* Abhängige Tabellen werden **vor** ihren Eltern gelöscht (`order` absteigend),
* sonst räumt z.B. `TransactionLines` über einen Subselect auf schon
* gelöschte `Transactions` nichts mehr weg.
*
* @param {any} conn Ziel-Verbindung
* @param {string} orgUid Organisations-UID
* @param {object} opts Optionen (nutzt `include`, `tables`, `dryRun`)
* @param {string[]|null} [tables] Tabellen des Dumps; `null` = Default-Satz
* @returns {Promise<Array<{table: string, removed: number}>>}
*/
async function purgeTargetOrg(conn, orgUid, opts, tables = null) {
// Frischer Scope: erst aufräumen, dann aufbauen — dieselbe Verbindung kann
// die Temporärtabellen aus einem früheren Lauf noch halten.
await conn.query('DROP TEMPORARY TABLE IF EXISTS __scope');
await conn.query('DROP TEMPORARY TABLE IF EXISTS __member_scope');
await buildScope(conn, { uid: orgUid }, { ...opts, include: opts.include ?? [] });
const wanted = tables?.length ? new Set(tables.map((t) => t.toLowerCase())) : null;
const specs = [...selectedTables(opts)].sort((a, b) => b.order - a.order);
const ctx = { org: { uid: orgUid }, q: quote };
const removed = [];
for (const spec of specs) {
if (wanted && !wanted.has(spec.name.toLowerCase())) continue;
let where;
if (spec.scope === 'member') where = 'UID IN (SELECT UID FROM __member_scope)';
else if (spec.scope === 'org') where = 'UID IN (SELECT UID FROM __scope)';
else if (typeof spec.where === 'function') where = spec.where(ctx);
else continue;
const table = id(spec.name);
if (opts.dryRun) {
const [row] = await conn.query(`SELECT COUNT(*) AS n FROM ${table} WHERE ${where}`);
removed.push({ table: spec.name, removed: Number(row?.n ?? 0) });
continue;
}
const res = await conn.query(`DELETE FROM ${table} WHERE ${where}`);
removed.push({ table: spec.name, removed: Number(res?.affectedRows ?? 0) });
}
return removed;
}
/** `--mode replace`-Report aufs Terminal. */
function printPurgeReport(purged, dryRun) {
const total = purged.reduce((n, p) => n + p.removed, 0);
log(
` --mode replace: ${total} Zeile(n) der vorhandenen Organisation entfernt` +
`${dryRun ? ' (dry-run — nur gezählt, nichts gelöscht)' : ''}`,
);
for (const p of purged) {
if (p.removed) log(` ${p.table.padEnd(18)} ${p.removed}`);
}
}
/**
* Dump-Statements auf eine Verbindung ausspielen.
*
* @param {any} conn Ziel-Verbindung
* @param {NodeJS.ReadableStream} readable Dump-Quelle
* @param {object} opts
*/
async function applyDump(conn, readable, opts) {
const progress = wantProgress(opts);
let count = 0;
let bytes = 0;
let peak = 0;
const t0 = Date.now();
const { count: total, bytes: totalBytes } = await runDump(readable, async (stmt) => {
count++;
bytes += stmt.length;
peak = Math.max(peak, stmt.length);
if (opts.dryRun) {
if (count <= 3) log(` [dry-run] ${stmt.slice(0, 110).replace(/\s+/g, ' ')}...`);
return;
}
await conn.query(stmt);
if (progress && count % 200 === 0) {
process.stderr.write(`\r ${count} Statements (${human(bytes)}) ...`);
}
});
if (progress) {
process.stderr.write(`\r ${total} Statements, ${human(totalBytes)} in ${secs(Date.now() - t0)} \n`);
}
return { count: total, bytes: totalBytes, peak };
}
// ---------------------------------------------------------------------------
// 10. Kommandos
// ---------------------------------------------------------------------------
const KNOWN_SCOPES = [...Object.keys(EXTRA_SCOPES), ...Object.keys(SCOPE_TABLES)];
/** CLI-Argumente in einen Options-Objekt normalisieren. */
function normalizeOptions(args) {
const opts = {
org: args.org && args.org !== true ? String(args.org) : null,
include: [],
tables: [],
history: Boolean(args['include-history']),
batch: Math.max(1, toInt(args.batch, 200)),
maxBytes: Math.max(64 * 1024, toInt(args['max-statement'], 1024) * 1024),
batchMode: !args['no-batch'],
limit: toInt(args.limit, 0),
dryRun: Boolean(args['dry-run']),
quiet: Boolean(args.quiet),
mode: args.mode && args.mode !== true ? String(args.mode) : 'abort',
outFile: null,
q: quote,
};
const list = (v) => String(v).split(',').map((s) => s.trim()).filter(Boolean);
if (args.include && args.include !== true) opts.include = list(args.include);
if (args.tables && args.tables !== true) opts.tables = list(args.tables);
for (const s of opts.include) {
if (!KNOWN_SCOPES.includes(s)) {
throw new Error(`Unbekannter Scope "${s}" für --include. Bekannt: ${KNOWN_SCOPES.join(', ')}`);
}
}
if (!['abort', 'skip', 'replace', 'overwrite'].includes(opts.mode)) {
throw new Error(`Unbekannter --mode "${opts.mode}". Bekannt: abort, skip, replace, overwrite`);
}
return opts;
}
/**
* Verbindungsdaten für's Log (ohne Passwort).
*
* Wichtig als Sicherheitsnetz: zeigt immer die *tatsächlich* benutzte
* Verbindung, nicht die aus der Umgebung — sonst könnte ein Export aus Prod
* versehentlich wie ein Export aus der Test-DB aussehen (oder umgekehrt).
*/
function describeConn(o = {}) {
return `${o.user ?? process.env.DB_USER ?? '?'}@${o.host ?? process.env.DB_HOST ?? '?'}/${o.database ?? process.env.DB_DATABASE ?? '?'}`;
}
/** Scope + Metadaten der Quelle aufbauen. */
async function buildContext(conn, opts) {
const org = await resolveOrganization(conn, opts.org, quote);
if (org.resolvedFrom) {
log(` --org ${org.resolvedFrom.uid} (${org.resolvedFrom.type}) -> Organisation ${org.uid} (${org.title})`);
} else {
log(` Organisation: ${org.uid} (${org.title})`);
}
await buildScope(conn, org, opts);
const report = await scopeReport(conn, opts);
const tables = selectedTables(opts).map((s) => s.name);
const tableTypes = await loadTableTypes(conn, tables);
const columns = await loadColumns(conn, tables);
return { org, report, tableTypes, columns, q: quote };
}
/** Dateiname für den Dump. */
function defaultOutName(org, gzip) {
const ts = new Date().toISOString().replace(/[-:]/g, '').replace(/\..+$/, '').replace('T', '-');
return `org-${shortUuid(org.uid)}-${ts}.sql${gzip ? '.gz' : ''}`;
}
/** Schreib-Sink auf eine Datei (optional gzip). */
async function openFileSink(file, gzip) {
const fileStream = fs.createWriteStream(file, { flags: 'w' });
let error = null;
fileStream.on('error', (e) => { error = e; });
let head = fileStream;
if (gzip) {
head = zlib.createGzip({ level: 6 });
// Fehler der gzip-Stufe auf den Dateistream durchreichen, damit ein
// einziger Fehlerkanal bleibt.
head.on('error', (e) => { error = e; });
head.pipe(fileStream);
}
// Backpressure: genau ein Drain-Waiter, damit nicht pro Schreibvorgang ein
// 'error'-Listener auf dem Stream landet (MaxListeners-Warnung).
let waiting = null;
head.on('drain', () => { const w = waiting; waiting = null; w?.(); });
head.on('error', () => { const w = waiting; waiting = null; w?.(); });
const sink = (chunk) => new Promise((resolve, reject) => {
if (error) return reject(error);
if (head.write(chunk)) return resolve();
waiting = () => (error ? reject(error) : resolve());
});
const finish = () => new Promise((resolve, reject) => {
if (error) return reject(error);
fileStream.once('finish', resolve);
fileStream.once('error', reject);
head.end();
});
return { sink, finish };
}
/** Quelldatei öffnen, gzip transparent entpacken. */
async function openDumpReadable(file) {
const stream = fs.createReadStream(file);
const head = Buffer.alloc(2);
const fh = await fsp.open(file, 'r');
await fh.read(head, 0, 2, 0);
await fh.close();
if (head[0] === 0x1f && head[1] === 0x8b) {
const gunzip = zlib.createGunzip();
stream.pipe(gunzip);
return gunzip;
}
return stream;
}
/** Erste Bytes einer Datei als Text (für den Kopf-Kommentar). */
async function readHead(file, bytes = 8192) {
const fh = await fsp.open(file, 'r');
try {
const buf = Buffer.alloc(bytes);
const { bytesRead } = await fh.read(buf, 0, bytes, 0);
return buf.toString('utf8', 0, bytesRead);
} finally {
await fh.close();
}
}
// --- inspect ---------------------------------------------------------------
async function cmdInspect(args) {
const opts = normalizeOptions(args);
if (!opts.org) throw new Error('inspect braucht --org <UUID>');
const conn = await openConnection(sourceOptions(args));
try {
log(` Quelle: ${describeConn(sourceOptions(args))} (${opts.include.length ? 'include=' + opts.include.join(',') : 'nur Kern-Scope'})`);
const ctx = await buildContext(conn, opts);
printScopeReport(ctx.report, opts);
log('');
log(' Geplante Tabellen:');
for (const spec of selectedTables(opts)) {
const type = ctx.tableTypes.get(spec.name);
const n = (ctx.columns.get(spec.name) ?? []).filter((c) => !c.generated).length;
log(` ${(type ?? 'FEHLT').padEnd(18)} ${spec.name.padEnd(18)} (scope=${spec.scope}, ${n} Spalten)`);
}
} finally {
await conn.end();
}
}
// --- export ---------------------------------------------------------------
async function cmdExport(args) {
const opts = normalizeOptions(args);
if (!opts.org) throw new Error('export braucht --org <UUID>');
const conn = await openConnection(sourceOptions(args));
try {
log(` Quelle: ${describeConn(sourceOptions(args))}`);
const ctx = await buildContext(conn, opts);
printScopeReport(ctx.report, opts);
if (opts.dryRun) {
log('\n --dry-run: es wurde nichts geschrieben.');
return;
}
if (ctx.report.roots.length === 0) {
throw new Error(
'Der Scope enthält keine Organisation — abgebrochen, um keinen Unsinn zu exportieren.\n' +
' Erwartet wird ein group-Objekt mit UID = UIDBelongsTo und Data.root = true.',
);
}
const gzip = Boolean(args.gzip);
opts.outFile = args.out && args.out !== true ? String(args.out) : defaultOutName(ctx.org, gzip);
const { sink, finish } = await openFileSink(opts.outFile, gzip);
log('');
log(` schreibe ${opts.outFile}${gzip ? ' (gzip)' : ''}`);
log(` ${'Tabelle'.padEnd(20)}${'Zeilen'.padStart(10)}`);
const t0 = Date.now();
const stats = await exportTo(conn, ctx, opts, sink);
await finish();
const { size } = await fsp.stat(opts.outFile);
log('');
log(` fertig in ${secs(Date.now() - t0)}: ${stats.rows} Zeilen, ${stats.statements} INSERTs, ${human(size)}`);
log(` Datei: ${path.resolve(opts.outFile)}`);
} finally {
await conn.end();
}
}
// --- import ---------------------------------------------------------------
async function cmdImport(args) {
const opts = normalizeOptions(args);
const inFile = args.in && args.in !== true ? String(args.in) : null;
if (!inFile) throw new Error('import braucht --in <DATEI> (oder --in - für stdin)');
const target = await openConnection(targetOptions(args));
try {
log(` Ziel: ${describeConn(targetOptions(args))}`);
await assertTargetReady(target);
let orgUid = null;
let manifest = null;
if (inFile !== '-') {
await fsp.access(inFile);
const head = await readHead(inFile, 16 * 1024);
orgUid = parseDumpOrg(head);
manifest = parseDumpManifest(head);
} else {
warn('--in - : Vorabprüfung von Organisation und Schema wird übersprungen');
}
if (manifest?.size) {
await checkTargetSchema(target, manifest);
log(` Schema-Check ok: ${[...manifest.keys()].join(', ')}`);
}
if (orgUid) {
const exists = await targetHasOrg(target, orgUid);
if (exists) {
if (opts.mode === 'abort') {
throw new Error(
`Organisation ${orgUid} existiert in ${targetOptions(args).database ?? process.env.DB_DATABASE} bereits.\n` +
` --mode skip -> nichts tun\n` +
` --mode replace -> vorhandene Organisation vorher entfernen (Backend-Stub!), dann einspielen\n` +
` --mode overwrite -> trotzdem einspielen (legt neue Versionen an; kann am PK scheitern)`,
);
}
if (opts.mode === 'skip') {
log(` Organisation ${orgUid} ist bereits vorhanden — nichts zu tun (--mode skip).`);
return;
}
if (opts.mode === 'replace') {
// Nur die Tabellen anfassen, die der Dump auch wieder füllt.
const purged = await purgeTargetOrg(
target,
orgUid,
opts,
manifest ? [...manifest.keys()] : null,
);
printPurgeReport(purged, opts.dryRun);
} else {
warn(`--mode overwrite: ${orgUid} ist vorhanden, wird zusätzlich eingespielt (neue Versionen).`);
}
}
}
const readable = inFile === '-' ? process.stdin : await openDumpReadable(inFile);
log('');
log(` spiele ${inFile} ein${opts.dryRun ? ' (dry-run)' : ''}`);
const res = await applyDump(target, readable, opts);
log(` fertig: ${res.count} Statements, ${human(res.bytes)}`);
} finally {
await target.end();
}
}
// --- copy -----------------------------------------------------------------
async function cmdCopy(args) {
const opts = normalizeOptions(args);
if (!opts.org) throw new Error('copy braucht --org <UUID>');
const source = await openConnection(sourceOptions(args));
const target = await openConnection(targetOptions(args));
try {
log(` Quelle: ${describeConn(sourceOptions(args))}`);
log(` Ziel: ${describeConn(targetOptions(args))}`);
await assertTargetReady(target);
const ctx = await buildContext(source, opts);
printScopeReport(ctx.report, opts);
if (!opts.dryRun) {
// Ziel-Schema gegen die tatsächlich geplanten Spalten prüfen.
const plan = planTables(ctx, opts);
await checkTargetSchema(target, new Map(plan.map((p) => [p.spec.name, p.cols.map((c) => c.name)])));
log(` Schema-Check ok: ${plan.map((p) => p.spec.name).join(', ')}`);
const exists = await targetHasOrg(target, ctx.org.uid);
if (exists) {
if (opts.mode === 'abort') {
throw new Error(
`Organisation ${ctx.org.uid} existiert im Ziel bereits.\n` +
` --mode skip -> nichts tun\n` +
` --mode replace -> vorhandene Organisation vorher entfernen (Backend-Stub!), dann einspielen\n` +
` --mode overwrite -> trotzdem einspielen`,
);
}
if (opts.mode === 'skip') {
log(` Organisation ${ctx.org.uid} ist im Ziel vorhanden — nichts zu tun.`);
return;
}
if (opts.mode === 'replace') {
const purged = await purgeTargetOrg(target, ctx.org.uid, opts, plan.map((p) => p.spec.name));
printPurgeReport(purged, false);
} else {
warn('--mode overwrite: Ziel enthält die Organisation schon.');
}
}
} else if (opts.mode === 'replace') {
// Im dry-run wird nichts geschrieben; trotzdem zeigen, was weggeräumt würde.
const plan = planTables(ctx, opts);
const purged = await purgeTargetOrg(target, ctx.org.uid, opts, plan.map((p) => p.spec.name));
printPurgeReport(purged, true);
}
const splitter = new StatementSplitter(async (stmt) => {
if (!opts.dryRun) await target.query(stmt);
});
log('');
log(` streame nach ${describeConn(targetOptions(args))} ...`);
const t0 = Date.now();
const stats = await exportTo(source, ctx, opts, (chunk) => splitter.push(chunk));
await splitter.end();
log(` fertig in ${secs(Date.now() - t0)}: ${stats.rows} Zeilen, ${splitter.count} Statements`);
} finally {
await source.end();
await target.end();
}
}
// ---------------------------------------------------------------------------
// 11. Einstieg
// ---------------------------------------------------------------------------
/**
* Secrets laden — aber explizite ENV-Vorgaben gewinnen.
*
* `loadSecretsFromVault()` schreibt seine Werte **bedingungslos** nach
* `process.env`, würde ein vorgesetztes `DB_HOST` also überschreiben. Deshalb
* wird die Umgebung vorher gesichert und danach wieder hergestellt:
*
* Treiber-Default < Vault (`GIT_SECRET_PATH`) < ENV < `--from-*` / `--to-*`
*
* So kann man eine abweichende Quelle bequem per `-e DB_HOST=…` mitgeben,
* ohne die Passwort-Flags auf der Kommandozeile zu brauchen.
*
* @returns {Promise<string[]>} Namen der Schlüssel, die Vorrang behalten haben
*/
async function loadSecretsPreservingEnv() {
const preset = { ...process.env };
await loadSecretsFromVault();
const kept = [];
for (const [k, v] of Object.entries(preset)) {
if (process.env[k] !== v) {
process.env[k] = v;
kept.push(k);
}
}
return kept.sort();
}
async function main() {
const argv = process.argv.slice(2);
const cmd = argv[0];
// Hilfe darf weder Vault lesen noch als unbekanntes Kommando enden.
if (!cmd || cmd === 'help' || cmd === '--help' || cmd === '-h') {
process.stdout.write(USAGE);
return;
}
const args = parseArgs(argv.slice(1));
if (args.help) {
process.stdout.write(USAGE);
return;
}
const kept = await loadSecretsPreservingEnv();
// Nur die Namen ausgeben, nie die Werte — und nur, wenn Vault sie überschreiben
// wollte. Das macht sichtbar, *warum* welche Verbindung benutzt wird.
if (kept.length) log(` ENV hat Vorrang vor Vault: ${kept.join(', ')}`);
switch (cmd) {
case 'inspect': return cmdInspect(args);
case 'export': return cmdExport(args);
case 'import': return cmdImport(args);
case 'copy': return cmdCopy(args);
default:
process.stdout.write(USAGE);
throw new Error(`Unbekanntes Kommando: ${cmd}`);
}
}
main()
.then(() => process.exit(0))
.catch((e) => {
process.stderr.write(`\nFEHLER: ${e?.message ?? e}\n`);
if (process.env.DEBUG) process.stderr.write((e?.stack ?? '') + '\n');
process.exit(1);
});