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