Paul Trowbridge
915981b4a7
Nav. The source and version dropdowns sat in a status bar that is otherwise read-only, so they did not read as the controls they were. The forecasts are now listed in the sidebar above the three steps, each row carrying its status and a star for the default. Not nested under the Forecast button, which is where this started: Baseline works on the same version, so hanging the list off one step would either misrepresent the scope or duplicate it under two. The source is gone from the nav entirely -- gs.osm_skinny means nothing to someone editing a forecast. The sidebar lists every source's versions and sets the source from the one picked, since a version carries its source_id; `versions` stays scoped to the selected source, because Baseline and Forecast are, so the sidebar gets its own list rather than widening what versions means everywhere. The picker only appears on Setup now, which is the one place the raw table name is the right label. The status bar keeps what follows FROM the selection -- the forecast's name and its row count. The table it writes to moved into the hover panel, still click-to-copy, since pasting it into psql is why it was ever on screen. Permissions. requireAdmin existed in lib/auth.js and was applied to nothing, so all 26 non-admin accounts could call every Setup and Baseline endpoint directly. It is now on source register/deregister, col_meta writes, Generate SQL, dim refresh, version create/delete/ close/reopen, PUT /versions/:id, all of plans.js, and the baseline and reference loads. Deliberately not on any read, nor on the forecast operations, which carry their own owner and territory rules. Both pages stay visible and go read-only, via one <fieldset disabled> each rather than a flag threaded through forty-five controls, with a banner saying why. The Tags panel on Baseline is excluded: its permission is ownership, not admin, and a rep may rename the initiatives they named themselves. The table browser reads are open to everyone, because gating what a page needs in order to render turned Setup into a 403 -- which it then rendered as a crash, since the error body was passed to .map during render and unmounting React blanks the app. Setup's list fetches now coerce to an array, so an unexpected response costs an empty panel and not the session. That fragility predated the gating. Toolbar. The zone labels were the same visual weight as the buttons they name, and the rules between groups were the same weight as the zone borders, so nothing separated at the level it belonged to. Labels are heavier and darker, group rules are stronger and taller, and the forecast zone has a container of its own instead of floating. Fit widths is removed, button and function. It measured the leaf header over viewport rows only, so anything below the fold never counted; it passed an option draw() does not have and raced draw()'s throttle, leaving columns cleared but unmeasured at their floor widths; and the persist that followed saved nothing, because saveColumnSizes returns only overrides, which resetAutoSize had just emptied. Recoverable from git if it is ever worth fixing rather than dropping. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1050 lines
51 KiB
JavaScript
1050 lines
51 KiB
JavaScript
const express = require('express');
|
||
const { tableFromArrays, tableToIPC } = require('apache-arrow');
|
||
const { applyTokens, buildWhere, buildWhereAny, COMPUTED_SLICE_COLS, buildScopeClause, buildTerritoryClause, buildExcludeClause, buildExcludePredicate, buildSetClause, dateGroupsOf, dimPeriodMapOf, esc,
|
||
SEGMENT_EXPR, BUCKET_EXPR, NOTE_EXPR, VERSION_JOIN,
|
||
ADJUSTMENT_SEGMENT, ADJUSTMENT_BUCKET } = require('../lib/sql_generator');
|
||
const { sessionUser, sessionTerritory, requireAdmin } = 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.
|
||
// The scope is ANDed onto every unit rather than folded into the slices: it
|
||
// applies to all of them equally, and under apply_mode 'each' folding it in
|
||
// would repeat the same predicate in every statement for no gain.
|
||
function sliceUnits(slices, ctx, applyMode, scope, req) {
|
||
const vid = ctx.version.id;
|
||
const scl = buildScopeClause(scope, ctx.filterCols, vid);
|
||
const terr = req ? territoryOf(req, ctx) : '';
|
||
const and = (w) => {
|
||
let out = w;
|
||
for (const extra of [scl, terr]) {
|
||
if (!extra) continue;
|
||
out = (out === 'TRUE' || !out) ? extra : `${out}\nAND ${extra}`;
|
||
}
|
||
return out;
|
||
};
|
||
return applyMode === 'each'
|
||
? slices.map(sl => ({ slices: [sl], where: and(buildWhere(sl, ctx.filterCols, vid)) }))
|
||
: [{ slices, where: and(buildWhereAny(slices, ctx.filterCols, vid)) }];
|
||
}
|
||
|
||
// The offset is interpolated into the statement as an interval literal, so a
|
||
// typo would surface as a Postgres parse error from the middle of a CTE. Ask
|
||
// Postgres to parse it alone first, where the failure is cheap and can name the
|
||
// field it came from. Negative intervals are valid and useful -- '-90 days'
|
||
// pulls a plan back a quarter -- so this checks validity, not sign.
|
||
async function assertInterval(value, res) {
|
||
try {
|
||
await pool.query(`SELECT $1::interval`, [value]);
|
||
return true;
|
||
} catch {
|
||
res.status(400).json({
|
||
error: `"${value}" is not a valid interval. Try something like `
|
||
+ `"4 months", "1 year", "-90 days" or "0 days".`
|
||
});
|
||
return false;
|
||
}
|
||
}
|
||
|
||
// 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, ...Object.keys(COMPUTED_SLICE_COLS)]);
|
||
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;
|
||
}
|
||
});
|
||
}
|
||
|
||
// Stamp what the entry did onto the entry itself.
|
||
//
|
||
// An indexed lookup on pf_logid, run once at write time, in place of the
|
||
// change log joining the whole forecast table on every open. Safe to store
|
||
// rather than derive because these rows never change: only this operation
|
||
// inserts them, and the only thing that removes them is undo, which deletes
|
||
// the log row too.
|
||
//
|
||
// Best-effort by design. A failure here must not roll back a write that
|
||
// succeeded -- the totals can always be recomputed, the adjustment cannot.
|
||
async function stampLogTotals(client, ctx, logId, extra = {}) {
|
||
if (!logId) return;
|
||
const v = ctx.valueCol, u = ctx.unitsCol;
|
||
// The state the write ran against, none of which can be reconstructed
|
||
// later: territory and exclude_iters are mutable rows elsewhere, and the
|
||
// template is overwritten in place every time Generate SQL runs. The
|
||
// generation timestamp is a fingerprint, not a version -- it cannot bring
|
||
// the old template back, only tell you the entry did not run under this
|
||
// one.
|
||
const env = {
|
||
territory: extra.territory ?? null,
|
||
territory_col: ctx.territoryCol || null,
|
||
exclude_iters: ctx.version.exclude_iters ?? null,
|
||
sql_generated_at: ctx.sqlGeneratedAt || null,
|
||
};
|
||
try {
|
||
await client.query(`
|
||
UPDATE pf.log SET
|
||
row_count = t.n,
|
||
value_total = t.v,
|
||
units_total = t.u,
|
||
measure_cols = $2::jsonb,
|
||
env = $3::jsonb,
|
||
sql_text = $4::text
|
||
FROM (
|
||
SELECT count(*)::int AS n
|
||
,${v ? `sum(f."${v}")::float8` : 'NULL::float8'} AS v
|
||
,${u ? `sum(f."${u}")::float8` : 'NULL::float8'} AS u
|
||
FROM ${ctx.table} f
|
||
WHERE f.pf_logid = $1
|
||
) t
|
||
WHERE pf.log.id = $1
|
||
`, [logId, JSON.stringify({ value: v || null, units: u || null }),
|
||
JSON.stringify(env), extra.sql || null]);
|
||
} catch (err) {
|
||
console.error('[stampLogTotals]', err);
|
||
}
|
||
}
|
||
|
||
// Moving a row between territories is reassignment, not forecasting, so a
|
||
// scoped account cannot set the territory column -- otherwise a rep could
|
||
// recode work into their own book, or quietly out of it, and the row would
|
||
// be gone from the view that would have shown what happened.
|
||
//
|
||
// Reads and writes are already scoped, so the *source* rows are safely the
|
||
// account's own; this is only about the destination.
|
||
function assertMayRecodeTerritory(req, ctx, set, res) {
|
||
if (!ctx.territoryCol) return true;
|
||
if (req.session?.user?.is_admin) return true;
|
||
if (!set || set[ctx.territoryCol] === undefined) return true;
|
||
res.status(403).json({
|
||
error: `Only an administrator can recode ${ctx.territoryCol} — that moves rows between territories`
|
||
});
|
||
return false;
|
||
}
|
||
|
||
// 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', 'scope'];
|
||
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.
|
||
//
|
||
// 'volume' — price holds, so volume scales with the dollars. The default.
|
||
// 'price' — volume holds, so price moves.
|
||
//
|
||
// Only meaningful when dollars were the input and units were not given
|
||
// explicitly; naming both means the caller has already decided.
|
||
//
|
||
// 'price' was the default, being what the API did before the flag existed,
|
||
// and that made every unqualified dollar edit a price edit. A forecast
|
||
// that moves price when nobody asked is wrong in a way that is invisible
|
||
// -- the dollars land on target either way -- so the default is now the
|
||
// conservative one and moving price takes saying so.
|
||
const plug = body.plug === 'price' ? 'price' : 'volume';
|
||
const unitsGiven = [tUnits, uIncr, uPct].some(v => v !== null);
|
||
const valueGiven = [tValue, vIncr, vPct].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 same question asked of a qty input, and the answer has to match:
|
||
// with dollars unnamed they follow at the current price, rather than
|
||
// holding while price absorbs the difference.
|
||
//
|
||
// Holding them made every qty edit a price edit -- move 10% more cases
|
||
// at the same revenue and the implied price drops 9% -- which is exactly
|
||
// the silent price change the plug default above exists to prevent. The
|
||
// deliberate forms are still available and both outrank this: name the
|
||
// dollars alongside the qty (then price falls out of the two), or name
|
||
// the price outright.
|
||
if (units !== 0 && !valueGiven && tPrice === null) {
|
||
const curValue = totals.value + fixedValue;
|
||
const curUnits = totals.units + fixedUnits;
|
||
// no units means no price to hold, so the dollars stay where they are
|
||
// and the edit is a pure qty change -- adding units to a slice that
|
||
// has none cannot infer what they are worth
|
||
if (curUnits !== 0) value = (curValue / curUnits) * units;
|
||
}
|
||
|
||
// 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, generated_at 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,
|
||
territoryCol: colMeta.find(c => c.is_territory)?.cname || null,
|
||
sql: sqlResult.rows[0].sql,
|
||
sqlGeneratedAt: sqlResult.rows[0].generated_at
|
||
};
|
||
}
|
||
|
||
// Every read and every write goes through this, so a scoped account cannot
|
||
// reach a row outside its territory by any route. ANDed on last, after the
|
||
// slice and the client's own scope, where nothing in the request can undo
|
||
// it.
|
||
function territoryOf(req, ctx) {
|
||
return buildTerritoryClause(sessionTerritory(req), ctx.territoryCol);
|
||
}
|
||
|
||
function andTerritory(where, req, ctx) {
|
||
const t = territoryOf(req, ctx);
|
||
if (!t) return where;
|
||
return (!where || where === 'TRUE') ? t : `${where}\nAND ${t}`;
|
||
}
|
||
|
||
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);
|
||
|
||
// /data does not go through getContext, so it resolves the territory
|
||
// column itself. The count is scoped too, or the progress bar
|
||
// promises rows this account will never be sent.
|
||
const terrCol = (await pool.query(
|
||
`SELECT cname FROM pf.col_meta WHERE source_id = $1 AND is_territory LIMIT 1`,
|
||
[verResult.rows[0].source_id]
|
||
)).rows[0]?.cname || null;
|
||
const territory = sessionTerritory(req);
|
||
const terrBare = buildTerritoryClause(territory, terrCol);
|
||
const terrAlias = buildTerritoryClause(territory, terrCol, 't');
|
||
const terrWhere = terrBare ? `WHERE ${terrBare}` : '';
|
||
|
||
const { rows: [{ count }] } = await pool.query(`SELECT COUNT(*) FROM ${tbl} ${terrWhere}`);
|
||
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.*
|
||
,${SEGMENT_EXPR} AS pf_segment
|
||
,${BUCKET_EXPR} AS pf_bucket
|
||
,${NOTE_EXPR} AS pf_note
|
||
FROM ${tbl} t
|
||
LEFT JOIN pf.log l
|
||
ON l.id = t.pf_logid${VERSION_JOIN}
|
||
${terrAlias ? `WHERE ${terrAlias}` : ''}
|
||
`);
|
||
|
||
// 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) => {
|
||
let client;
|
||
let committed = false;
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'get_agg');
|
||
// Before the GROUP BY, not after: the territory column need not be
|
||
// part of the grain, so an aggregated row may not carry it at all.
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
territory_clause:
|
||
buildTerritoryClause(sessionTerritory(req), ctx.territoryCol, 't') || 'TRUE',
|
||
});
|
||
|
||
// The GROUP BY spans every in_grain column plus pf_logid and the
|
||
// pf.log/pf.version columns joined in for segment/bucket — 30+ keys
|
||
// on a wide source. The planner sees pf_logid arrive pre-sorted off
|
||
// its index and picks a sort-based GroupAggregate on the strength of
|
||
// that "free" sort, but comparing that many columns per row makes it
|
||
// 5-7x slower in practice than a HashAggregate — confirmed via
|
||
// EXPLAIN ANALYZE on fc_osm_skinny_34 (34-50s sorted vs ~7s hashed,
|
||
// independent of work_mem or stats). Scoped to this transaction only:
|
||
// disabling incremental sort globally could hurt some other query's plan.
|
||
client = await pool.connect();
|
||
await client.query('BEGIN');
|
||
await client.query('SET LOCAL enable_sort = off');
|
||
await client.query('SET LOCAL enable_incremental_sort = off');
|
||
const { rows } = await runSQL(sql, client);
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
|
||
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();
|
||
} 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', requireAdmin, async (req, res) => {
|
||
const { where_clause, date_offset, note, filters, raw_where, label, bucket, tag } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const dateOffset = date_offset || '0 days';
|
||
if (!await assertInterval(dateOffset, res)) return;
|
||
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 || ''),
|
||
label: esc(label || ''),
|
||
bucket: esc(bucket || ''),
|
||
tag: esc(tag || ''),
|
||
params: esc(paramsJson),
|
||
filter_clause: filterClause,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
|
||
const result = await runSQL(sql);
|
||
await stampLogTotals(pool, ctx, result.rows[0]?.log_id, { sql, territory: sessionTerritory(req) });
|
||
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', requireAdmin, async (req, res) => {
|
||
const versionId = parseInt(req.params.id);
|
||
const logid = parseInt(req.params.logid);
|
||
const { where_clause, date_offset, note, filters, raw_where, label, bucket, tag } = req.body;
|
||
const pf_user = sessionUser(req);
|
||
const dateOffset = date_offset || '0 days';
|
||
if (!await assertInterval(dateOffset, res)) return;
|
||
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 } : {}))
|
||
});
|
||
// This route deletes the log row and inserts a fresh one, so every
|
||
// annotation on it has to be handed back or it is lost. `??`, not `||`:
|
||
// an empty string is the form clearing a field on purpose, undefined is
|
||
// the form not carrying it at all -- the segment form has no tag input,
|
||
// so tag is always the latter and must survive an edit made for any
|
||
// other reason.
|
||
const keep = (sent, prior) => esc(sent ?? prior ?? '');
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
label: keep(label, oldLog.label),
|
||
bucket: keep(bucket, oldLog.bucket),
|
||
tag: keep(tag, oldLog.tag),
|
||
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');
|
||
await stampLogTotals(pool, ctx, insResult.rows[0]?.log_id, { sql, territory: sessionTerritory(req) });
|
||
|
||
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', requireAdmin, 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', requireAdmin, async (req, res) => {
|
||
const { where_clause, date_offset, note, filters, raw_where, label, bucket, tag } = 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 || ''),
|
||
label: esc(label || ''),
|
||
bucket: esc(bucket || ''),
|
||
tag: esc(tag || ''),
|
||
params: esc(paramsJson),
|
||
filter_clause: filterClause,
|
||
date_offset: esc(dateOffset)
|
||
});
|
||
|
||
const result = await runSQL(sql);
|
||
await stampLogTotals(pool, ctx, result.rows[0]?.log_id, { sql, territory: sessionTerritory(req) });
|
||
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, req.body.scope, req);
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
// the statement that produced each entry, for pf.log.sql_text
|
||
const sqlByLogId = new Map();
|
||
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);
|
||
for (const r of result.rows) {
|
||
if (r.pf_logid != null) sqlByLogId.set(r.pf_logid, sql);
|
||
}
|
||
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;
|
||
// one log id per unit: apply_mode 'each' writes an entry per slice
|
||
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
|
||
await stampLogTotals(pool, ctx, id, {
|
||
sql: sqlByLogId.get(id) || null,
|
||
territory: sessionTerritory(req),
|
||
});
|
||
}
|
||
const opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, 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);
|
||
if (!assertMayRecodeTerritory(req, ctx, set, res)) return;
|
||
|
||
const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
|
||
const setClause = buildSetClause(ctx.dimCols, set);
|
||
const units = sliceUnits(slices, ctx, apply_mode === 'each' ? 'each' : 'prorate', req.body.scope, req);
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
// the statement that produced each entry, for pf.log.sql_text
|
||
const sqlByLogId = new Map();
|
||
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);
|
||
for (const r of result.rows) {
|
||
if (r.pf_logid != null) sqlByLogId.set(r.pf_logid, sql);
|
||
}
|
||
allRows.push(...result.rows);
|
||
}
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
// one log id per unit: apply_mode 'each' writes an entry per slice
|
||
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
|
||
await stampLogTotals(pool, ctx, id, {
|
||
sql: sqlByLogId.get(id) || null,
|
||
territory: sessionTerritory(req),
|
||
});
|
||
}
|
||
const opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, 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';
|
||
|
||
if (!await assertInterval(dateOffset, res)) return;
|
||
|
||
// 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', req.body.scope, req);
|
||
|
||
const client = await pool.connect();
|
||
let committed = false;
|
||
try {
|
||
await client.query('BEGIN');
|
||
const allRows = [];
|
||
// the statement that produced each entry, for pf.log.sql_text
|
||
const sqlByLogId = new Map();
|
||
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);
|
||
for (const r of result.rows) {
|
||
if (r.pf_logid != null) sqlByLogId.set(r.pf_logid, sql);
|
||
}
|
||
allRows.push(...result.rows);
|
||
}
|
||
await client.query('COMMIT');
|
||
committed = true;
|
||
// one log id per unit: apply_mode 'each' writes an entry per slice
|
||
for (const id of new Set(allRows.map(r => r.pf_logid).filter(Boolean))) {
|
||
await stampLogTotals(pool, ctx, id, {
|
||
sql: sqlByLogId.get(id) || null,
|
||
territory: sessionTerritory(req),
|
||
});
|
||
}
|
||
const opLabel = (req.body.tag || '').trim() || note || null;
|
||
const rows = allRows.map(r => ({ ...r, pf_segment: ADJUSTMENT_SEGMENT, pf_bucket: ADJUSTMENT_BUCKET, 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;
|
||
};
|