Paul Trowbridge
81c2324220
The label on a baseline or reference load ("Open Orders", "Prior Year")
lived only on the pf.log row; the data stream was a straight dump of the
forecast table, so Perspective saw pf_logid and never the name.
The stream now joins pf.log and emits two columns rather than one, because
commingling them makes neither useful: pf_segment names the load a row came
from, and is '(adjustment)' for everything else; pf_note carries the free
text on scale/recode/clone and is null on loads. The operation routes stamp
the same two fields on the rows they push back incrementally.
The tag column those labels belong in is declared in 01_schema.sql but
predates some installs, so the backfill seeding it from note joins the
ALTER already there. Adjustment notes are free text, not labels, and are
left alone.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
697 lines
32 KiB
JavaScript
697 lines
32 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 t.*
|
||
,CASE WHEN l.operation IN ('baseline','reference')
|
||
THEN COALESCE(NULLIF(l.tag, ''), NULLIF(l.note, ''), '(unlabeled load)')
|
||
ELSE '(adjustment)' END AS pf_segment
|
||
,CASE WHEN l.operation IN ('baseline','reference')
|
||
THEN NULL
|
||
ELSE COALESCE(NULLIF(l.tag, ''), NULLIF(l.note, '')) END AS pf_note
|
||
FROM ${tbl} t
|
||
LEFT JOIN pf.log l
|
||
ON l.id = t.pf_logid
|
||
`);
|
||
|
||
// 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 opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: '(adjustment)', pf_note: opLabel, 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 opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: '(adjustment)', pf_note: opLabel, 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 opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: '(adjustment)', pf_note: opLabel, 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;
|
||
};
|