A conflict means an incoming op and a local op both touched the same entity since the last sync. The incoming op was NOT applied; pick which version wins.
Use theirs: applies the incoming op now (overwrites local change). The local op stays in the journal but its effect is overwritten.
Use mine: marks the incoming op as rejected. The local state stands. The peer may re-send the op on future syncs; subsequent pulls will skip it via idempotency.
`;
adminPage(req, res, "conflicts",
{ type: "card", title: "Pending conflicts", contents: intro + conflictsTable }
);
};
const conflictsResolve = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
try {
const opId = (req.body.op_id || "").trim();
const action = (req.body.action || "").trim();
if (!opId) throw new Error("op_id required");
if (!["theirs", "mine"].includes(action)) throw new Error("action must be 'theirs' or 'mine'");
// For file ops, "use theirs" re-applies create_file which fetches bytes
// from the originating peer -- resolve the peer from the op's source env.
const op = await db.selectMaybeOne("_dd_ops", { op_id: opId });
const peer = op ? await peers.findPeerByEnvId(op.source_env_id) : null;
const env = await getEnv();
const opts = {
peerId: peer ? peer.peer_id : null,
myEnvId: env ? env.env_id : null
};
const r = await resolveConflict(opId, action, opts);
res.redirect("/admin/dev-deploy/conflicts?msg=" + encodeURIComponent(`resolved ${opId.slice(0, 8)} with action=${action}: ${JSON.stringify(r)}`));
} catch (err) {
res.redirect("/admin/dev-deploy/conflicts?err=" + encodeURIComponent(err.message));
}
};
const renderValue = (v) => {
if (v === null || v === undefined) {
return '(unset)';
}
if (typeof v === "object") {
return `
${escape(JSON.stringify(v, null, 2))}
`;
}
return escape(String(v));
};
const conflictsMergeView = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
const opId = (req.query.op_id || "").trim();
if (!opId) { res.redirect("/admin/dev-deploy/conflicts?err=op_id+required"); return; }
try {
const op = await db.selectMaybeOne("_dd_ops", { op_id: opId });
if (!op) throw new Error(`op ${opId} not found`);
if (op.status !== "conflict") throw new Error(`op ${opId} is not in conflict status`);
const diff = await conflictFieldDiff(op);
const diffs = diff.diffs || [];
const ent = diff.instance
? `
Entity: ${escape(diff.kind)}${escape(diff.instance.name || diff.instance.role || diff.instance.id)} (local id ${escape(diff.instance.id)})
`
: `
${escape(diff.reason || "no entity diff available")}
`;
let formBody;
if (diffs.length === 0) {
formBody = `
No field-level differences detected (current state already matches the incoming op's patch on every field). Applying the merge will just mark this conflict as resolved.
`;
} else {
const rows = diffs.map((d) => `
${escape(d.field)}
${renderValue(d.currentValue)}
${renderValue(d.incomingValue)}
`).join("");
formBody = `
field
current (mine)
incoming (theirs)
resolution
${rows}
Defaults to keep current for every field — submit as-is for a no-op resolution that just clears the conflict marker.
`;
}
const contents = `${ent}
Incoming op ${escape(op.op_id.slice(0, 8))} from ${escape((op.source_env_id || "").slice(0, 8))}; ${escape(op.op_type)}.
`;
adminPage(req, res, "merge", { type: "card", title: "Merge conflict per field", contents: contents });
} catch (err) {
res.redirect("/admin/dev-deploy/conflicts?err=" + encodeURIComponent(err.message));
}
};
// Parse number/boolean/null literal strings from a text input. JSON.parse first
// (handles "true", "42", "null"); fall back to the raw string.
const coerce = (s) => {
if (s === undefined || s === null) return s;
try {
const parsed = JSON.parse(s);
if (parsed === null || ["boolean", "number"].includes(typeof parsed)) return parsed;
return s;
} catch (e) {
return s;
}
};
const conflictsMergeApply = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
try {
const opId = (req.body.op_id || "").trim();
if (!opId) throw new Error("op_id required");
const op = await db.selectMaybeOne("_dd_ops", { op_id: opId });
if (!op) throw new Error(`op ${opId} not found`);
if (op.status !== "conflict") throw new Error(`op ${opId} is not in conflict status`);
const payload = typeof op.payload === "string" ? JSON.parse(op.payload) : (op.payload || {});
const incomingPatch = payload.patch || {};
// For each "choice_" entry in the form body, decide what value
// to write -- if any. "current" means don't touch the field.
const choices = {};
for (const [k, v] of Object.entries(req.body || {})) {
if (!k.startsWith("choice_")) continue;
const field = k.substring("choice_".length);
if (v === "incoming") {
choices[field] = incomingPatch[field];
} else if (v === "custom") {
const customVal = req.body[`custom_${field}`];
choices[field] = coerce(customVal);
}
}
const r = await resolveConflictByMerge(opId, choices);
res.redirect("/admin/dev-deploy/conflicts?msg=" + encodeURIComponent(`merged ${opId.slice(0, 8)}: ${JSON.stringify(r)}`));
} catch (err) {
res.redirect("/admin/dev-deploy/conflicts?err=" + encodeURIComponent(err.message));
}
};
// ---------------- Tables (data_mode) ----------------
const tablesView = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
const schema = db.getTenantSchemaPrefix();
const rows = (await db.query(`
SELECT e.uuid, e.current_name, e.current_id,
COALESCE(m.data_mode, 'user') AS data_mode,
m.updated_at,
m.starter_shipped_at
FROM ${schema}_dd_entity_ids e
LEFT JOIN ${schema}_dd_table_modes m ON m.table_uuid = e.uuid
WHERE e.kind = 'table'
ORDER BY e.current_name
`)).rows;
const lockedNames = new Set(["users"]);
const modeOpts = [DATA_MODES.USER, DATA_MODES.STARTER, DATA_MODES.MANAGED];
const tablesTable = rows.length === 0
? p({ class: "text-muted" }, "No tables tracked yet")
: mkTable([
{ label: "table", key: (r) => escape(r.current_name) },
{ label: "uuid", key: (r) => code(escape(r.uuid.slice(0, 8))) },
{ label: "local id", key: (r) => escape(r.current_id) },
{ label: "data_mode", key: (r) => {
const shipped = r.starter_shipped_at ? " " + span({ class: "text-muted" }, "shipped " + escape(r.starter_shipped_at)) : "";
if (lockedNames.has(r.current_name)) return span({ class: "badge bg-secondary" }, "user (locked)") + shipped;
const opts = modeOpts.map((m) => ``).join("");
return `` + shipped;
} },
{ label: "updated_at", key: (r) => escape(r.updated_at || "-") },
], rows);
const intro = `
Controls how each table's row content propagates between environments. Choose carefully - switching from user to managed or starter rewrites the table's schema (adds a hidden _dd_row_uuid column) and ships existing rows.
user (default) — rows belong to the local environment; deploys never touch them. The only safe choice for end-user-entered data.
starter — rows ship to target on first install, then the target owns them; future changes on this side don't propagate. Good for default user roles, sample categories, template data the user expects to customize.
managed — rows always sync from source. Source is canonical; target's edits get overwritten or surface as conflicts. Good for catalogs, lookup tables, anything dev-curated.
The Saltcorn users table is locked to user and cannot be changed.
`;
adminPage(req, res, "tables",
{ type: "card", title: "Tables - data mode", contents: intro + tablesTable }
);
};
const tablesSet = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
try {
const tableUuid = (req.body.table_uuid || "").trim();
const dataMode = (req.body.data_mode || "").trim();
if (!tableUuid) throw new Error("table_uuid required");
const allowed = new Set(Object.values(DATA_MODES));
if (!allowed.has(dataMode)) throw new Error(`data_mode must be one of ${[...allowed].join(", ")}`);
const ent = await db.selectMaybeOne("_dd_entity_ids", { uuid: tableUuid });
if (!ent || ent.kind !== "table") throw new Error("table not found");
if (ent.current_name === "users") throw new Error("the users table is locked to data_mode=user");
const { ensureManagedSchema, dropManagedSchema, allRowsWithUuid, setRowUuid, newRowUuid, COLUMN_NAME: ROW_UUID_COL } = require("./rowIdentity");
const { rowToPortable, markStarterShipped } = require("./rowPayload");
const Table = require("@saltcorn/data/models/table");
const { randomUuid } = require("./ids");
const { enterOp } = require("./context");
const { recordOpSafely } = require("./ops");
const { refreshState } = require("./state");
await refreshState();
const prior = await db.selectMaybeOne("_dd_table_modes", { table_uuid: tableUuid });
const now = new Date().toISOString();
// Upsert mode row first.
if (prior) {
await db.updateWhere("_dd_table_modes", { data_mode: dataMode, updated_at: now, starter_shipped_at: null }, { table_uuid: tableUuid });
} else {
await db.insert("_dd_table_modes", { table_uuid: tableUuid, data_mode: dataMode, updated_at: now, starter_shipped_at: null }, { noid: true });
}
// Journal set_table_mode FIRST so target's apply sees mode change before any row ops.
{
const opId = randomUuid();
await enterOp(opId, async () => {
await recordOpSafely({
op_id: opId,
op_type: "set_table_mode",
entity_kind: "table_mode",
entity_uuid: tableUuid,
payload: { table_uuid: tableUuid, data_mode: dataMode, before_mode: (prior && prior.data_mode) || DATA_MODES.USER }
});
});
}
let initialShipped = 0;
if (dataMode === DATA_MODES.MANAGED || dataMode === DATA_MODES.STARTER) {
// Make sure THIS instance has the hidden column + UUIDs on existing rows.
await ensureManagedSchema(ent.current_name);
// Initial ship: journal an insert_row op for every existing row.
const table = Table.findOne({ id: ent.current_id });
if (table) {
const rows = await allRowsWithUuid(ent.current_name);
for (const row of rows) {
let rowUuid = row[ROW_UUID_COL];
if (!rowUuid) {
rowUuid = newRowUuid();
await setRowUuid(ent.current_name, row.id, rowUuid);
}
const { portable } = await rowToPortable(row, table);
const opId = randomUuid();
await enterOp(opId, async () => {
await recordOpSafely({
op_id: opId,
op_type: "insert_row",
entity_kind: "table_row",
entity_uuid: rowUuid,
payload: { table_uuid: tableUuid, after: portable }
});
});
initialShipped++;
}
}
// For starter: lock out further row ops.
if (dataMode === DATA_MODES.STARTER) {
await markStarterShipped(tableUuid);
}
} else if (prior && (prior.data_mode === DATA_MODES.MANAGED || prior.data_mode === DATA_MODES.STARTER)) {
// Reverting to user — drop the hidden column for cleanliness. Best-effort.
try {
await dropManagedSchema(ent.current_name);
} catch (e) {
// ignore on older SQLite that doesn't support DROP COLUMN
}
}
const summary = initialShipped > 0
? `set ${ent.current_name} to ${dataMode}; shipped ${initialShipped} rows`
: `set ${ent.current_name} to ${dataMode}`;
res.redirect("/admin/dev-deploy/tables?msg=" + encodeURIComponent(summary));
} catch (err) {
res.redirect("/admin/dev-deploy/tables?err=" + encodeURIComponent(err.message));
}
};
// ---------------- Revert ----------------
const revertView = async (req, res) => {
if (!isAdmin(req)) { res.status(403).type("text/plain").send("admin only"); return; }
try {
const opId = (req.body.op_id || "").trim();
if (!opId) throw new Error("op_id required");
const result = await revertOp(opId);
res.redirect("/admin/dev-deploy/ops?msg=" + encodeURIComponent(`reverted op ${opId.slice(0, 8)}: ${JSON.stringify(result)}`));
} catch (err) {
res.redirect("/admin/dev-deploy/ops?err=" + encodeURIComponent(err.message));
}
};
// ---------------- Machine endpoints ----------------
const apiJournal = async (req, res) => {
const peer = await requirePeerAuth(req, res);
if (!peer) return;
const since = req.query.since;
const env = await getEnv();
const schema = db.getTenantSchemaPrefix();
let sql = `SELECT op_id, source_env_id, op_type, entity_kind, entity_uuid, payload, parent_op_id, correlation_id, schema_version, created_at, status FROM ${schema}_dd_ops WHERE source_env_id = $1`;
const params = [env.env_id];
if (since) {
const anchorRow = await db.selectMaybeOne("_dd_ops", { op_id: since });
if (anchorRow) {
sql += ` AND created_at > $${params.length + 1}`;
params.push(anchorRow.created_at);
}
}
sql += ` ORDER BY created_at ASC LIMIT 1000`;
const ops = (await db.query(sql, params)).rows;
res.json({ source_env_id: env.env_id, ops: ops });
};
const apiHealth = async (req, res) => {
const peer = await requirePeerAuth(req, res);
if (!peer) return;
const env = await getEnv();
const plugins = (await db.query(`SELECT name, source, version FROM _sc_plugins ORDER BY name`)).rows;
res.json({
env_id: env.env_id,
label: env.env_label,
plugins: 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 returns non-200, returns a single "couldn't
// reach peer's health endpoint" warning and lets the caller proceed.
const diffPluginsWithPeer = async (peerRow, env, localPlugins) => {
let r;
try {
const secret = await peers.peerSecret(peerRow.peer_id);
r = await signedFetch({
baseUrl: peerRow.base_url,
method: "GET",
path: "/dev-deploy/api/health",
body: null,
sourceEnvId: env.env_id,
secret: secret,
requireTls: peerRow.require_tls
});
} catch (e) {
return [`could not check peer plugin list: ${e.message}`];
}
if (!r.ok || !r.body || !Array.isArray(r.body.plugins)) {
return [`peer's health endpoint returned ${r.status}`];
}
const localByName = new Map(localPlugins.map((p) => [p.name, p]));
const peerByName = new Map(r.body.plugins.map((p) => [p.name, p]));
const warnings = [];
for (const [name, mine] of localByName) {
const theirs = peerByName.get(name);
if (!theirs) {
warnings.push(`peer missing plugin "${name}"`);
} else if ((mine.version || "") !== (theirs.version || "")) {
warnings.push(`plugin version mismatch on "${name}": local ${mine.version || "?"}, peer ${theirs.version || "?"}`);
}
}
for (const [name, theirs] of peerByName) {
if (!localByName.has(name)) {
warnings.push(`peer has plugin not installed here: "${name}"`);
}
}
return warnings;
};
const apiFile = async (req, res) => {
try {
const peer = await requirePeerAuth(req, res);
if (!peer) return;
const uuid = req.params.uuid;
if (!uuid) {
res.status(400).json({ error: "uuid required" });
return;
}
const mapping = await db.selectMaybeOne("_dd_entity_ids", { uuid: uuid, kind: "file" });
if (!mapping) {
res.status(404).json({ error: "file not found", uuid: uuid });
return;
}
const path = require("path");
const dbMod = require("@saltcorn/data/db");
const absPath = path.join(dbMod.connectObj.file_store, dbMod.getTenantSchema(), mapping.current_name);
res.type("application/octet-stream");
// dotfiles: 'allow' so paths containing .dev-state (etc.) aren't
// silently treated as not-found by Express's default dotfile policy.
res.sendFile(absPath, { dotfiles: "allow" }, (err) => {
if (err && !res.headersSent) {
// eslint-disable-next-line no-console
console.error(`[dev-deploy] sendFile failed for ${absPath}:`, err.message);
res.status(500).json({ error: "failed to read file: " + err.message, path: absPath });
}
});
} catch (err) {
// eslint-disable-next-line no-console
console.error(`[dev-deploy] apiFile crashed:`, err && err.stack ? err.stack : err);
if (!res.headersSent) {
res.status(500).json({ error: err.message });
}
}
};
const apiIngest = async (req, res) => {
const peer = await requirePeerAuth(req, res);
if (!peer) return;
const ops = (req.body && req.body.ops) || [];
if (!Array.isArray(ops)) {
res.status(400).json({ error: "ops must be an array" });
return;
}
const env = await getEnv();
const results = await applyBatch(ops, { peerId: peer.peer_id, myEnvId: env.env_id });
// Advance inbound anchor to the last op_id from the source side
if (ops.length > 0) {
const lastOp = ops[ops.length - 1];
const now = new Date().toISOString();
const existing = await db.selectMaybeOne("_dd_anchors", { peer_id: peer.peer_id, direction: "inbound" });
if (existing) {
await db.updateWhere("_dd_anchors", { last_op_id: lastOp.op_id, updated_at: now }, { peer_id: peer.peer_id, direction: "inbound" });
} else {
await db.insert("_dd_anchors", { peer_id: peer.peer_id, direction: "inbound", last_op_id: lastOp.op_id, updated_at: now }, { noid: true });
}
}
res.json({ received: ops.length, results: results });
};
// ---------------- Route registration ----------------
const routes = [
{ url: "/admin/dev-deploy/", method: "get", callback: dashboard },
{ url: "/admin/dev-deploy/ops", method: "get", callback: opsView },
{ url: "/admin/dev-deploy/peers", method: "get", callback: peersView },
{ url: "/admin/dev-deploy/peers/add", method: "post", callback: peersAdd },
{ url: "/admin/dev-deploy/peers/rotate", method: "post", callback: peersRotate },
{ url: "/admin/dev-deploy/peers/delete", method: "post", callback: peersDelete },
{ url: "/admin/dev-deploy/plan", method: "get", callback: planView },
{ url: "/admin/dev-deploy/promote", method: "post", callback: promote },
{ url: "/admin/dev-deploy/pull", method: "post", callback: pull },
{ url: "/admin/dev-deploy/revert", method: "post", callback: revertView },
{ url: "/admin/dev-deploy/tables", method: "get", callback: tablesView },
{ url: "/admin/dev-deploy/tables/set", method: "post", callback: tablesSet },
{ url: "/admin/dev-deploy/conflicts", method: "get", callback: conflictsView },
{ url: "/admin/dev-deploy/conflicts/resolve", method: "post", callback: conflictsResolve },
{ url: "/admin/dev-deploy/conflicts/merge", method: "get", callback: conflictsMergeView },
{ url: "/admin/dev-deploy/conflicts/merge/apply", method: "post", callback: conflictsMergeApply },
{ url: "/dev-deploy/api/journal", method: "get", callback: apiJournal, noCsrf: true },
{ url: "/dev-deploy/api/ingest", method: "post", callback: apiIngest, noCsrf: true },
{ url: "/dev-deploy/api/file/:uuid", method: "get", callback: apiFile, noCsrf: true },
{ url: "/dev-deploy/api/health", method: "get", callback: apiHealth, noCsrf: true }
];
module.exports = {
routes
};