Paul Trowbridge
8f96fa3f7f
A selection of five AOP rows cloned nothing, with no explanation. The cause was exclude_iters, which clone applied along with scale and recode. It should not. That exclusion exists to stop operations *modifying* reference rows: scale distributes an increment across its pool, so including reference would attribute forecast movement to prior-year rows, and recode writes negative rows that zero the original out. Clone does neither -- it reads rows and inserts new pf_iter = 'clone' rows, leaving the source untouched. Copying a plan or a prior year out of reference and into adjustments is the operation doing exactly what it is for. So from_logid stops being the only way to reach those rows and becomes what it should be: a narrowing, for a selection spanning AOP and Prior Year where only one is wanted. The "would just duplicate the rows" guard no longer fires when the selection is entirely non-adjustable, since moving rows from reference into adjustments changes what they are even at factor 1 with no shift. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
834 lines
40 KiB
JavaScript
834 lines
40 KiB
JavaScript
const express = require('express');
|
||
const { tableFromArrays, tableToIPC } = require('apache-arrow');
|
||
const { applyTokens, buildWhere, buildWhereAny, buildExcludeClause, buildExcludePredicate, buildSetClause, dateGroupsOf, dimPeriodMapOf, 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;
|
||
}
|
||
|
||
// How a multi-slice request is split into statements.
|
||
//
|
||
// 'each' — one statement per slice, so every slice reaches its target on
|
||
// its own and gets its own log entry
|
||
// 'prorate' — a single statement over all of them, letting the SQL's
|
||
// sum() OVER () distribute across the whole pool
|
||
//
|
||
// Only scale has a target to prorate towards; recode and clone rewrite rows
|
||
// rather than distribute an amount, so for them this only decides whether the
|
||
// work lands as one log entry or several.
|
||
function sliceUnits(slices, ctx, applyMode) {
|
||
return applyMode === 'each'
|
||
? slices.map(sl => ({ slices: [sl], where: buildWhere(sl, ctx.filterCols) }))
|
||
: [{ slices, where: buildWhereAny(slices, ctx.filterCols) }];
|
||
}
|
||
|
||
// 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);
|
||
let units = resolve(tUnits, uPct, uIncr, totals.units, fixedUnits);
|
||
|
||
// A price target is the "edit price" mode of the Excel form: price and
|
||
// volume are the inputs and dollars fall out of them. With a units target
|
||
// alongside it, both move; without one, volume holds and price alone carries
|
||
// the change. An explicit value target outranks it either way.
|
||
if (tPrice !== null && tValue === null) {
|
||
const targetUnits = tUnits !== null
|
||
? (tUnits - fixedUnits) + 0 // the units target is already absolute
|
||
: (totals.units + fixedUnits);
|
||
value = (tPrice * targetUnits) - (totals.value + fixedValue);
|
||
}
|
||
|
||
// Which side of price x volume absorbs a dollar change.
|
||
//
|
||
// 'price' — volume holds, so price moves. This is what the API has always
|
||
// done, and stays the default so existing callers are unaffected.
|
||
// 'volume' — price holds, so volume scales with the dollars.
|
||
//
|
||
// Only meaningful when dollars were the input and units were not given
|
||
// explicitly; naming both means the caller has already decided.
|
||
const plug = body.plug === 'volume' ? 'volume' : 'price';
|
||
const unitsGiven = [tUnits, uIncr, uPct].some(v => v !== null);
|
||
|
||
if (plug === 'volume' && value !== 0 && !unitsGiven) {
|
||
const curValue = totals.value + fixedValue;
|
||
const curUnits = totals.units + fixedUnits;
|
||
if (curValue === 0) {
|
||
const err = new Error(
|
||
'Cannot hold price constant here: the selection currently has no value, ' +
|
||
'so there is no price to hold. Scale units directly, or let price absorb ' +
|
||
'the change.'
|
||
);
|
||
err.status = 400; throw err;
|
||
}
|
||
// price constant means value and units move by the same proportion:
|
||
// fVol = curVol * (fVal / curVal), so the units delta is curVol * value/curVal
|
||
units = curUnits * (value / curValue);
|
||
}
|
||
|
||
// 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
|
||
-- What the row counts towards. A load falls back to its own
|
||
-- name until it is labelled; an adjustment falls back to
|
||
-- 'Forecast', because that is what an adjustment is -- exclude_iters
|
||
-- keeps operations off the reference segments, so there is no
|
||
-- adjustment that is not part of the forecast.
|
||
,COALESCE(NULLIF(l.bucket, ''),
|
||
CASE WHEN l.operation IN ('baseline','reference')
|
||
THEN COALESCE(NULLIF(l.tag, ''), NULLIF(l.note, ''), '(unlabeled load)')
|
||
ELSE 'Forecast' END) AS pf_bucket
|
||
,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();
|
||
}
|
||
}
|
||
});
|
||
|
||
// Aggregate a version to its display grain and return it as Arrow IPC.
|
||
// This replaces /data for sources that define a grain (col_meta.in_grain):
|
||
// the aggregation collapses the row count by orders of magnitude, so the
|
||
// result loads as one small native Perspective table indexed on pf_gkey and
|
||
// the WASM view still does all rollup/expand/collapse locally.
|
||
router.get('/versions/:id/agg', async (req, res) => {
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'get_agg');
|
||
const sql = applyTokens(ctx.sql, { fc_table: ctx.table });
|
||
const { rows } = await runSQL(sql);
|
||
|
||
res.setHeader('Content-Type', 'application/vnd.apache.arrow.stream');
|
||
res.setHeader('X-Row-Count', String(rows.length));
|
||
if (rows.length === 0) { res.end(); return; }
|
||
|
||
// column arrays, one Arrow record batch — same constraint as /data:
|
||
// per-batch dictionaries crash Perspective's Arrow reader
|
||
const 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]);
|
||
}
|
||
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();
|
||
}
|
||
});
|
||
|
||
// 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 = sliceUnits(slices, ctx, applyMode);
|
||
|
||
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_bucket: 'Forecast', 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 === 'each' ? 'each' : 'prorate');
|
||
|
||
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_bucket: 'Forecast', 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, from_logid, date_offset } = 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' });
|
||
|
||
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 dateOffset = (date_offset || '0 days').trim() || '0 days';
|
||
|
||
// The offset is interpolated into the SQL as an interval literal, so a
|
||
// typo would surface as a Postgres parse error mid-statement. Ask
|
||
// Postgres to parse it on its own first, where the failure is cheap and
|
||
// can be reported against the field the user typed it into. Negative
|
||
// intervals are fine and useful -- '-90 days' pulls a plan back a
|
||
// quarter -- so this checks validity, not sign.
|
||
try {
|
||
await pool.query(`SELECT $1::interval`, [dateOffset]);
|
||
} catch {
|
||
return res.status(400).json({
|
||
error: `"${dateOffset}" is not a valid interval. Try something like `
|
||
+ `"12 months", "-90 days" or "0 days".`
|
||
});
|
||
}
|
||
|
||
// exclude_iters deliberately does not apply here. It exists to stop
|
||
// operations *modifying* reference rows: scale would attribute forecast
|
||
// movement to prior-year rows by distributing across them, and recode
|
||
// writes negative rows that zero the original out. Clone does neither --
|
||
// it reads rows and writes new pf_iter = 'clone' rows, leaving the source
|
||
// untouched. Copying a plan or a prior year out of reference and into
|
||
// adjustments is the operation working as intended, and excluding them
|
||
// meant a visible, deliberate selection silently produced nothing.
|
||
//
|
||
// from_logid narrows instead: a selection spanning AOP and Prior Year
|
||
// where only one is wanted.
|
||
let excludeClause = '';
|
||
if (from_logid != null) {
|
||
const srcLog = await pool.query(
|
||
`SELECT id FROM pf.log WHERE id = $1 AND version_id = $2`,
|
||
[parseInt(from_logid), ctx.version.id]
|
||
);
|
||
if (!srcLog.rows.length) {
|
||
return res.status(400).json({ error: `No log entry ${from_logid} on this version` });
|
||
}
|
||
excludeClause = `AND pf_logid = ${parseInt(from_logid)}`;
|
||
}
|
||
|
||
// Period dimensions come from the calendar against the shifted date, not
|
||
// from the row being copied -- otherwise a mix moved forward a year
|
||
// keeps last year's period labels. An explicit set wins over both.
|
||
const dateGroups = dateGroupsOf(ctx.colMeta);
|
||
const derivedExprs = Object.fromEntries(
|
||
[...dimPeriodMapOf(dateGroups)].map(([cname, { alias, periodCol }]) =>
|
||
[cname, `${alias}."${periodCol}"`])
|
||
);
|
||
const setClause = buildSetClause(ctx.dimCols, set, { derivedExprs, alias: 's' });
|
||
const units = sliceUnits(slices, ctx, apply_mode === 'each' ? 'each' : 'prorate');
|
||
|
||
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,
|
||
date_offset: dateOffset,
|
||
...(from_logid != null ? { from_logid: parseInt(from_logid) } : {}),
|
||
})),
|
||
slice: esc(JSON.stringify(loggedSlice)),
|
||
where_clause: unit.where,
|
||
exclude_clause: excludeClause,
|
||
set_clause: setClause,
|
||
scale_factor: scaleFactor,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
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_bucket: 'Forecast', 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;
|
||
};
|