sc-site-sync/lib/operations.js

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