john
440 lines
20 KiB
JavaScript
440 lines
20 KiB
JavaScript
import { execFileSync } from 'node:child_process';
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import fs from 'node:fs';
|
|
import os from 'node:os';
|
|
import path from 'node:path';
|
|
|
|
export const CONTROL_SCHEMA = 'mindspace_control';
|
|
|
|
export function assertUserId(value) {
|
|
const userId = String(value ?? '').trim().toLowerCase();
|
|
if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(userId)) {
|
|
throw new Error('user id 必须是标准 UUID');
|
|
}
|
|
return userId;
|
|
}
|
|
|
|
export function quotePgIdentifier(value) {
|
|
return `"${String(value).replaceAll('"', '""')}"`;
|
|
}
|
|
|
|
export function deriveUserSpaceNames(value) {
|
|
const userId = assertUserId(value);
|
|
const compact = userId.replaceAll('-', '');
|
|
const short = compact.slice(0, 12);
|
|
return {
|
|
userId,
|
|
schemaName: `u_${compact}`,
|
|
ownerRole: `ms_u_${short}_owner`,
|
|
agentRole: `ms_u_${short}_agent`,
|
|
};
|
|
}
|
|
|
|
export function buildControlSchemaSql() {
|
|
return `
|
|
CREATE SCHEMA IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)};
|
|
REVOKE ALL ON SCHEMA ${quotePgIdentifier(CONTROL_SCHEMA)} FROM PUBLIC;
|
|
|
|
CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces (
|
|
user_id uuid PRIMARY KEY,
|
|
schema_name name NOT NULL UNIQUE,
|
|
owner_role name NOT NULL UNIQUE,
|
|
agent_role name NOT NULL UNIQUE,
|
|
backend_type text NOT NULL DEFAULT 'postgres' CHECK (backend_type IN ('sqlite', 'postgres')),
|
|
source_sqlite_path text,
|
|
migration_state text NOT NULL DEFAULT 'pending' CHECK (
|
|
migration_state IN ('pending', 'snapshot_created', 'schema_converted', 'data_imported', 'verified', 'shadow', 'cutover', 'rolled_back', 'failed')
|
|
),
|
|
schema_version integer NOT NULL DEFAULT 1,
|
|
quota_bytes bigint NOT NULL DEFAULT 104857600 CHECK (quota_bytes > 0),
|
|
created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
cutover_at timestamptz,
|
|
rollback_until timestamptz
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs (
|
|
id uuid PRIMARY KEY,
|
|
user_id uuid NOT NULL REFERENCES ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces(user_id) ON DELETE CASCADE,
|
|
source_path text NOT NULL,
|
|
source_size_bytes bigint NOT NULL DEFAULT 0,
|
|
source_sha256 text,
|
|
status text NOT NULL,
|
|
table_count integer NOT NULL DEFAULT 0,
|
|
source_row_count bigint NOT NULL DEFAULT 0,
|
|
target_row_count bigint NOT NULL DEFAULT 0,
|
|
detail_json jsonb NOT NULL DEFAULT '{}'::jsonb,
|
|
started_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
finished_at timestamptz
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.dataset_catalog (
|
|
user_id uuid NOT NULL REFERENCES ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces(user_id) ON DELETE CASCADE,
|
|
dataset_name text NOT NULL,
|
|
physical_table name NOT NULL,
|
|
config_json jsonb NOT NULL,
|
|
sensitivity text NOT NULL DEFAULT 'normal' CHECK (sensitivity IN ('normal', 'personal', 'credential', 'health')),
|
|
created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (user_id, dataset_name)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.audit_events (
|
|
id bigserial PRIMARY KEY,
|
|
user_id uuid,
|
|
actor_type text NOT NULL,
|
|
actor_id text,
|
|
action text NOT NULL,
|
|
object_type text,
|
|
object_name text,
|
|
statement_hash text,
|
|
affected_rows bigint,
|
|
duration_ms integer,
|
|
result text NOT NULL,
|
|
detail_json jsonb NOT NULL DEFAULT '{}'::jsonb,
|
|
created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
REVOKE ALL ON ALL TABLES IN SCHEMA ${quotePgIdentifier(CONTROL_SCHEMA)} FROM PUBLIC;
|
|
`.trim();
|
|
}
|
|
|
|
export function buildProvisionUserSql(userId, { sourceSqlitePath = null, quotaBytes = 104857600 } = {}) {
|
|
const names = deriveUserSpaceNames(userId);
|
|
const schema = quotePgIdentifier(names.schemaName);
|
|
const owner = quotePgIdentifier(names.ownerRole);
|
|
const agent = quotePgIdentifier(names.agentRole);
|
|
return {
|
|
...names,
|
|
statements: [
|
|
`DO $$ BEGIN CREATE ROLE ${owner} NOLOGIN; EXCEPTION WHEN duplicate_object THEN NULL; END $$`,
|
|
`DO $$ BEGIN CREATE ROLE ${agent} NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION; EXCEPTION WHEN duplicate_object THEN NULL; END $$`,
|
|
`DO $$ BEGIN
|
|
IF NOT EXISTS (
|
|
SELECT 1 FROM pg_auth_members am
|
|
JOIN pg_roles r ON r.oid = am.roleid
|
|
JOIN pg_roles m ON m.oid = am.member
|
|
WHERE r.rolname = '${names.ownerRole}' AND m.rolname = CURRENT_USER AND am.set_option
|
|
) THEN
|
|
EXECUTE format('GRANT %I TO %I', '${names.ownerRole}', CURRENT_USER);
|
|
END IF;
|
|
END $$`,
|
|
`CREATE SCHEMA IF NOT EXISTS ${schema} AUTHORIZATION ${owner}`,
|
|
`REVOKE ALL ON SCHEMA ${schema} FROM PUBLIC`,
|
|
`GRANT USAGE, CREATE ON SCHEMA ${schema} TO ${agent}`,
|
|
`ALTER ROLE ${agent} SET search_path = ${schema}, pg_catalog`,
|
|
`ALTER ROLE ${agent} SET statement_timeout = '30s'`,
|
|
`ALTER ROLE ${agent} SET lock_timeout = '5s'`,
|
|
`ALTER ROLE ${agent} SET idle_in_transaction_session_timeout = '60s'`,
|
|
`ALTER ROLE ${agent} SET work_mem = '4MB'`,
|
|
`ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${schema} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${agent}`,
|
|
`ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${schema} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${agent}`,
|
|
`INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces
|
|
(user_id, schema_name, owner_role, agent_role, source_sqlite_path, quota_bytes, updated_at)
|
|
VALUES ($1::uuid, $2::name, $3::name, $4::name, $5, $6, CURRENT_TIMESTAMP)
|
|
ON CONFLICT (user_id) DO UPDATE SET
|
|
source_sqlite_path = EXCLUDED.source_sqlite_path,
|
|
quota_bytes = EXCLUDED.quota_bytes,
|
|
updated_at = CURRENT_TIMESTAMP`,
|
|
],
|
|
params: [names.userId, names.schemaName, names.ownerRole, names.agentRole, sourceSqlitePath, quotaBytes],
|
|
};
|
|
}
|
|
|
|
export function mapSqliteTypeToPostgres(column) {
|
|
const declared = String(column.type ?? '').trim().toUpperCase();
|
|
if (Number(column.pk) === 1 && declared.includes('INT')) return 'bigint GENERATED BY DEFAULT AS IDENTITY';
|
|
if (declared.includes('TEXT') && isSqliteTimestampDefault(column.dflt_value)) return 'timestamptz';
|
|
if (declared.includes('INT')) return 'bigint';
|
|
if (declared.includes('REAL') || declared.includes('FLOA') || declared.includes('DOUB')) return 'double precision';
|
|
if (declared.includes('BLOB')) return 'bytea';
|
|
if (declared.includes('NUM') || declared.includes('DEC')) return 'numeric';
|
|
if (declared.includes('BOOL')) return 'boolean';
|
|
return 'text';
|
|
}
|
|
|
|
export function isSqliteTimestampDefault(value) {
|
|
return /^\(?datetime\(\s*'now'\s*(?:,\s*'(?:\+8 hours|localtime)'\s*)?\)\)?$/i.test(String(value ?? '').trim());
|
|
}
|
|
|
|
export function translateSqliteDefault(value) {
|
|
if (value == null) return null;
|
|
const input = String(value).trim();
|
|
if (/^NULL$/i.test(input)) return 'NULL';
|
|
if (/^[+-]?\d+(?:\.\d+)?$/.test(input)) return input;
|
|
if (/^CURRENT_TIMESTAMP$/i.test(input)) return 'CURRENT_TIMESTAMP';
|
|
if (/^\(?datetime\(\s*'now'\s*(?:,\s*'(?:\+8 hours|localtime)'\s*)?\)\)?$/i.test(input)) return 'CURRENT_TIMESTAMP';
|
|
if (/^".*"$/.test(input)) return `'${input.slice(1, -1).replaceAll("'", "''")}'`;
|
|
if (/^'.*'$/.test(input)) return input;
|
|
return null;
|
|
}
|
|
|
|
export function buildPostgresTableSql(schemaName, tableName, columns) {
|
|
if (!Array.isArray(columns) || columns.length === 0) throw new Error(`表 ${tableName} 没有字段`);
|
|
const primaryKeys = columns.filter((column) => Number(column.pk) > 0).sort((a, b) => Number(a.pk) - Number(b.pk));
|
|
const definitions = columns.map((column) => {
|
|
const parts = [quotePgIdentifier(column.name), mapSqliteTypeToPostgres(column)];
|
|
if (Number(column.notnull) === 1) parts.push('NOT NULL');
|
|
const translatedDefault = translateSqliteDefault(column.dflt_value);
|
|
if (translatedDefault != null) parts.push(`DEFAULT ${translatedDefault}`);
|
|
if (primaryKeys.length === 1 && primaryKeys[0].name === column.name) parts.push('PRIMARY KEY');
|
|
return parts.join(' ');
|
|
});
|
|
if (primaryKeys.length > 1) {
|
|
definitions.push(`PRIMARY KEY (${primaryKeys.map((column) => quotePgIdentifier(column.name)).join(', ')})`);
|
|
}
|
|
return `CREATE TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} (\n ${definitions.join(',\n ')}\n)`;
|
|
}
|
|
|
|
function stableObjectName(prefix, ...parts) {
|
|
const readable = `${prefix}_${parts.join('_')}`.replace(/[^a-zA-Z0-9_]+/g, '_').toLowerCase();
|
|
const hash = createHash('sha256').update(readable).digest('hex').slice(0, 8);
|
|
return `${readable.slice(0, 50)}_${hash}`;
|
|
}
|
|
|
|
export function extractSqliteChecks(createSql) {
|
|
const sql = String(createSql ?? '');
|
|
const checks = [];
|
|
const pattern = /\bCHECK\s*\(/ig;
|
|
let match;
|
|
while ((match = pattern.exec(sql))) {
|
|
const start = pattern.lastIndex;
|
|
let depth = 1;
|
|
let quote = '';
|
|
let index = start;
|
|
for (; index < sql.length && depth > 0; index += 1) {
|
|
const char = sql[index];
|
|
if (quote) {
|
|
if (char === quote && sql[index + 1] === quote) index += 1;
|
|
else if (char === quote) quote = '';
|
|
continue;
|
|
}
|
|
if (char === "'" || char === '"') quote = char;
|
|
else if (char === '(') depth += 1;
|
|
else if (char === ')') depth -= 1;
|
|
}
|
|
if (depth === 0) checks.push(sql.slice(start, index - 1).trim());
|
|
pattern.lastIndex = index;
|
|
}
|
|
return checks;
|
|
}
|
|
|
|
export function buildPostgresIndexSql(schemaName, tableName, index) {
|
|
if (!index.columns?.length) return null;
|
|
if (index.partial) return null;
|
|
const prefix = index.unique ? 'ux' : 'ix';
|
|
const name = stableObjectName(prefix, tableName, ...index.columns);
|
|
return `CREATE ${index.unique ? 'UNIQUE ' : ''}INDEX ${quotePgIdentifier(name)} ON ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} (${index.columns.map(quotePgIdentifier).join(', ')})`;
|
|
}
|
|
|
|
export function buildPostgresForeignKeySql(schemaName, tableName, foreignKey) {
|
|
const name = stableObjectName('fk', tableName, foreignKey.id, foreignKey.table);
|
|
const onUpdate = String(foreignKey.onUpdate ?? 'NO ACTION').toUpperCase();
|
|
const onDelete = String(foreignKey.onDelete ?? 'NO ACTION').toUpperCase();
|
|
const allowedActions = new Set(['NO ACTION', 'RESTRICT', 'CASCADE', 'SET NULL', 'SET DEFAULT']);
|
|
if (!allowedActions.has(onUpdate) || !allowedActions.has(onDelete)) throw new Error('SQLite 外键动作不受支持');
|
|
return `ALTER TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} ADD CONSTRAINT ${quotePgIdentifier(name)} FOREIGN KEY (${foreignKey.from.map(quotePgIdentifier).join(', ')}) REFERENCES ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(foreignKey.table)} (${foreignKey.to.map(quotePgIdentifier).join(', ')}) ON UPDATE ${onUpdate} ON DELETE ${onDelete}`;
|
|
}
|
|
|
|
export function buildPostgresCheckSql(schemaName, tableName, expression, index) {
|
|
if (/;|--|\/\*/.test(expression)) throw new Error(`CHECK 表达式包含不安全内容：${tableName}`);
|
|
const name = stableObjectName('ck', tableName, index);
|
|
return `ALTER TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} ADD CONSTRAINT ${quotePgIdentifier(name)} CHECK (${expression})`;
|
|
}
|
|
|
|
export function convertSqliteValueForPostgres(value, column) {
|
|
if (value == null) return null;
|
|
if (isSqliteTimestampDefault(column.dflt_value) && typeof value === 'string') {
|
|
const trimmed = value.trim();
|
|
if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d+)?$/.test(trimmed)) {
|
|
return `${trimmed.replace(' ', 'T')}+08:00`;
|
|
}
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function sqliteJson(sqliteBin, databasePath, sql) {
|
|
const output = execFileSync(sqliteBin, ['-readonly', '-json', databasePath, sql], {
|
|
encoding: 'utf8',
|
|
maxBuffer: 32 * 1024 * 1024,
|
|
}).trim();
|
|
return output ? JSON.parse(output) : [];
|
|
}
|
|
|
|
function quoteSqliteIdentifier(value) {
|
|
return `"${String(value).replaceAll('"', '""')}"`;
|
|
}
|
|
|
|
export function inspectSqliteDatabase(databasePath, { sqliteBin = 'sqlite3' } = {}) {
|
|
const resolved = path.resolve(databasePath);
|
|
if (!fs.statSync(resolved).isFile()) throw new Error(`SQLite 文件不存在：${resolved}`);
|
|
const integrity = sqliteJson(sqliteBin, resolved, 'PRAGMA integrity_check;');
|
|
if (integrity[0]?.integrity_check !== 'ok') throw new Error(`SQLite integrity_check 未通过：${resolved}`);
|
|
const tables = sqliteJson(
|
|
sqliteBin,
|
|
resolved,
|
|
"SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name;",
|
|
).map((table) => {
|
|
const columns = sqliteJson(sqliteBin, resolved, `PRAGMA table_info(${quoteSqliteIdentifier(table.name)});`);
|
|
const rows = sqliteJson(sqliteBin, resolved, `SELECT * FROM ${quoteSqliteIdentifier(table.name)};`);
|
|
const indexes = sqliteJson(sqliteBin, resolved, `PRAGMA index_list(${quoteSqliteIdentifier(table.name)});`).map((index) => ({
|
|
name: index.name,
|
|
unique: Boolean(index.unique),
|
|
origin: index.origin,
|
|
partial: Boolean(index.partial),
|
|
columns: sqliteJson(sqliteBin, resolved, `PRAGMA index_info(${quoteSqliteIdentifier(index.name)});`).map((item) => item.name),
|
|
})).filter((index) => index.origin !== 'pk');
|
|
const foreignKeyRows = sqliteJson(sqliteBin, resolved, `PRAGMA foreign_key_list(${quoteSqliteIdentifier(table.name)});`);
|
|
const foreignKeys = [...new Set(foreignKeyRows.map((item) => item.id))].map((id) => {
|
|
const items = foreignKeyRows.filter((item) => item.id === id).sort((a, b) => a.seq - b.seq);
|
|
return {
|
|
id,
|
|
table: items[0].table,
|
|
from: items.map((item) => item.from),
|
|
to: items.map((item) => item.to),
|
|
onUpdate: items[0].on_update,
|
|
onDelete: items[0].on_delete,
|
|
};
|
|
});
|
|
return { ...table, columns, rows, indexes, foreignKeys, checks: extractSqliteChecks(table.sql) };
|
|
});
|
|
return { path: resolved, sizeBytes: fs.statSync(resolved).size, tables };
|
|
}
|
|
|
|
export function createSqliteSnapshot(sourcePath, { sqliteBin = 'sqlite3' } = {}) {
|
|
const target = path.join(os.tmpdir(), `mindspace-userdata-${Date.now()}-${process.pid}.sqlite`);
|
|
if (/\s/.test(target)) throw new Error('SQLite snapshot 临时路径不能包含空格');
|
|
execFileSync(sqliteBin, [path.resolve(sourcePath), `.backup ${target}`], { stdio: 'pipe' });
|
|
// A standalone snapshot must not depend on the source WAL/SHM files.
|
|
execFileSync(sqliteBin, [target, 'PRAGMA journal_mode=DELETE;'], { stdio: 'pipe' });
|
|
return target;
|
|
}
|
|
|
|
export async function provisionUserSpace(client, userId, options = {}) {
|
|
const provision = buildProvisionUserSql(userId, options);
|
|
await client.query('BEGIN');
|
|
try {
|
|
for (let index = 0; index < provision.statements.length; index += 1) {
|
|
const params = index === provision.statements.length - 1 ? provision.params : [];
|
|
await client.query(provision.statements[index], params);
|
|
}
|
|
await client.query('COMMIT');
|
|
return provision;
|
|
} catch (error) {
|
|
await client.query('ROLLBACK');
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function migrateSqliteSnapshot(client, snapshot, userId, { replaceShadow = false, sourcePath = snapshot.path } = {}) {
|
|
const names = deriveUserSpaceNames(userId);
|
|
const owner = quotePgIdentifier(names.ownerRole);
|
|
const agent = quotePgIdentifier(names.agentRole);
|
|
const migrationId = randomUUID();
|
|
let sourceRows = 0;
|
|
let targetRows = 0;
|
|
await client.query('BEGIN');
|
|
try {
|
|
const stateResult = await client.query(
|
|
`SELECT migration_state FROM ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces WHERE user_id = $1::uuid FOR UPDATE`,
|
|
[names.userId],
|
|
);
|
|
const migrationState = stateResult.rows[0]?.migration_state;
|
|
if (migrationState === 'cutover') throw new Error('用户空间已经 cutover，禁止迁移器覆盖');
|
|
const existingResult = await client.query(
|
|
`SELECT count(*)::integer AS count FROM pg_catalog.pg_tables WHERE schemaname = $1`,
|
|
[names.schemaName],
|
|
);
|
|
if (Number(existingResult.rows[0]?.count ?? 0) > 0 && !replaceShadow) {
|
|
throw new Error('目标影子空间已有表；如确认重建，显式使用 --replace-shadow');
|
|
}
|
|
await client.query(
|
|
`INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs
|
|
(id, user_id, source_path, source_size_bytes, status, started_at)
|
|
VALUES ($1::uuid, $2::uuid, $3, $4, 'importing', CURRENT_TIMESTAMP)`,
|
|
[migrationId, names.userId, sourcePath, snapshot.sizeBytes],
|
|
);
|
|
for (const table of snapshot.tables) {
|
|
await client.query(`DROP TABLE IF EXISTS ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} CASCADE`);
|
|
await client.query(buildPostgresTableSql(names.schemaName, table.name, table.columns));
|
|
await client.query(`ALTER TABLE ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} OWNER TO ${owner}`);
|
|
await client.query(
|
|
`GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} TO ${agent}`,
|
|
);
|
|
for (const row of table.rows) {
|
|
const columns = Object.keys(row);
|
|
const params = columns.map((_, index) => `$${index + 1}`);
|
|
await client.query(
|
|
`INSERT INTO ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} (${columns.map(quotePgIdentifier).join(', ')}) VALUES (${params.join(', ')})`,
|
|
columns.map((column) => convertSqliteValueForPostgres(
|
|
row[column],
|
|
table.columns.find((item) => item.name === column) ?? {},
|
|
)),
|
|
);
|
|
}
|
|
for (const column of table.columns.filter(
|
|
(item) => Number(item.pk) === 1 && String(item.type ?? '').toUpperCase().includes('INT'),
|
|
)) {
|
|
const qualifiedTable = `${names.schemaName}.${table.name}`;
|
|
await client.query(
|
|
`SELECT setval(
|
|
pg_get_serial_sequence($1, $2),
|
|
GREATEST(COALESCE(MAX(${quotePgIdentifier(column.name)}), 0), 1),
|
|
COALESCE(MAX(${quotePgIdentifier(column.name)}), 0) > 0
|
|
)
|
|
FROM ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)}`,
|
|
[qualifiedTable, column.name],
|
|
);
|
|
}
|
|
sourceRows += table.rows.length;
|
|
const count = await client.query(
|
|
`SELECT count(*)::bigint AS count FROM ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)}`,
|
|
);
|
|
targetRows += Number(count.rows[0].count);
|
|
}
|
|
for (const table of snapshot.tables) {
|
|
for (const index of table.indexes ?? []) {
|
|
const sql = buildPostgresIndexSql(names.schemaName, table.name, index);
|
|
if (sql) await client.query(sql);
|
|
}
|
|
for (const foreignKey of table.foreignKeys ?? []) {
|
|
await client.query(buildPostgresForeignKeySql(names.schemaName, table.name, foreignKey));
|
|
}
|
|
for (let index = 0; index < (table.checks ?? []).length; index += 1) {
|
|
await client.query(buildPostgresCheckSql(names.schemaName, table.name, table.checks[index], index));
|
|
}
|
|
}
|
|
await client.query(`GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA ${quotePgIdentifier(names.schemaName)} TO ${agent}`);
|
|
const registry = snapshot.tables.find((table) => table.name === '__page_data_datasets');
|
|
for (const row of registry?.rows ?? []) {
|
|
await client.query(
|
|
`INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.dataset_catalog
|
|
(user_id, dataset_name, physical_table, config_json, updated_at)
|
|
VALUES ($1::uuid, $2, $3::name, $4::jsonb, CURRENT_TIMESTAMP)
|
|
ON CONFLICT (user_id, dataset_name) DO UPDATE SET
|
|
physical_table = EXCLUDED.physical_table,
|
|
config_json = EXCLUDED.config_json,
|
|
updated_at = CURRENT_TIMESTAMP`,
|
|
[names.userId, row.name, row.table_name, row.config_json],
|
|
);
|
|
}
|
|
await client.query(
|
|
`UPDATE ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces
|
|
SET migration_state = 'shadow', updated_at = CURRENT_TIMESTAMP
|
|
WHERE user_id = $1::uuid`,
|
|
[names.userId],
|
|
);
|
|
await client.query(
|
|
`UPDATE ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs
|
|
SET status = 'verified', table_count = $2, source_row_count = $3,
|
|
target_row_count = $4, finished_at = CURRENT_TIMESTAMP
|
|
WHERE id = $1::uuid`,
|
|
[migrationId, snapshot.tables.length, sourceRows, targetRows],
|
|
);
|
|
await client.query('COMMIT');
|
|
return { migrationId, ...names, tableCount: snapshot.tables.length, sourceRows, targetRows };
|
|
} catch (error) {
|
|
await client.query('ROLLBACK');
|
|
throw error;
|
|
}
|
|
}
|