Paul Trowbridge
1c40e5b7b0
Spreading a large RETURNING * result into push() overflowed the call stack on scale, recode and clone. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
1193 lines
60 KiB
JavaScript
1193 lines
60 KiB
JavaScript
const express = require('express');
|
||
const { tableFromArrays, tableToIPC } = require('apache-arrow');
|
||
const { applyTokens, buildWhere, buildWhereAny, COMPUTED_SLICE_COLS, TABLE_SLICE_COLS, buildScopeClause, buildTerritoryClause, buildExcludeClause, buildExcludePredicate, buildSetClause, dateGroupsOf, dimPeriodMapOf, esc,
|
||
SEGMENT_EXPR, BUCKET_EXPR, NOTE_EXPR,
|
||
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();
|
||
|
||
// The segment and bucket an adjustment is stamped with: written onto its
|
||
// pf.log entry by the template, and onto the rows pushed back to the pivot,
|
||
// so the two agree. The version's own defaults when set, else the
|
||
// built-ins. Resolved at write time, so changing a version's default
|
||
// afterwards names new adjustments and leaves existing ones as they are.
|
||
function adjustmentNames(version) {
|
||
const pick = (v, dflt) => (v || '').trim() || dflt;
|
||
return {
|
||
pf_segment: pick(version?.adjustment_segment, ADJUSTMENT_SEGMENT),
|
||
pf_bucket: pick(version?.adjustment_bucket, ADJUSTMENT_BUCKET),
|
||
};
|
||
}
|
||
|
||
// The segment and bucket a load is stamped with. A load's label is its
|
||
// segment name and is required -- it used to fall back through tag, note and
|
||
// the version's unlabeled_load at read time, which left pf.log unable to say
|
||
// what a row was called. The bucket is the label unless one is given, which
|
||
// is how an unbucketed load has always read. Sends the 400 and returns null
|
||
// when there is no label.
|
||
function loadNames(label, bucket, res) {
|
||
const segment = String(label ?? '').trim();
|
||
if (!segment) {
|
||
res.status(400).json({ error: 'Give the segment a label — it is what the pivot shows and what reports group on' });
|
||
return null;
|
||
}
|
||
return { label: segment, bucket: String(bucket ?? '').trim() || segment };
|
||
}
|
||
|
||
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.
|
||
// An empty slice is a total cell -- everything the pivot's filter shows -- so
|
||
// it is kept only when a scope came with it; without one it would be the
|
||
// whole version, and is dropped 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];
|
||
const scoped = Array.isArray(body.scope) && body.scope.length > 0;
|
||
return raw.filter(s => s && typeof s === 'object' && (scoped || 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.
|
||
// A date offset has to be a whole number of days. Every date it shifts is
|
||
// cast back to a date, so anything finer is rounded away -- and Postgres
|
||
// reads a bare number as seconds, so "30" meant thirty seconds, moved
|
||
// nothing, and a shifted recode took rows out of a cell and put them back
|
||
// into the same one.
|
||
async function assertInterval(value, res) {
|
||
const example = `Try something like "30 days", "1 month", "-90 days" or "0 days".`;
|
||
let whole;
|
||
try {
|
||
whole = (await pool.query(
|
||
`SELECT date_trunc('day', $1::interval) = $1::interval AS whole`, [value]
|
||
)).rows[0].whole;
|
||
} catch {
|
||
res.status(400).json({ error: `"${value}" is not a valid interval. ${example}` });
|
||
return false;
|
||
}
|
||
if (!whole) {
|
||
res.status(400).json({
|
||
error: `"${value}" is not a whole number of days, so it would not move any date `
|
||
+ `(a number with no unit is read as seconds). ${example}`
|
||
});
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
// 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.
|
||
// An empty slice passes only under a scope (see normalizeSlices), which
|
||
// buildScopeClause refuses to let reduce to nothing.
|
||
function assertSelective(slices, ctx, scope) {
|
||
// pf_iter, pf_note, pf_op, pf_user and pf_logid narrow a slice once it
|
||
// is selective, but do not make it so: they were only just accepted at
|
||
// all, and `pf_iter = scale` alone is every adjustment in the version
|
||
const allowed = new Set([...ctx.filterCols, 'pf_segment', 'pf_bucket']);
|
||
const scoped = Array.isArray(scope) && scope.length > 0;
|
||
slices.forEach((sl, i) => {
|
||
if (scoped && Object.keys(sl).length === 0) return;
|
||
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', 'plug',
|
||
// What the panel's Result line promised, kept beside
|
||
// `resolved` so a disagreement between the two can be
|
||
// found by reading the log. Not checked -- yet.
|
||
'expected'];
|
||
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. The price
|
||
// named holds, and whichever other measure was named decides the rest:
|
||
// price alone qty holds, dollars = price x qty
|
||
// price + qty qty lands as named (target, change or %),
|
||
// dollars = price x that qty
|
||
// price + dollars dollars land as named, qty = dollars / price
|
||
// all three refused -- two degrees of freedom, three numbers
|
||
// All on the basis the panel measures against, so under 'selected' the
|
||
// price is that of the whole selection, fixed rows included.
|
||
//
|
||
// This used to read only a qty *target*, so a qty change or % sized the
|
||
// dollars on the old qty and missed the price; took the fixed qty off a
|
||
// qty target but not the fixed dollars off the value, so under 'selected'
|
||
// it landed short; and let a dollar figure outrank the price outright.
|
||
const unitsGiven = [tUnits, uIncr, uPct].some(v => v !== null);
|
||
const valueGiven = [tValue, vIncr, vPct].some(v => v !== null);
|
||
if (tPrice !== null) {
|
||
const bad = (msg) => { const err = new Error(msg); err.status = 400; return err; };
|
||
if (valueGiven && unitsGiven) {
|
||
throw bad('Price, dollars and qty were all given. Any two decide the third -- clear one of them.');
|
||
}
|
||
const curValue = totals.value + fixedValue;
|
||
const curUnits = totals.units + fixedUnits;
|
||
if (valueGiven) {
|
||
if (!(tPrice > 0)) throw bad('A price of zero or less cannot set qty from dollars.');
|
||
units = (curValue + value) / tPrice - curUnits;
|
||
} else {
|
||
value = tPrice * (curUnits + units) - curValue;
|
||
}
|
||
}
|
||
|
||
// 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';
|
||
|
||
// The price held is the *adjustable* rows' own, never a blend with the
|
||
// fixed ones: the scale SQL prorates both increments over the adjustable
|
||
// rows only, so a units increment sized at the blended price cuts qty
|
||
// faster than dollars whenever the fixed rows are priced differently.
|
||
// Taking a slice to its fixed total -- dollars to zero -- then drove qty
|
||
// past zero, leaving grains with positive sales and negative qty.
|
||
//
|
||
// Not when a price was named. That is an edit-price, not a dollar edit:
|
||
// qty holds and the dollars were sized off the price above. The panel
|
||
// greys the plug out then and does not send it, so `plug` reads as its
|
||
// 'volume' default, and this block used to scale qty by the dollar
|
||
// change as well -- which put price straight back where it started.
|
||
if (plug === 'volume' && value !== 0 && !unitsGiven && tPrice === null) {
|
||
if (totals.value === 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 = adjVol * (fVal / adjVal), so the units delta is adjVol * value/adjVal
|
||
units = totals.units * (value / totals.value);
|
||
}
|
||
|
||
// 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.
|
||
// Again at the adjustable rows' own price, for the reason given above.
|
||
if (units !== 0 && !valueGiven && tPrice === null) {
|
||
// 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 (totals.units !== 0) value = (totals.value / totals.units) * 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);
|
||
|
||
// Which of price or volume the edit moved, written onto the log entry.
|
||
// The request's own fields do not say -- `plug` is a per-browser
|
||
// preference, and an unqualified dollar figure is a price edit or a
|
||
// volume edit depending on it -- so without this the log could not tell
|
||
// the two apart, and the only way to find a price edit afterwards was to
|
||
// notice qty had not moved.
|
||
const method =
|
||
tPrice !== null ? 'price' // price named
|
||
: unitsGiven && valueGiven ? 'both' // dollars and qty named, price falls out
|
||
: unitsGiven ? 'volume' // qty named, dollars follow at price
|
||
: plug; // dollars only: the plug decides
|
||
|
||
return { value: round(value, 6), units: round(units, 6), method };
|
||
}
|
||
|
||
// 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
|
||
${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';
|
||
const names = loadNames(label, bucket, res);
|
||
if (!names) return;
|
||
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(names.label),
|
||
bucket: esc(names.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 names = loadNames(label ?? oldLog.label, bucket ?? oldLog.bucket, res);
|
||
if (!names) return;
|
||
const sql = applyTokens(ctx.sql, {
|
||
fc_table: ctx.table,
|
||
version_id: ctx.version.id,
|
||
pf_user: esc(pf_user || ''),
|
||
note: esc(note || ''),
|
||
label: esc(names.label),
|
||
bucket: esc(names.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';
|
||
const names = loadNames(label, bucket, res);
|
||
if (!names) return;
|
||
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(names.label),
|
||
bucket: esc(names.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, req.body.scope);
|
||
|
||
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 names = adjustmentNames(ctx.version);
|
||
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) { for (const s of unit.slices) skipped.push(s); 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 || ''),
|
||
label: esc(names.pf_segment),
|
||
bucket: esc(names.pf_bucket),
|
||
params: esc(JSON.stringify({
|
||
slices: unit.slices,
|
||
apply_mode: applyMode,
|
||
...pickIntent(req.body),
|
||
resolved: { value_incr: incr.value, units_incr: incr.units, method: incr.method }
|
||
})),
|
||
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);
|
||
}
|
||
for (const r of result.rows) allRows.push(r);
|
||
}
|
||
|
||
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, ...names, 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
|
||
//
|
||
// date_offset moves the positive rows in time as well -- the negative leg stays
|
||
// where the rows were, so the pair is a timing move (pull forward, push out)
|
||
// rather than a copy. With an offset, set may be empty: moving a slice without
|
||
// renaming anything is the whole request.
|
||
router.post('/versions/:id/recode', async (req, res) => {
|
||
const { note, set, apply_mode, 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' });
|
||
const dateOffset = String(date_offset ?? '').trim() || '0 days';
|
||
if (!await assertInterval(dateOffset, res)) return;
|
||
const shifted = (await pool.query(`SELECT $1::interval <> '0'::interval AS s`, [dateOffset])).rows[0].s;
|
||
if ((!set || Object.keys(set).length === 0) && !shifted) {
|
||
return res.status(400).json({ error: 'Recode needs a new value or a date offset' });
|
||
}
|
||
|
||
try {
|
||
const ctx = await getContext(parseInt(req.params.id), 'recode');
|
||
if (!guardOpen(ctx.version, res)) return;
|
||
assertSelective(slices, ctx, req.body.scope);
|
||
if (!assertMayRecodeTerritory(req, ctx, set, res)) return;
|
||
|
||
const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
|
||
|
||
// Moved rows take their period dimensions from the calendar against the
|
||
// shifted date, as clone does; an explicit set still wins. A row whose
|
||
// date is null has nothing to shift and keeps the label it had, rather
|
||
// than having it blanked by a calendar lookup that found nothing. Not
|
||
// derived at all without an offset, so a plain recode copies them as
|
||
// it always has.
|
||
const dateGroups = dateGroupsOf(ctx.colMeta);
|
||
const dateColOf = Object.fromEntries(dateGroups.map(g => [g.alias, g.dateCol]));
|
||
const derivedExprs = shifted ? Object.fromEntries(
|
||
[...dimPeriodMapOf(dateGroups)].map(([cname, { alias, periodCol }]) =>
|
||
[cname, `CASE WHEN s."${dateColOf[alias]}" IS NULL THEN s."${cname}" ELSE ${alias}."${periodCol}" END`])
|
||
) : undefined;
|
||
const setClause = buildSetClause(ctx.dimCols, set, { derivedExprs, alias: 's' });
|
||
const units = sliceUnits(slices, ctx, apply_mode === 'each' ? 'each' : 'prorate', req.body.scope, req);
|
||
|
||
const names = adjustmentNames(ctx.version);
|
||
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 || ''),
|
||
label: esc(names.pf_segment),
|
||
bucket: esc(names.pf_bucket),
|
||
params: esc(JSON.stringify({ slices: unit.slices, set, apply_mode: unit.mode, date_offset: dateOffset })),
|
||
slice: esc(JSON.stringify(loggedSlice)),
|
||
where_clause: unit.where,
|
||
exclude_clause: excludeClause,
|
||
set_clause: setClause,
|
||
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);
|
||
}
|
||
for (const r of result.rows) allRows.push(r);
|
||
}
|
||
// Every coordinate in the selection nets to zero -- nothing to move.
|
||
// Refused like scale's no-op rather than committing an empty log
|
||
// entry that reads as a recode that happened.
|
||
if (allRows.length === 0) {
|
||
await client.query('ROLLBACK');
|
||
return res.status(400).json({
|
||
error: 'Nothing to recode — the selection nets to zero, or holds only rows recode cannot touch'
|
||
});
|
||
}
|
||
|
||
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, ...names, 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, req.body.scope);
|
||
|
||
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 names = adjustmentNames(ctx.version);
|
||
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 || ''),
|
||
label: esc(names.pf_segment),
|
||
bucket: esc(names.pf_bucket),
|
||
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);
|
||
}
|
||
for (const r of result.rows) allRows.push(r);
|
||
}
|
||
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, ...names, 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;
|
||
};
|