318 lines
12 KiB
JavaScript
318 lines
12 KiB
JavaScript
// Registered Saltcorn tables for plugin state.
|
|
//
|
|
// A plugin keeps backed-up state in REGISTERED tables (_sc_tables + _sc_fields,
|
|
// auto `id` PK) so it rides Saltcorn backup/restore. Making one is harder than
|
|
// Table.create:
|
|
// - Registration is decided from _sc_tables in the DB, never this process's
|
|
// table cache: right after a live install (or another worker's first boot)
|
|
// the cache is stale, and Table.create on an existing table throws.
|
|
// - A registered table is never dropped.
|
|
// - A STRAY physical table of the same name that is not registered (a legacy
|
|
// raw table from an older plugin version, or an orphan left by a failed or
|
|
// racing install) is adopted: its rows are read, the table dropped, the
|
|
// registered table created and the rows copied back by column name (ids
|
|
// preserved), all in ONE transaction -- an empty orphan is simply replaced,
|
|
// and a failure rolls back to the untouched stray table.
|
|
// - Concurrently booting workers are serialized: Postgres takes a
|
|
// transaction-scoped advisory lock per plugin + tenant (CREATE ... IF NOT
|
|
// EXISTS is not safe against a concurrent create); SQLite takes the write
|
|
// lock up front (BEGIN IMMEDIATE, retried while another process holds it).
|
|
// Everything is tenant-schema-qualified on Postgres.
|
|
|
|
const { ADMIN_ROLE_ID } = require("./adminGate");
|
|
|
|
// SQLite: how often / how long to wait for another process's write lock (on top
|
|
// of the driver's own busy timeout per attempt).
|
|
const SQLITE_BUSY_ATTEMPTS = 20;
|
|
const SQLITE_BUSY_DELAY_MS = 250;
|
|
|
|
|
|
const getDb = () => {
|
|
return require("@saltcorn/data/db");
|
|
};
|
|
|
|
|
|
const sleep = (ms) => {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
};
|
|
|
|
|
|
// The schema-qualified, quoted SQL reference of a table in the current tenant.
|
|
const tableRef = (name) => {
|
|
return `${getDb().getTenantSchemaPrefix()}"${name}"`;
|
|
};
|
|
|
|
|
|
// Whether a table named `name` physically exists in this tenant's schema
|
|
// (registered or not).
|
|
const physicalTableExists = async (name) => {
|
|
const db = getDb();
|
|
const rs = db.isSQLite
|
|
? await db.query("SELECT name FROM sqlite_master WHERE type = 'table' AND name = $1", [name])
|
|
: await db.query("SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2", [db.getTenantSchema(), name]);
|
|
return rs.rows.length > 0;
|
|
};
|
|
|
|
|
|
// The column names a physical table has in this tenant's schema.
|
|
const physicalColumns = async (name) => {
|
|
const db = getDb();
|
|
if (db.isSQLite) {
|
|
return new Set((await db.query(`PRAGMA table_info("${name}")`)).rows.map((r) => r.name));
|
|
}
|
|
const rs = await db.query(
|
|
"SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
|
|
[db.getTenantSchema(), name]
|
|
);
|
|
return new Set(rs.rows.map((r) => r.column_name));
|
|
};
|
|
|
|
|
|
const isSqliteBusy = (e) => {
|
|
return !!e && (e.code === "SQLITE_BUSY" || /SQLITE_BUSY/.test(String(e.message)));
|
|
};
|
|
|
|
|
|
// Run a statement, retrying while another SQLite process holds the lock.
|
|
const sqliteWithRetry = async (sql) => {
|
|
const db = getDb();
|
|
for (let attempt = 1; ; attempt++) {
|
|
try {
|
|
await db.query(sql);
|
|
return;
|
|
} catch (e) {
|
|
if (!isSqliteBusy(e) || attempt >= SQLITE_BUSY_ATTEMPTS) {
|
|
throw e;
|
|
}
|
|
await sleep(SQLITE_BUSY_DELAY_MS);
|
|
}
|
|
}
|
|
};
|
|
|
|
|
|
// Run fn in ONE transaction covering every db.* call it makes, serialized
|
|
// against every other process doing schema work for `owner` in this tenant.
|
|
// Saltcorn's SQLite withTransaction is a no-op, so SQLite gets an explicit
|
|
// BEGIN IMMEDIATE; on Postgres withTransaction only routes queries through its
|
|
// client inside a request context, so one is established when absent (e.g.
|
|
// during onLoad). Never nest it.
|
|
const inSchemaTransaction = async (owner, fn) => {
|
|
const db = getDb();
|
|
if (db.isSQLite) {
|
|
await sqliteWithRetry("BEGIN IMMEDIATE");
|
|
try {
|
|
const result = await fn();
|
|
await sqliteWithRetry("COMMIT");
|
|
return result;
|
|
} catch (e) {
|
|
await db.query("ROLLBACK");
|
|
throw e;
|
|
}
|
|
}
|
|
const run = () => db.withTransaction(async () => {
|
|
await db.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`${owner}:schema:${db.getTenantSchema()}`]);
|
|
return await fn();
|
|
});
|
|
return db.getRequestContext() ? await run() : await db.runWithTenant(db.getTenantSchema(), run);
|
|
};
|
|
|
|
|
|
const registeredRow = async (name) => {
|
|
return await getDb().selectMaybeOne("_sc_tables", { name: name });
|
|
};
|
|
|
|
|
|
// What a registered table lacks: field defs without an _sc_fields row, and
|
|
// registered fields whose physical column went missing (a damaged table).
|
|
const missingParts = async (tableRow, spec) => {
|
|
const db = getDb();
|
|
const registered = new Set((await db.select("_sc_fields", { table_id: tableRow.id })).map((f) => f.name));
|
|
const columns = await physicalColumns(spec.name);
|
|
return {
|
|
fields: spec.fields.filter((f) => !registered.has(f.name)),
|
|
columns: spec.fields.filter((f) => registered.has(f.name) && !columns.has(f.name))
|
|
};
|
|
};
|
|
|
|
|
|
// Stray physical tables whose rows still have to be adopted.
|
|
const presentLegacyTables = async (spec) => {
|
|
const present = [];
|
|
for (const name of spec.legacyTables || []) {
|
|
if (await physicalTableExists(name)) {
|
|
present.push(name);
|
|
}
|
|
}
|
|
return present;
|
|
};
|
|
|
|
|
|
const needsWork = async (spec) => {
|
|
const tableRow = await registeredRow(spec.name);
|
|
if (!tableRow) {
|
|
return true;
|
|
}
|
|
const missing = await missingParts(tableRow, spec);
|
|
return missing.fields.length > 0 || missing.columns.length > 0 || (await presentLegacyTables(spec)).length > 0;
|
|
};
|
|
|
|
|
|
const createField = async (table, f) => {
|
|
const Field = require("@saltcorn/data/models/field");
|
|
await Field.create({
|
|
table: table,
|
|
name: f.name,
|
|
label: f.label || f.name,
|
|
type: f.type,
|
|
is_unique: !!f.is_unique,
|
|
required: !!f.required,
|
|
attributes: f.attributes || {}
|
|
});
|
|
};
|
|
|
|
|
|
// Read a stray table's rows, then drop it (inside the caller's transaction).
|
|
const takeRows = async (name) => {
|
|
const db = getDb();
|
|
const rows = (await db.query(`SELECT * FROM ${tableRef(name)} ORDER BY 1`)).rows;
|
|
await db.query(`DROP TABLE ${tableRef(name)}`);
|
|
return rows;
|
|
};
|
|
|
|
|
|
// Postgres: after inserting explicit ids, move the serial past them.
|
|
const bumpIdSequence = async (name) => {
|
|
const db = getDb();
|
|
if (db.isSQLite) {
|
|
return;
|
|
}
|
|
const qualified = tableRef(name);
|
|
await db.query(`SELECT setval(pg_get_serial_sequence('${qualified.replace(/'/g, "''")}', 'id'), (SELECT COALESCE(MAX(id), 1) FROM ${qualified}))`);
|
|
};
|
|
|
|
|
|
// Copy adopted rows into the registered table by column name. A row is first
|
|
// passed through rowTransform (null drops it); a row whose keyField value is
|
|
// already present is skipped; a legacy id is kept unless already taken.
|
|
const insertRows = async (spec, rows) => {
|
|
const db = getDb();
|
|
let keptIds = false;
|
|
for (const r of rows) {
|
|
const row = spec.rowTransform ? await spec.rowTransform(r) : r;
|
|
if (!row) {
|
|
continue;
|
|
}
|
|
const clean = {};
|
|
for (const f of spec.fields) {
|
|
if (row[f.name] !== undefined && row[f.name] !== null) {
|
|
clean[f.name] = row[f.name];
|
|
}
|
|
}
|
|
if (spec.keyField && clean[spec.keyField] !== undefined && (await db.selectMaybeOne(spec.name, { [spec.keyField]: clean[spec.keyField] }))) {
|
|
continue;
|
|
}
|
|
if (r.id !== undefined && r.id !== null && !(await db.selectMaybeOne(spec.name, { id: r.id }))) {
|
|
clean.id = r.id;
|
|
keptIds = true;
|
|
}
|
|
await db.insert(spec.name, clean, { noid: true });
|
|
}
|
|
if (keptIds) {
|
|
await bumpIdSequence(spec.name);
|
|
}
|
|
};
|
|
|
|
|
|
// The transaction body: re-check everything (another worker may have done it
|
|
// while this one waited for the lock), then register / heal / adopt.
|
|
const registerLocked = async (spec) => {
|
|
const db = getDb();
|
|
const Table = require("@saltcorn/data/models/table");
|
|
const Field = require("@saltcorn/data/models/field");
|
|
const { getState } = require("@saltcorn/data/db/state");
|
|
const scope = spec.createScope || ((fn) => fn());
|
|
const tableRow = await registeredRow(spec.name);
|
|
let adopted = [];
|
|
if (!tableRow) {
|
|
if (await physicalTableExists(spec.name)) {
|
|
adopted = await takeRows(spec.name);
|
|
}
|
|
await scope(async () => {
|
|
const table = await Table.create(spec.name, {
|
|
min_role_read: spec.minRoleRead || ADMIN_ROLE_ID,
|
|
min_role_write: spec.minRoleWrite || ADMIN_ROLE_ID
|
|
});
|
|
for (const f of spec.fields) {
|
|
await createField(table, f);
|
|
}
|
|
});
|
|
} else {
|
|
const missing = await missingParts(tableRow, spec);
|
|
if (missing.fields.length > 0) {
|
|
// Table.findOne reads the cache, which may predate this table.
|
|
await getState().refresh_tables(true);
|
|
const table = Table.findOne({ name: spec.name });
|
|
await scope(async () => {
|
|
for (const f of missing.fields) {
|
|
await createField(table, f);
|
|
}
|
|
});
|
|
}
|
|
for (const f of missing.columns) {
|
|
const sqlType = new Field({ name: f.name, type: f.type, attributes: f.attributes || {} }).sql_type;
|
|
await db.query(`ALTER TABLE ${tableRef(spec.name)} ADD COLUMN "${f.name}" ${sqlType}`);
|
|
}
|
|
}
|
|
for (const legacy of await presentLegacyTables(spec)) {
|
|
adopted = adopted.concat(await takeRows(legacy));
|
|
}
|
|
await insertRows(spec, adopted);
|
|
return !tableRow;
|
|
};
|
|
|
|
|
|
// Make spec.name a registered table with an auto `id` PK plus spec.fields.
|
|
// Idempotent; returns true when this call registered it.
|
|
// spec = {
|
|
// owner plugin name (lock namespace, error messages)
|
|
// name the table
|
|
// fields [{ name, type, label?, is_unique?, required?, attributes? }]
|
|
// minRoleRead / minRoleWrite default: admin
|
|
// keyField optional logical key: an adopted row whose value is already
|
|
// present is skipped (makes repeated adoption idempotent)
|
|
// rowTransform optional async (strayRow) => row | null (null drops it)
|
|
// legacyTables optional further stray tables (e.g. a renamed-aside copy
|
|
// from an interrupted older migration) whose rows are adopted
|
|
// into the registered table and which are then dropped
|
|
// createScope optional (fn) => runs fn around Table.create / Field.create
|
|
// (e.g. to keep the plugin's own CRUD wraps from journaling)
|
|
// }
|
|
const ensureRegisteredTable = async (spec) => {
|
|
if (!(await needsWork(spec))) {
|
|
return false;
|
|
}
|
|
const { getState } = require("@saltcorn/data/db/state");
|
|
try {
|
|
return await inSchemaTransaction(spec.owner, () => registerLocked(spec));
|
|
} catch (e) {
|
|
// Done by another worker meanwhile (e.g. the SQLite lock wait ran out)?
|
|
if (!(await needsWork(spec))) {
|
|
return false;
|
|
}
|
|
throw new Error(`${spec.owner}: could not register table ${spec.name}: ${e.message}`);
|
|
} finally {
|
|
// Table.create / Field.create skip their cache refresh inside a Postgres
|
|
// transaction, and on SQLite a rolled-back create already refreshed it
|
|
// (a phantom table): reload here and tell the other workers.
|
|
await getState().refresh_tables();
|
|
}
|
|
};
|
|
|
|
|
|
module.exports = {
|
|
tableRef,
|
|
physicalTableExists,
|
|
physicalColumns,
|
|
inSchemaTransaction,
|
|
ensureRegisteredTable
|
|
};
|