// The site-sync admin operations, transport-free. // // The admin UI (routes.js) and the MCP tools (mcpTools.js, served by the // saltcorn-mcp plugin) are both thin adapters over these, so the rules live in one // place: which settings values are valid, which peer a reference names, how a pull // is orchestrated (plugin parity, signed state fetch, plan build), what a plan // allows now and which tables may take which data mode. That only one apply or // rollback runs on a site at a time is enforced below these, by the site lock the // apply and rollback libraries take (lib/siteLock.js), so every caller -- the // machine API's push-apply included -- shares it. // // A failure the caller can act on throws OpError(message, extra); each adapter maps // it to its own shape (a flash, the apply-failure page, a tool error, a signed // reply). Library steps whose Error messages are written for the admin (what the UI // has always flashed) are rethrown as OpError with the same message. const Table = require("@saltcorn/data/models/table"); const { getSessionSecret } = require("./common/sessionKey"); const { parseSecretHex } = require("./common/peerStore"); const { localPluginList, comparePluginLists } = require("./common/pluginParity"); const { PLUGIN_NAME, PLUGIN_VERSION, SYNC_MODES, DELETE_POLICY, APPLY_POLICY, PLAN_STATUS, DATA_MODES, isInfraName, isLockedTable, isSyncedMode } = require("./constants"); const { requireEnv, setEnvFields } = require("./env"); const peers = require("./peers"); const { peerJson, API_PATHS } = require("./transport"); const capture = require("./capture"); const planLib = require("./plan"); const applyLib = require("./apply"); const rollback = require("./rollback"); const rowsLib = require("./rows"); const siteLock = require("./siteLock"); // Plans listed when no limit is given. const PLAN_LIST_DEFAULT = 20; // Settings choices: one list per policy drives the dashboard form, the MCP tool // schema and updateSettings' validation. const SETTING_OPTIONS = { sync_mode: [ { label: "mirror (prune entities the source dropped)", value: SYNC_MODES.MIRROR }, { label: "additive (never prune)", value: SYNC_MODES.ADDITIVE }, ], on_delete: [ { label: "confirm (drop only tables typed into the apply form, plus the confirmation tick)", value: DELETE_POLICY.CONFIRM }, { label: "refuse (never drop a table/field with data)", value: DELETE_POLICY.REFUSE }, { label: "auto (drop tables typed into the apply form, no confirmation tick)", value: DELETE_POLICY.AUTO }, ], on_apply: [ { label: "confirm (only this admin UI applies; remote push-apply is refused)", value: APPLY_POLICY.CONFIRM }, { label: "auto (paired peers may also push-apply remotely; never drops data or renames tables)", value: APPLY_POLICY.AUTO }, ] }; class OpError extends Error { constructor(message, extra) { super(message); this.name = "OpError"; this.extra = extra; } } // Run a library step; an Error it throws (written for the admin: what the UI // flashes) becomes an OpError with the same message. const adminStep = async (fn) => { try { return await fn(); } catch (e) { throw e instanceof OpError ? e : new OpError(e.message); } }; // ---------------- Environment ---------------- // Return `value` if it is one of the options' values; otherwise throw. const pickOption = (field, value, options) => { const hit = options.find((o) => o.value === value); if (!hit) { throw new OpError(`invalid ${field}: ${value}`); } return hit.value; }; // Normalizer (and validator) of each setting updateSettings accepts. const SETTINGS = { env_label: (v) => String(v === null || v === undefined ? "" : v).trim() || null, sync_mode: (v) => pickOption("sync mode", v, SETTING_OPTIONS.sync_mode), on_delete: (v) => pickOption("delete policy", v, SETTING_OPTIONS.on_delete), on_apply: (v) => pickOption("apply policy", v, SETTING_OPTIONS.on_apply), require_tls: (v) => (v ? 1 : 0) }; // Whether the session secret the peer-secret key derives from is configured (its // value is never exposed). const sessionSecretConfigured = () => { try { getSessionSecret(PLUGIN_NAME); return true; } catch (e) { return false; } }; // The environment, its peers and synced tables at a glance. const status = async () => { const env = await requireEnv(); const modes = Object.values(await rowsLib.modeMap()); const syncedTables = {}; for (const mode of Object.values(DATA_MODES).filter(isSyncedMode)) { syncedTables[mode] = modes.filter((m) => m === mode).length; } return { env: env, version: PLUGIN_VERSION, peerCount: (await peers.listPeers()).length, syncedTables: syncedTables, sessionSecretConfigured: sessionSecretConfigured(), applyLock: await applyLockView() }; }; // The site lock for an admin: { state: running, holder } while an apply or // rollback holds a live lease; { state: interrupted, plans } when plans were left // in flight by a process that stopped (each with whether a restore point can roll // it back; the next apply or rollback recovers them); else { state: idle }. const applyLockView = async () => { const lock = await siteLock.lockState(); if (lock && lock.live) { return { state: siteLock.LOCK_STATES.RUNNING, holder: lock }; } const restoring = await rollback.restorePointIds(); const plans = []; for (const status of siteLock.IN_FLIGHT) { for (const r of await planLib.listPlans({ status, limit: planLib.PLAN_RETENTION })) { plans.push({ plan_id: r.plan_id, status: r.status, direction: r.direction, has_restore_point: restoring.has(r.plan_id) }); } } return plans.length > 0 ? { state: siteLock.LOCK_STATES.INTERRUPTED, plans } : { state: siteLock.LOCK_STATES.IDLE }; }; // Change some settings (a partial patch of SETTINGS keys; undefined = keep). // Returns the updated environment. const updateSettings = async (patch) => { const fields = {}; for (const [key, value] of Object.entries(patch || {})) { if (value === undefined) { continue; } if (!Object.prototype.hasOwnProperty.call(SETTINGS, key)) { throw new OpError(`unknown setting: ${key}`); } fields[key] = SETTINGS[key](value); } if (Object.keys(fields).length === 0) { throw new OpError(`no settings given (any of ${Object.keys(SETTINGS).join(", ")})`); } return await setEnvFields(fields); }; // ---------------- Peers ---------------- const peerName = (peer) => { return peer.label || peer.env_id; }; // A peer by reference: its peer_id (a number, or a string of digits), else its // env_id, else its label (which must then be unique). const resolvePeer = async (ref) => { const key = String(ref === null || ref === undefined ? "" : ref).trim(); if (!key) { throw new OpError("peer required"); } const list = await peers.listPeers(); const found = list.find((p) => String(p.peer_id) === key) || list.find((p) => p.env_id === key); if (found) { return found; } const labelled = list.filter((p) => p.label === key); if (labelled.length !== 1) { throw new OpError(labelled.length ? `several peers are labelled "${key}"; name it by id or env_id` : `unknown peer "${key}"`); } return labelled[0]; }; // Pair with a peer. Returns { peer, secret (hex), generated }: the plaintext shared // secret exists only here (the admin copies it to the other side); it is generated // unless existingSecret (64 hex characters) was given. const addPeer = async ({ envId, label, baseUrl, requireTls, existingSecret }) => { return await adminStep(async () => { const pasted = parseSecretHex(existingSecret); const { peer, secret } = await peers.addPeer({ envId: String(envId || "").trim(), label: String(label || "").trim() || null, baseUrl: String(baseUrl || "").trim(), requireTls: !!requireTls, existingSecret: pasted }); return { peer, secret: secret.toString("hex"), generated: !pasted }; }); }; // Replace a peer's shared secret: with secretHex (the value the OTHER side rotated // to and displayed) or, when blank, a generated one this side displays. Returns // { peer, secret (hex), generated }. const rotatePeerSecret = async (ref, secretHex) => { const peer = await resolvePeer(ref); return await adminStep(async () => { const pasted = parseSecretHex(secretHex); const { secret } = await peers.rotatePeerSecret(peer.peer_id, pasted); return { peer, secret: secret.toString("hex"), generated: !pasted }; }); }; // Unpair a peer. Its plans go with it (a reused peer_id must never inherit them), // except those that can still be rolled back: kept, detached from any peer. // Returns { peer, keptForRollback }. const deletePeer = async (ref) => { const peer = await resolvePeer(ref); const keptForRollback = await rollback.detachRollbackablePlans(peer.peer_id); await peers.deletePeer(peer.peer_id); return { peer, keptForRollback }; }; // The peer's signed health answer ({ plugin, version, env_id, label, plugins }). const fetchPeerHealth = async (peerRow, env) => { return await peerJson(peerRow, env.env_id, "GET", API_PATHS.HEALTH, null); }; // Plugin parity warnings from a health answer (empty when both sides match). const parityWarnings = (localPlugins, health) => { if (!Array.isArray(health.plugins)) { return ["peer's health response carries no plugin list"]; } return comparePluginLists(localPlugins, health.plugins); }; // Compare local plugin list with peer's. Returns array of human-readable // warning strings (empty if all match). Best-effort: if the peer's health // endpoint is unreachable or errors, returns a single warning and lets the caller // proceed. Plugin PARITY (presence + version) matters for state sync: applying a // snapshot that references a plugin the peer lacks will fail -- but we never push // plugin CONFIG (that would leak deploy_private_key), only check parity. const diffPluginsWithPeer = async (peerRow, env, localPlugins) => { let health; try { health = await fetchPeerHealth(peerRow, env); } catch (e) { return [`could not check peer plugin list: ${e.message}`]; } return parityWarnings(localPlugins, health); }; // Whether a peer answers, as whom, at which version, and its plugin parity. // An unreachable or unauthenticated peer is a result (reachable: false), not an // error. const peerHealth = async (ref) => { const peer = await resolvePeer(ref); const env = await requireEnv(); let health; try { health = await fetchPeerHealth(peer, env); } catch (e) { return { peer, reachable: false, error: e.message }; } return { peer, reachable: true, health, parityWarnings: parityWarnings(await localPluginList(), health) }; }; // Target-side helper: pull a peer's desired-state pack over the signed transport. // Returns { hash, pack, rows, env_id }; throws on transport/authz failure or when // the (signed) answer does not come from the paired env. const fetchPeerState = async (peerRow, env) => { const body = await peerJson(peerRow, env.env_id, "GET", API_PATHS.STATE, null); if (!body.pack || body.env_id !== peerRow.env_id) { throw new Error("peer state response has no pack or names the wrong env_id"); } return { hash: body.hash, pack: body.pack, rows: body.rows || {}, env_id: body.env_id }; }; // ---------------- Plans ---------------- // Pull a peer's current desired state and persist a plan for making THIS site // match it. Nothing changes on either site. Returns { plan_id, plan, peer }. const pullPlan = async (ref) => { const peer = await resolvePeer(ref); return await adminStep(async () => { const env = await requireEnv(); const warnings = await diffPluginsWithPeer(peer, env, await localPluginList()); const state = await fetchPeerState(peer, env); const built = await planLib.buildPlan({ peerRow: peer, env, desired: state.pack, desiredHash: state.hash, sourceEnvId: state.env_id, direction: "pull", extraWarnings: warnings, sourceRows: state.rows }); return { ...built, peer }; }); }; const requirePlanId = (planId) => { const id = String(planId || "").trim(); if (!id) { throw new OpError("plan_id required"); } return id; }; // What can be done with a plan (row: its status, claimed_from + precondition hash) // now: apply it (its status allows it and this site has not changed since its // preview -- else the drift check refuses it: stale), and roll it back (a restore // point exists and its status allows it). stale is null when the plan cannot be // applied anyway. interrupted = in flight (applying / rolling_back) while no live // lease holds the site lock (lock: siteLock.lockState()): the process running it // stopped, so it is judged by the status the next apply or rollback recovers it to. const planStanding = (row, hasRestorePoint, currentHash, lock) => { const interrupted = siteLock.IN_FLIGHT.includes(row.status) && !(lock && lock.live); const status = interrupted ? siteLock.recoveredStatus(row) : row.status; const applicable = applyLib.APPLY_FROM.includes(status); const stale = applicable ? currentHash !== row.precondition_hash : null; return { has_restore_point: hasRestorePoint, can_apply: applicable && !stale, can_rollback: hasRestorePoint && rollback.ROLLBACK_FROM.includes(status), stale: stale, interrupted: interrupted }; }; // The most recent plans, newest first (status: only those; limit: 1 to // PLAN_RETENTION, default PLAN_LIST_DEFAULT), each with its peer and standing. const listPlans = async ({ status, limit } = {}) => { if (status !== undefined && !Object.values(PLAN_STATUS).includes(status)) { throw new OpError(`invalid plan status: ${status} (one of ${Object.values(PLAN_STATUS).join(", ")})`); } const n = limit === undefined ? PLAN_LIST_DEFAULT : limit; if (!Number.isInteger(n) || n < 1 || n > planLib.PLAN_RETENTION) { throw new OpError(`limit must be an integer from 1 to ${planLib.PLAN_RETENTION}`); } const rows = await planLib.listPlans({ status, limit: n }); const restoring = await rollback.restorePointIds(); const peerById = new Map((await peers.listPeers()).map((p) => [p.peer_id, p])); const lock = await siteLock.lockState(); const current = rows.some((r) => applyLib.APPLY_FROM.includes(r.status) || siteLock.IN_FLIGHT.includes(r.status)) ? await capture.buildDesiredState() : null; return rows.map((r) => ({ plan_id: r.plan_id, status: r.status, direction: r.direction, peer: peerById.has(r.peer_id) ? peerName(peerById.get(r.peer_id)) : null, source_env_id: r.source_env_id, created_at: r.created_at, ...planStanding(r, restoring.has(r.plan_id), current && current.hash, lock) })); }; // A stored plan with what it needs to be judged: { loaded (planLib.loadPlan), peer // (or null), env (current policies), current (this site's desired state now), // standing }. const inspectPlan = async (planId) => { const id = requirePlanId(planId); const loaded = await planLib.loadPlan(id); if (!loaded) { throw new OpError(`unknown plan ${id}`); } const current = await capture.buildDesiredState(); const hasPeer = loaded.peer_id !== null && loaded.peer_id !== undefined; return { loaded: loaded, peer: hasPeer ? await peers.findPeer(loaded.peer_id) : null, env: await requireEnv(), current: current, standing: planStanding(loaded, await rollback.hasRestorePoint(id), current.hash, await siteLock.lockState()) }; }; // Apply a previewed plan. The human gates are the only options a caller has (never // force): prunable = table names whose data may be dropped, confirmDataDrops = the // data-loss acknowledgement (on_delete=confirm), confirmRenames = old names of // detected renames to apply as data-preserving renames. Returns apply.applyPlan's // result plus canRollback; a failure carries extra = { planId, canRollback } (a // part-way failure leaves its restore point) -- except a refusal because another // apply or rollback holds the site lock, which changed nothing. const applyPlan = async (planId, { prunable, confirmDataDrops, confirmRenames } = {}) => { const id = requirePlanId(planId); let result; try { result = await applyLib.applyPlan(id, { applyDeletes: true, prunable: prunable || [], confirmDataDrops: !!confirmDataDrops, confirmRenames: confirmRenames || [] }); } catch (e) { if (e instanceof siteLock.LockBusyError) { throw new OpError(e.message); } throw new OpError(e.message, { planId: id, canRollback: await rollback.hasRestorePoint(id) }); } return { ...result, canRollback: await rollback.hasRestorePoint(id) }; }; // Revert an applied plan to its restore point. prunable / confirmDataDrops gate // dropping a table the apply created that has gained rows since. const rollbackPlan = async (planId, { prunable, confirmDataDrops } = {}) => { const id = requirePlanId(planId); return await adminStep(() => rollback.rollbackApply(id, { prunable: prunable || [], confirmDataDrops: !!confirmDataDrops })); }; // ---------------- Tables ---------------- // Every application table (infra excluded): row count, data mode, whether it is // locked to user mode, and when a starter table was seeded. const listTables = async () => { const records = await rowsLib.modeRecords(); const out = []; for (const t of (await Table.find({})).filter((x) => !isInfraName(x.name))) { const rec = records.get(t.name); out.push({ name: t.name, rows: await t.countRows(), mode: (rec && rec.data_mode) || DATA_MODES.USER, locked: isLockedTable(t.name), starter_shipped_at: (rec && rec.starter_shipped_at) || null }); } return out; }; // Set a table's data mode (rows.setDataMode validates the mode, the table and the // infra/locked rules). Returns { table, mode, previous }. const setTableMode = async (table, mode) => { const name = String(table || "").trim(); const previous = await rowsLib.dataModeOf(name); await adminStep(() => rowsLib.setDataMode(name, mode)); return { table: name, mode, previous }; }; // ---------------- This site's state ---------------- // This site's desired state as a peer pulling from it sees it: the content hash, // the entity count per kind, the syncable config keys present, and the row count of // each table it ships. Read-only (no identity backfill, unlike a real state answer). const localState = async () => { const { pack, hash } = await capture.buildDesiredState(); const counts = {}; for (const kind of capture.PACK_KINDS) { counts[kind] = pack[kind].length; } const shipped = await rowsLib.shippedRowCounts(); return { hash, counts, config: Object.keys(pack.config), rows: shipped.counts, warnings: shipped.warnings }; }; module.exports = { OpError, SETTING_OPTIONS, PLAN_LIST_DEFAULT, status, updateSettings, peerName, resolvePeer, addPeer, rotatePeerSecret, deletePeer, peerHealth, diffPluginsWithPeer, fetchPeerState, pullPlan, listPlans, inspectPlan, applyPlan, rollbackPlan, listTables, setTableMode, localState };