The server had no authentication: every /api route was open, CORS allowed any origin, and the identity written to the audit log came from the request body — the UI sent a hardcoded pf_user: 'admin', which any client could have set to anything it liked. Accounts live in pf.app_user with scrypt hashes from node's own crypto, so there is no native build step and the parameters travel with each hash. Sessions are express-session over connect-pg-simple in pf.session: a restart no longer signs everyone out, and a session can be revoked by deleting its row, which is how disable-user cuts off access immediately rather than at cookie expiry. Everything under /api except login/logout/me now requires a session, and the React app is mounted only once there is one — its load effects call the API on mount, so a logged-out mount would just fire a burst of 401s. A session that expires while the app is open lands back on the login screen: auth.jsx wraps fetch once rather than teaching every call site to check. Identity is now read from the session for pf_user, created_by and closed_by, and the body values are ignored. Hardened for an internet-facing deployment: trust proxy so req.ip and secure-cookie detection are right behind TLS termination, httpOnly + SameSite=Lax + Secure cookies, ten login failures per IP per fifteen minutes, one error message for unknown, wrong and disabled alike, and a fresh session id on success. CORS is off entirely unless CORS_ORIGIN names an origin — a wildcard alongside a session cookie would be CSRF by construction. The server refuses to boot without SESSION_SECRET rather than falling back to a guessable default. pf.sh grows add-user, passwd, list-users, disable-user and enable-user; passwords are read on stdin and hashed before they reach psql, so no plaintext in argv or shell history. install.sh generates the secret, applies 02_auth.sql, and creates the first account. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
685 lines
31 KiB
JavaScript
685 lines
31 KiB
JavaScript
const express = require('express');
|
||
const { tableFromArrays, tableToIPC } = require('apache-arrow');
|
||
const { applyTokens, buildWhere, buildWhereAny, buildExcludeClause, buildExcludePredicate, buildSetClause, esc } = require('../lib/sql_generator');
|
||
const { sessionUser } = require('../lib/auth');
|
||
const { fcTable } = require('../lib/utils');
|
||
|
||
module.exports = function(pool) {
|
||
const router = express.Router();
|
||
|
||
async function runSQL(sql, client) {
|
||
console.log('--- SQL ---\n', sql, '\n--- END SQL ---');
|
||
return (client || pool).query(sql);
|
||
}
|
||
|
||
// accept either the legacy single `slice` object or the newer `slices` array,
|
||
// and drop any empty entries so an empty selection can never widen to TRUE
|
||
function normalizeSlices(body) {
|
||
const raw = Array.isArray(body.slices) && body.slices.length ? body.slices : [body.slice];
|
||
return raw.filter(s => s && typeof s === 'object' && Object.keys(s).length > 0);
|
||
}
|
||
|
||
// Stamp the tag onto the log entry the operation just created.
|
||
// Done as a follow-up UPDATE rather than inside the generated SQL: those
|
||
// templates live in pf.sql per source, so adding a {{tag}} token would strand
|
||
// every source that has not re-run "Generate SQL".
|
||
async function tagLog(client, rows, tag) {
|
||
const clean = (tag || '').trim();
|
||
if (!clean) return null;
|
||
const ids = [...new Set(rows.map(r => r.pf_logid).filter(id => id != null))];
|
||
if (ids.length === 0) return null;
|
||
await client.query(`UPDATE pf.log SET tag = $1 WHERE id = ANY($2::bigint[])`, [clean, ids]);
|
||
return clean;
|
||
}
|
||
|
||
// A slice is only meaningful if at least one of its keys is a filterable
|
||
// column. buildWhere silently drops unknown keys, so {"typo": "x"} would
|
||
// otherwise reduce to TRUE and apply the operation to the whole version.
|
||
// Refuse rather than let a malformed selection rewrite every row.
|
||
function assertSelective(slices, ctx) {
|
||
const allowed = new Set(ctx.filterCols);
|
||
slices.forEach((sl, i) => {
|
||
const hits = Object.keys(sl).filter(k => allowed.has(k));
|
||
if (hits.length === 0) {
|
||
const err = new Error(
|
||
`Slice ${i + 1} does not name any filterable column ` +
|
||
`(${JSON.stringify(sl)}). Expected one of: ${ctx.filterCols.join(', ')}.`
|
||
);
|
||
err.status = 400;
|
||
throw err;
|
||
}
|
||
});
|
||
}
|
||
|
||
// echo back what the caller asked for, for the audit log
|
||
function pickIntent(body) {
|
||
const keys = ['mode', 'target_basis', 'value_incr', 'units_incr', 'value_pct', 'units_pct', 'pct',
|
||
'target_value', 'target_units', 'target_price'];
|
||
const out = {};
|
||
for (const k of keys) if (body[k] !== undefined && body[k] !== null && body[k] !== '') out[k] = body[k];
|
||
return out;
|
||
}
|
||
|
||
// Totals for a WHERE clause, split into the rows operations can change and the
|
||
// rows they cannot. Excluded iters (typically 'reference') are still visible in
|
||
// the pivot, so their contribution has to be reported rather than dropped —
|
||
// otherwise a target set against what the grid shows lands somewhere else.
|
||
async function sliceTotals(client, ctx, whereClause, excludeClause) {
|
||
const pred = buildExcludePredicate(ctx.version.exclude_iters);
|
||
const agg = (col, filter) => col ? `sum(${col}) FILTER (WHERE ${filter})` : 'NULL';
|
||
const v = ctx.valueCol ? `"${ctx.valueCol}"` : null;
|
||
const u = ctx.unitsCol ? `"${ctx.unitsCol}"` : null;
|
||
|
||
const r = await client.query(`
|
||
SELECT ${agg(v, pred)} AS total_value,
|
||
${agg(u, pred)} AS total_units,
|
||
${agg(v ? `abs(${v})` : null, pred)} AS abs_value,
|
||
${agg(u ? `abs(${u})` : null, pred)} AS abs_units,
|
||
${agg(v, `NOT (${pred})`)} AS excl_value,
|
||
${agg(u, `NOT (${pred})`)} AS excl_units
|
||
FROM ${ctx.table} WHERE ${whereClause}
|
||
`);
|
||
const n = (x) => parseFloat(r.rows[0][x]) || 0;
|
||
return {
|
||
value: n('total_value'), units: n('total_units'),
|
||
absValue: n('abs_value'), absUnits: n('abs_units'),
|
||
exclValue: n('excl_value'), exclUnits: n('excl_units'),
|
||
};
|
||
}
|
||
|
||
// Resolve each measure independently into the increment the scale SQL expects.
|
||
// Value and units each accept exactly one of: an absolute target, a change
|
||
// amount, or a percentage — whichever the caller sent. They are resolved
|
||
// separately so a target on one measure and a percentage on the other can be
|
||
// submitted together. Everything is measured against the totals of *this*
|
||
// WHERE clause, which is what makes apply_mode 'each' land per slice.
|
||
async function resolveIncrs(client, ctx, whereClause, excludeClause, body) {
|
||
const num = (v) => (v === undefined || v === null || v === '') ? null : parseFloat(v);
|
||
|
||
const tValue = num(body.target_value);
|
||
const tUnits = num(body.target_units);
|
||
const tPrice = num(body.target_price);
|
||
const vIncr = num(body.value_incr);
|
||
const uIncr = num(body.units_incr);
|
||
let vPct = num(body.value_pct);
|
||
let uPct = num(body.units_pct);
|
||
|
||
// legacy shape: a single `pct` flag meaning "the increments are percentages"
|
||
if (body.pct) {
|
||
if (vPct === null && vIncr !== null) vPct = vIncr;
|
||
if (uPct === null && uIncr !== null) uPct = uIncr;
|
||
}
|
||
const legacyPct = !!body.pct;
|
||
|
||
const anyInput = [tValue, tUnits, tPrice, vIncr, uIncr, vPct, uPct].some(v => v !== null);
|
||
if (!anyInput) return { value: 0, units: 0 };
|
||
|
||
const totals = await sliceTotals(client, ctx, whereClause, excludeClause);
|
||
|
||
// What the number is measured against:
|
||
// 'adjustable' — only the rows this operation can write (the default, and
|
||
// what every earlier version of this API did)
|
||
// 'selected' — everything the pivot shows for the slice, excluded rows
|
||
// included. Those rows cannot move, so reaching the target
|
||
// means the adjustable rows absorb the whole difference.
|
||
const basis = body.target_basis === 'selected' ? 'selected' : 'adjustable';
|
||
const fixedValue = basis === 'selected' ? totals.exclValue : 0;
|
||
const fixedUnits = basis === 'selected' ? totals.exclUnits : 0;
|
||
|
||
// one measure: target wins, then percentage, then a plain change amount
|
||
const resolve = (target, pct, incr, current, fixed) => {
|
||
// subtract the immovable part: current + incr + fixed === target
|
||
if (target !== null) return (target - fixed) - current;
|
||
// a percentage of the basis, which may include the immovable part
|
||
if (pct !== null) return (current + fixed) * pct / 100;
|
||
if (incr !== null && !legacyPct) return incr;
|
||
return 0;
|
||
};
|
||
|
||
let value = resolve(tValue, vPct, vIncr, totals.value, fixedValue);
|
||
const units = resolve(tUnits, uPct, uIncr, totals.units, fixedUnits);
|
||
|
||
// a price target holds units constant: new value = price x current units.
|
||
// An explicit value target outranks it.
|
||
if (tPrice !== null && tValue === null) {
|
||
value = (tPrice * (totals.units + fixedUnits)) - (totals.value + fixedValue);
|
||
}
|
||
|
||
// the scale SQL divides by the slice total; with no rows there is
|
||
// nothing to prorate across and the increment would vanish anyway
|
||
if (totals.value === 0 && totals.units === 0) return { value: 0, units: 0 };
|
||
|
||
// Refuse to prorate across a pool that nets to ~zero. Each row's new value is
|
||
// (row / total) * increment, so as the net approaches zero the multiplier
|
||
// explodes and rows fly apart in opposite directions to hit the target — a
|
||
// mathematically faithful, practically useless result. Selecting slices that
|
||
// offset each other is the usual cause, and 'each' handles that correctly.
|
||
assertProratable(totals, value, units);
|
||
|
||
return { value: round(value, 6), units: round(units, 6) };
|
||
}
|
||
|
||
// a pool is proratable only if its net is a meaningful fraction of its gross
|
||
const NET_TO_GROSS_FLOOR = 0.01;
|
||
|
||
function assertProratable(totals, value, units) {
|
||
const check = (net, gross, incr, label) => {
|
||
if (!incr) return;
|
||
if (gross === 0) return;
|
||
if (Math.abs(net) >= gross * NET_TO_GROSS_FLOOR) return;
|
||
const err = new Error(
|
||
`Cannot prorate ${label} across this selection: the rows net to ` +
|
||
`${net.toFixed(2)} against a gross of ${gross.toFixed(2)}, so they very ` +
|
||
`nearly cancel out. Scaling to a target would push them to extreme ` +
|
||
`opposite values. Use "Each" to scale every slice on its own, or narrow ` +
|
||
`the selection so it does not mix offsetting rows.`
|
||
);
|
||
err.status = 400;
|
||
throw err;
|
||
};
|
||
check(totals.value, totals.absValue, value, 'value');
|
||
check(totals.units, totals.absUnits, units, 'units');
|
||
}
|
||
|
||
function round(n, dp) {
|
||
if (!isFinite(n)) return 0;
|
||
const f = Math.pow(10, dp);
|
||
return Math.round(n * f) / f;
|
||
}
|
||
|
||
// fetch everything needed to execute an operation:
|
||
// version + source info, col_meta, fc_table name, stored SQL
|
||
async function getContext(versionId, operation) {
|
||
const verResult = await pool.query(`
|
||
SELECT v.*, s.schema, s.tname, s.id AS source_id
|
||
FROM pf.version v
|
||
JOIN pf.source s ON s.id = v.source_id
|
||
WHERE v.id = $1
|
||
`, [versionId]);
|
||
if (verResult.rows.length === 0) {
|
||
const err = new Error('Version not found'); err.status = 404; throw err;
|
||
}
|
||
const version = verResult.rows[0];
|
||
|
||
const colResult = await pool.query(
|
||
`SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`,
|
||
[version.source_id]
|
||
);
|
||
const colMeta = colResult.rows;
|
||
const dimCols = colMeta.filter(c => c.role === 'dimension').map(c => c.cname);
|
||
const dateCols = colMeta.filter(c => c.role === 'date').map(c => c.cname);
|
||
const valueCol = colMeta.find(c => c.role === 'value')?.cname;
|
||
const unitsCol = colMeta.find(c => c.role === 'units')?.cname;
|
||
|
||
const sqlResult = await pool.query(
|
||
`SELECT sql FROM pf.sql WHERE source_id = $1 AND operation = $2`,
|
||
[version.source_id, operation]
|
||
);
|
||
if (sqlResult.rows.length === 0) {
|
||
const err = new Error(`No generated SQL for operation "${operation}" — run generate-sql first`);
|
||
err.status = 400; throw err;
|
||
}
|
||
|
||
return {
|
||
version,
|
||
table: fcTable(version.tname, version.id),
|
||
colMeta,
|
||
dimCols,
|
||
dateCols,
|
||
filterCols: [...dimCols, ...dateCols],
|
||
valueCol,
|
||
unitsCol,
|
||
sql: sqlResult.rows[0].sql
|
||
};
|
||
}
|
||
|
||
function guardOpen(version, res) {
|
||
if (version.status === 'closed') {
|
||
res.status(403).json({ error: 'Version is closed' });
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
// stream all rows for a version as Arrow IPC (all iters including reference)
|
||
router.get('/versions/:id/data', async (req, res) => {
|
||
const versionId = parseInt(req.params.id);
|
||
let client, committed = false;
|
||
try {
|
||
const verResult = await pool.query(
|
||
`SELECT v.*, s.tname FROM pf.version v JOIN pf.source s ON s.id = v.source_id WHERE v.id = $1`,
|
||
[versionId]
|
||
);
|
||
if (!verResult.rows.length) {
|
||
const err = new Error('Version not found'); err.status = 404; throw err;
|
||
}
|
||
const tbl = fcTable(verResult.rows[0].tname, versionId);
|
||
|
||
const { rows: [{ count }] } = await pool.query(`SELECT COUNT(*) FROM ${tbl}`);
|
||
const rowCount = parseInt(count);
|
||
|
||
res.setHeader('Content-Type', 'application/vnd.apache.arrow.stream');
|
||
res.setHeader('X-Row-Count', String(rowCount));
|
||
|
||
if (rowCount === 0) { res.end(); return; }
|
||
|
||
client = await pool.connect();
|
||
await client.query('BEGIN');
|
||
await client.query(`
|
||
DECLARE pf_cur CURSOR FOR
|
||
SELECT * FROM ${tbl}
|
||
`);
|
||
|
||
// Accumulate into column arrays (not row objects) to avoid allocating one JS
|
||
// object per row — cuts peak heap by ~3-5× for large datasets.
|
||
// Still emits a single Arrow record batch so Perspective WASM never sees
|
||
// dictionary REPLACEMENT messages (which crash its Arrow reader).
|
||
let colArrays = null;
|
||
while (true) {
|
||
const { rows } = await client.query('FETCH 10000 FROM pf_cur');
|
||
if (!rows.length) break;
|
||
if (!colArrays) {
|
||
colArrays = Object.fromEntries(Object.keys(rows[0]).map(k => [k, []]));
|
||
}
|
||
for (const row of rows) {
|
||
for (const k of Object.keys(colArrays)) colArrays[k].push(row[k]);
|
||
}
|
||
}
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
|
||
const buf = tableToIPC(tableFromArrays(colArrays || {}), 'stream');
|
||
res.setHeader('Content-Length', String(buf.byteLength));
|
||
res.end(Buffer.from(buf.buffer, buf.byteOffset, buf.byteLength));
|
||
} catch (err) {
|
||
console.error(err);
|
||
if (!res.headersSent) res.status(err.status || 500).json({ error: err.message });
|
||
else res.destroy();
|
||
} finally {
|
||
if (client) {
|
||
if (!committed) try { await client.query('ROLLBACK'); } catch {}
|
||
client.release();
|
||
}
|
||
}
|
||
});
|
||
|
||
// load baseline rows from source table — additive, no delete
|
||
router.post('/versions/:id/baseline', async (req, res) => {
|
||
const { where_clause, date_offset, note, filters, raw_where } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const dateOffset = date_offset || '0 days';
|
||
const filterClause = (raw_where || where_clause || '').trim() || 'TRUE';
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'baseline');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
const paramsJson = JSON.stringify({
|
||
where_clause: filterClause,
|
||
date_offset: dateOffset,
|
||
...(raw_where ? { raw_where } : (filters ? { filters } : {}))
|
||
});
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(paramsJson),
|
||
filter_clause: filterClause,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
|
||
const result = await runSQL(sql);
|
||
res.json(result.rows[0]);
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// edit a baseline or reference segment in place — only allowed before any
|
||
// scale/recode/clone has been applied on this version, since those would
|
||
// have been calibrated against the old segment's totals.
|
||
router.put('/versions/:id/baseline/:logid', async (req, res) => {
|
||
const versionId = parseInt(req.params.id);
|
||
const logid = parseInt(req.params.logid);
|
||
const { where_clause, date_offset, note, filters, raw_where } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const dateOffset = date_offset || '0 days';
|
||
const filterClause = (raw_where || where_clause || '').trim() || 'TRUE';
|
||
|
||
const client = await pool.connect();
|
||
try {
|
||
const logResult = await client.query(
|
||
`SELECT * FROM pf.log WHERE id = $1 AND version_id = $2`,
|
||
[logid, versionId]
|
||
);
|
||
if (logResult.rows.length === 0) {
|
||
return res.status(404).json({ error: 'Log entry not found' });
|
||
}
|
||
const oldLog = logResult.rows[0];
|
||
if (!['baseline', 'reference'].includes(oldLog.operation)) {
|
||
return res.status(400).json({ error: 'Only baseline or reference segments can be edited' });
|
||
}
|
||
|
||
const opsResult = await client.query(
|
||
`SELECT COUNT(*)::int AS n FROM pf.log
|
||
WHERE version_id = $1 AND operation IN ('scale', 'recode', 'clone')`,
|
||
[versionId]
|
||
);
|
||
if (opsResult.rows[0].n > 0) {
|
||
return res.status(409).json({
|
||
error: 'Cannot edit segments after forecast operations have been applied. Undo the operations first.'
|
||
});
|
||
}
|
||
|
||
const ctx = await getContext(versionId, oldLog.operation);
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
|
||
const paramsJson = JSON.stringify({
|
||
where_clause: filterClause,
|
||
date_offset: dateOffset,
|
||
...(raw_where ? { raw_where } : (filters ? { filters } : {}))
|
||
});
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(paramsJson),
|
||
filter_clause: filterClause,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
|
||
await client.query('BEGIN');
|
||
const delRows = await client.query(
|
||
`DELETE FROM ${ctx.table} WHERE pf_logid = $1 RETURNING pf_id`,
|
||
[logid]
|
||
);
|
||
await client.query(`DELETE FROM pf.log WHERE id = $1`, [logid]);
|
||
const insResult = await client.query(sql);
|
||
await client.query('COMMIT');
|
||
|
||
res.json({
|
||
rows_deleted: delRows.rowCount,
|
||
pf_ids: delRows.rows.map(r => r.pf_id),
|
||
rows_affected: insResult.rows[0]?.rows_affected ?? 0
|
||
});
|
||
} catch (err) {
|
||
try { await client.query('ROLLBACK'); } catch {}
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
} finally {
|
||
client.release();
|
||
}
|
||
});
|
||
|
||
// delete all baseline rows and log entries for a version
|
||
router.delete('/versions/:id/baseline', async (req, res) => {
|
||
const versionId = parseInt(req.params.id);
|
||
try {
|
||
const ctx = await getContext(versionId, 'baseline');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
const client = await pool.connect();
|
||
try {
|
||
await client.query('BEGIN');
|
||
const delRows = await client.query(
|
||
`DELETE FROM ${ctx.table} WHERE pf_iter = 'baseline' RETURNING pf_id`
|
||
);
|
||
const delLog = await client.query(
|
||
`DELETE FROM pf.log WHERE version_id = $1 AND operation = 'baseline'`,
|
||
[versionId]
|
||
);
|
||
await client.query('COMMIT');
|
||
res.json({
|
||
rows_deleted: delRows.rowCount,
|
||
log_entries_deleted: delLog.rowCount,
|
||
pf_ids: delRows.rows.map(r => r.pf_id)
|
||
});
|
||
} catch (err) {
|
||
await client.query('ROLLBACK');
|
||
throw err;
|
||
} finally {
|
||
client.release();
|
||
}
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// load reference rows from source table (additive — does not clear prior reference rows)
|
||
router.post('/versions/:id/reference', async (req, res) => {
|
||
const { where_clause, date_offset, note, filters, raw_where } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const dateOffset = date_offset || '0 days';
|
||
const filterClause = (raw_where || where_clause || '').trim() || 'TRUE';
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'reference');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
const paramsJson = JSON.stringify({
|
||
where_clause: filterClause,
|
||
date_offset: dateOffset,
|
||
...(raw_where ? { raw_where } : (filters ? { filters } : {}))
|
||
});
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(paramsJson),
|
||
filter_clause: filterClause,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
|
||
const result = await runSQL(sql);
|
||
res.json(result.rows[0]);
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// scale one or more slices — adjust value and/or units toward an absolute
|
||
// target or by an increment. With several slices selected, apply_mode decides
|
||
// whether they are treated as one pool ('prorate') or independently ('each').
|
||
router.post('/versions/:id/scale', async (req, res) => {
|
||
const { note, apply_mode } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const slices = normalizeSlices(req.body);
|
||
if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });
|
||
|
||
const applyMode = apply_mode === 'each' ? 'each' : 'prorate';
|
||
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'scale');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
assertSelective(slices, ctx);
|
||
|
||
const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
|
||
|
||
// 'prorate' pools every slice into one WHERE and lets the SQL's
|
||
// sum() OVER () distribute the increment across the whole pool.
|
||
// 'each' runs the same statement once per slice, so every slice
|
||
// reaches the target on its own and gets its own log entry.
|
||
const units = applyMode === 'each'
|
||
? slices.map(sl => ({ slices: [sl], where: buildWhere(sl, ctx.filterCols) }))
|
||
: [{ slices, where: buildWhereAny(slices, ctx.filterCols) }];
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
let applied = 0;
|
||
const skipped = [];
|
||
|
||
for (const unit of units) {
|
||
const incr = await resolveIncrs(client, ctx, unit.where, excludeClause, req.body);
|
||
// no rows, or already at the target — nothing to write for this slice
|
||
if (incr.value === 0 && incr.units === 0) { skipped.push(...unit.slices); continue; }
|
||
applied++;
|
||
|
||
const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(JSON.stringify({
|
||
slices: unit.slices,
|
||
apply_mode: applyMode,
|
||
...pickIntent(req.body),
|
||
resolved: { value_incr: incr.value, units_incr: incr.units }
|
||
})),
|
||
slice: esc(JSON.stringify(loggedSlice)),
|
||
where_clause: unit.where,
|
||
exclude_clause: excludeClause,
|
||
value_incr: incr.value,
|
||
units_incr: incr.units
|
||
});
|
||
const result = await runSQL(sql, client);
|
||
await tagLog(client, result.rows, req.body.tag);
|
||
allRows.push(...result.rows);
|
||
}
|
||
|
||
if (allRows.length === 0) {
|
||
await client.query('ROLLBACK');
|
||
return res.status(400).json({
|
||
error: 'Nothing to scale — the target matches the current total, or the increment is zero'
|
||
});
|
||
}
|
||
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'scale' }));
|
||
res.json({
|
||
rows,
|
||
rows_affected: rows.length,
|
||
slices_applied: applied,
|
||
...(skipped.length ? { slices_skipped: skipped } : {})
|
||
});
|
||
} finally {
|
||
if (!committed) try { await client.query('ROLLBACK'); } catch {}
|
||
client.release();
|
||
}
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// recode dimension values on one or more slices
|
||
// inserts negative rows to zero out the original, positive rows with new dimension values
|
||
router.post('/versions/:id/recode', async (req, res) => {
|
||
const { note, set, apply_mode } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const slices = normalizeSlices(req.body);
|
||
if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });
|
||
if (!set || Object.keys(set).length === 0) return res.status(400).json({ error: 'set is required' });
|
||
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'recode');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
assertSelective(slices, ctx);
|
||
|
||
const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
|
||
const setClause = buildSetClause(ctx.dimCols, set);
|
||
const units = sliceUnits(slices, ctx, apply_mode);
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
for (const unit of units) {
|
||
const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(JSON.stringify({ slices: unit.slices, set, apply_mode: unit.mode })),
|
||
slice: esc(JSON.stringify(loggedSlice)),
|
||
where_clause: unit.where,
|
||
exclude_clause: excludeClause,
|
||
set_clause: setClause
|
||
});
|
||
const result = await runSQL(sql, client);
|
||
await tagLog(client, result.rows, req.body.tag);
|
||
allRows.push(...result.rows);
|
||
}
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'recode' }));
|
||
res.json({ rows, rows_affected: rows.length, slices_applied: units.length });
|
||
} finally {
|
||
if (!committed) try { await client.query('ROLLBACK'); } catch {}
|
||
client.release();
|
||
}
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// clone one or more slices as new business under new dimension values
|
||
// does not offset the original slice
|
||
router.post('/versions/:id/clone', async (req, res) => {
|
||
const { note, set, scale, apply_mode } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const slices = normalizeSlices(req.body);
|
||
if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });
|
||
if (!set || Object.keys(set).length === 0) return res.status(400).json({ error: 'set is required' });
|
||
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'clone');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
assertSelective(slices, ctx);
|
||
|
||
const scaleFactor = (scale != null) ? parseFloat(scale) : 1.0;
|
||
const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
|
||
const setClause = buildSetClause(ctx.dimCols, set);
|
||
const units = sliceUnits(slices, ctx, apply_mode);
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
for (const unit of units) {
|
||
const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
params: esc(JSON.stringify({ slices: unit.slices, set, scale: scaleFactor, apply_mode: unit.mode })),
|
||
slice: esc(JSON.stringify(loggedSlice)),
|
||
where_clause: unit.where,
|
||
exclude_clause: excludeClause,
|
||
set_clause: setClause,
|
||
scale_factor: scaleFactor
|
||
});
|
||
const result = await runSQL(sql, client);
|
||
await tagLog(client, result.rows, req.body.tag);
|
||
allRows.push(...result.rows);
|
||
}
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'clone' }));
|
||
res.json({ rows, rows_affected: rows.length, slices_applied: units.length });
|
||
} finally {
|
||
if (!committed) try { await client.query('ROLLBACK'); } catch {}
|
||
client.release();
|
||
}
|
||
} catch (err) {
|
||
console.error(err);
|
||
res.status(err.status || 500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// log routes (GET /versions/:id/log, DELETE /log/:logid, PATCH /log/:logid)
|
||
// live in routes/log.js — see that file.
|
||
|
||
return router;
|
||
};
|