Paul Trowbridge
f2734ab5b2
table-info feeds two things and was scoped for neither: the status bar's "total rows", and the figure the load progress promises while it waits. So an account that can see a fraction of the table was told the whole size of it -- jbukowski's 542k rows reported as 2.8M. It only showed on the larger territories. The seeded figure is replaced by X-Row-Count when the headers arrive, and on a small territory the aggregate returns fast enough that the wrong number barely appears. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
442 lines
19 KiB
JavaScript
442 lines
19 KiB
JavaScript
const express = require('express');
|
|
const { fcTable, mapType } = require('../lib/utils');
|
|
const { sessionUser, sessionTerritory } = require('../lib/auth');
|
|
const { buildTerritoryClause } = require('../lib/sql_generator');
|
|
|
|
module.exports = function(pool) {
|
|
const router = express.Router();
|
|
|
|
// list versions for a source
|
|
router.get('/sources/:id/versions', async (req, res) => {
|
|
try {
|
|
const result = await pool.query(
|
|
`SELECT * FROM pf.version WHERE source_id = $1 ORDER BY created_at DESC`,
|
|
[req.params.id]
|
|
);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// create a new version
|
|
// inserts version row, then CREATE TABLE pf.fc_{tname}_{version_id} in one transaction
|
|
router.post('/sources/:id/versions', async (req, res) => {
|
|
const sourceId = parseInt(req.params.id);
|
|
const { name, description, exclude_iters } = req.body;
|
|
const created_by = sessionUser(req);
|
|
if (!name) return res.status(400).json({ error: 'name is required' });
|
|
|
|
const client = await pool.connect();
|
|
try {
|
|
// fetch source
|
|
const srcResult = await client.query(
|
|
`SELECT * FROM pf.source WHERE id = $1`, [sourceId]
|
|
);
|
|
if (srcResult.rows.length === 0) {
|
|
return res.status(404).json({ error: 'Source not found' });
|
|
}
|
|
const source = srcResult.rows[0];
|
|
|
|
// fetch col_meta joined to information_schema for data types
|
|
const colResult = await client.query(`
|
|
SELECT
|
|
m.cname,
|
|
m.role,
|
|
m.opos,
|
|
i.data_type,
|
|
i.numeric_precision,
|
|
i.numeric_scale
|
|
FROM pf.col_meta m
|
|
JOIN information_schema.columns i
|
|
ON i.table_schema = $2
|
|
AND i.table_name = $3
|
|
AND i.column_name = m.cname
|
|
WHERE m.source_id = $1
|
|
AND m.role NOT IN ('ignore')
|
|
ORDER BY m.opos
|
|
`, [sourceId, source.schema, source.tname]);
|
|
|
|
if (colResult.rows.length === 0) {
|
|
return res.status(400).json({
|
|
error: 'No usable columns in col_meta — configure roles before creating a version'
|
|
});
|
|
}
|
|
|
|
await client.query('BEGIN');
|
|
|
|
// insert version to get id
|
|
const verResult = await client.query(`
|
|
INSERT INTO pf.version (source_id, name, description, created_by, exclude_iters)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
RETURNING *
|
|
`, [
|
|
sourceId,
|
|
name,
|
|
description || null,
|
|
created_by || null,
|
|
exclude_iters ? JSON.stringify(exclude_iters) : '["reference"]'
|
|
]);
|
|
const version = verResult.rows[0];
|
|
|
|
// build CREATE TABLE DDL using col_meta + mapped data types
|
|
const table = fcTable(source.tname, version.id);
|
|
const systemCols = new Set(['pf_id', 'pf_iter', 'pf_logid', 'pf_user', 'pf_created_at']);
|
|
const colDefs = colResult.rows
|
|
.filter(c => !systemCols.has(c.cname))
|
|
.map(c => {
|
|
const pgType = mapType(c.data_type, c.numeric_precision, c.numeric_scale);
|
|
const quoted = `"${c.cname}"`;
|
|
return ` ${quoted.padEnd(26)}${pgType}`;
|
|
}).join(',\n');
|
|
|
|
const ddl = `
|
|
CREATE TABLE ${table} (
|
|
pf_id bigserial PRIMARY KEY,
|
|
${colDefs},
|
|
pf_iter text NOT NULL,
|
|
pf_logid bigint NOT NULL,
|
|
pf_user text,
|
|
pf_created_at timestamptz NOT NULL DEFAULT now()
|
|
)
|
|
`;
|
|
await client.query(ddl);
|
|
|
|
// pf_logid is how every entry-level operation finds its rows: undo
|
|
// deletes by it, the change log aggregates by it, and it is part of the
|
|
// grain key. Without an index each of those is a sequential scan of the
|
|
// whole forecast table -- 2.5M rows to total two adjustments.
|
|
await client.query(
|
|
`CREATE INDEX ${table.split('.').pop()}_logid_idx ON ${table} (pf_logid)`
|
|
);
|
|
|
|
await client.query('COMMIT');
|
|
res.status(201).json({ ...version, fc_table: table });
|
|
} catch (err) {
|
|
await client.query('ROLLBACK');
|
|
console.error(err);
|
|
if (err.code === '23505') {
|
|
return res.status(409).json({ error: 'A version with that name already exists for this source' });
|
|
}
|
|
res.status(500).json({ error: err.message });
|
|
} finally {
|
|
client.release();
|
|
}
|
|
});
|
|
|
|
// where this version's writes actually land: the physical forecast table,
|
|
// its current row count, and the source table rows are read from.
|
|
// Surfaced in the status bar so the write target is never a mystery.
|
|
// Distinct values of a dimension as they appear in one version's forecast
|
|
// table, for completing recode and clone.
|
|
//
|
|
// Deliberately not the source: the source view reaches back over all of
|
|
// history, so completing from it offers parts discontinued years ago. The
|
|
// version holds what was actually loaded, which is what a forecast is being
|
|
// written against.
|
|
//
|
|
// Held in memory because the scan is not cheap -- 2.0s for 11,290 parts across
|
|
// 2.5M rows on fc_osm_skinny_29 -- and completion is typed into. Keyed on the
|
|
// version's latest log id, so any load, adjustment or undo rebuilds it on the
|
|
// next request without anything having to remember to invalidate.
|
|
const valueCache = new Map();
|
|
|
|
router.get('/versions/:id/values/:col', async (req, res) => {
|
|
const versionId = parseInt(req.params.id);
|
|
const col = req.params.col;
|
|
try {
|
|
const verResult = await pool.query(`
|
|
SELECT v.id, 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) return res.status(404).json({ error: 'Version not found' });
|
|
const { tname, source_id } = verResult.rows[0];
|
|
|
|
// the column name is interpolated, so it has to be one col_meta names
|
|
const okCol = await pool.query(`
|
|
SELECT 1 FROM pf.col_meta
|
|
WHERE source_id = $1 AND cname = $2 AND role IN ('dimension', 'date')
|
|
`, [source_id, col]);
|
|
if (!okCol.rows.length) {
|
|
return res.status(400).json({ error: `"${col}" is not a dimension on this source` });
|
|
}
|
|
|
|
const table = fcTable(tname, versionId);
|
|
const { rows: [{ rev }] } = await pool.query(
|
|
`SELECT coalesce(max(id), 0)::text AS rev FROM pf.log WHERE version_id = $1`,
|
|
[versionId]
|
|
);
|
|
|
|
const key = `${versionId}:${col}`;
|
|
let entry = valueCache.get(key);
|
|
if (!entry || entry.rev !== rev) {
|
|
const { rows } = await pool.query(
|
|
`SELECT DISTINCT "${col}"::text AS val FROM ${table}
|
|
WHERE "${col}" IS NOT NULL ORDER BY 1`
|
|
);
|
|
entry = { rev, values: rows.map(r => r.val) };
|
|
valueCache.set(key, entry);
|
|
}
|
|
|
|
const q = (req.query.q || '').trim().toLowerCase();
|
|
const limit = Math.min(parseInt(req.query.limit) || 50, 500);
|
|
const picked = q
|
|
? entry.values.filter(v => v.toLowerCase().includes(q))
|
|
: entry.values;
|
|
res.json(picked.slice(0, limit));
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(err.status || 500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
router.get('/versions/:id/table-info', async (req, res) => {
|
|
try {
|
|
const verResult = await pool.query(`
|
|
SELECT v.id, v.name, v.status, 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
|
|
`, [req.params.id]);
|
|
if (verResult.rows.length === 0) return res.status(404).json({ error: 'Version not found' });
|
|
|
|
const v = verResult.rows[0];
|
|
const fc = fcTable(v.tname, v.id);
|
|
const [schema, table] = fc.split('.');
|
|
|
|
const existsResult = await pool.query(
|
|
`SELECT to_regclass($1) IS NOT NULL AS exists`, [fc]
|
|
);
|
|
const exists = existsResult.rows[0].exists;
|
|
|
|
// Scoped like every other read. Unscoped it reported the whole
|
|
// table to an account that can see a fraction of it -- the status
|
|
// bar's row count, and the figure the load progress promises, both
|
|
// came from here: a rep with 330k rows was told the load was
|
|
// fetching 2.8M.
|
|
const terrCol = (await pool.query(
|
|
`SELECT cname FROM pf.col_meta WHERE source_id = $1 AND is_territory LIMIT 1`,
|
|
[v.source_id]
|
|
)).rows[0]?.cname || null;
|
|
const terr = buildTerritoryClause(sessionTerritory(req), terrCol);
|
|
|
|
let rows = null, byIter = [];
|
|
if (exists) {
|
|
const countResult = await pool.query(
|
|
`SELECT pf_iter, count(*)::int AS n FROM ${fc}
|
|
${terr ? `WHERE ${terr}` : ''}
|
|
GROUP BY pf_iter ORDER BY pf_iter`
|
|
);
|
|
byIter = countResult.rows;
|
|
rows = byIter.reduce((a, r) => a + r.n, 0);
|
|
}
|
|
|
|
res.json({
|
|
version_id: v.id,
|
|
version_name: v.name,
|
|
status: v.status,
|
|
source: `${v.schema}.${v.tname}`,
|
|
fc_table: fc,
|
|
fc_schema: schema,
|
|
fc_tname: table,
|
|
exists,
|
|
rows,
|
|
by_iter: byIter
|
|
});
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// Tags already used on this source, newest first — feeds the tag autocomplete.
|
|
// Scoped to the source rather than the version so an initiative name carries
|
|
// across versions, which is the point of naming it.
|
|
router.get('/sources/:id/tags', async (req, res) => {
|
|
try {
|
|
const result = await pool.query(`
|
|
SELECT l.tag,
|
|
count(*)::int AS uses,
|
|
max(l.stamp) AS last_used
|
|
FROM pf.log l
|
|
JOIN pf.version v ON v.id = l.version_id
|
|
WHERE v.source_id = $1 AND l.tag IS NOT NULL AND l.tag <> ''
|
|
GROUP BY l.tag
|
|
ORDER BY max(l.stamp) DESC
|
|
`, [req.params.id]);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// Bridge: how this version got from its baseline to where it stands, grouped by
|
|
// initiative. Amounts come from the version's own forecast table, so the figures
|
|
// reconcile with the pivot rather than being recomputed from the log's params.
|
|
router.get('/versions/:id/bridge', async (req, res) => {
|
|
try {
|
|
const verResult = await pool.query(`
|
|
SELECT v.id, v.exclude_iters, 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
|
|
`, [req.params.id]);
|
|
if (verResult.rows.length === 0) return res.status(404).json({ error: 'Version not found' });
|
|
const v = verResult.rows[0];
|
|
const fc = fcTable(v.tname, v.id);
|
|
|
|
const exists = await pool.query(`SELECT to_regclass($1) IS NOT NULL AS ok`, [fc]);
|
|
if (!exists.rows[0].ok) return res.json({ fc_table: fc, exists: false, rows: [] });
|
|
|
|
const colResult = await pool.query(
|
|
`SELECT cname, role FROM pf.col_meta WHERE source_id = $1`, [v.source_id]);
|
|
const valueCol = colResult.rows.find(c => c.role === 'value')?.cname;
|
|
const unitsCol = colResult.rows.find(c => c.role === 'units')?.cname;
|
|
if (!valueCol) return res.status(400).json({ error: 'No value column configured' });
|
|
|
|
const excl = (v.exclude_iters || []).length
|
|
? `t.pf_iter NOT IN (${v.exclude_iters.map(i => `'${String(i).replace(/'/g, "''")}'`).join(', ')})`
|
|
: 'TRUE';
|
|
|
|
const result = await pool.query(`
|
|
SELECT CASE WHEN t.pf_iter = 'baseline' THEN '(baseline)'
|
|
ELSE coalesce(nullif(l.tag, ''), '(untagged)') END AS tag,
|
|
bool_or(t.pf_iter = 'baseline') AS is_baseline,
|
|
count(DISTINCT l.id)::int AS entries,
|
|
count(*)::int AS row_count,
|
|
round(sum(t."${valueCol}")::numeric, 2) AS value
|
|
${unitsCol ? `, round(sum(t."${unitsCol}")::numeric, 2) AS units` : ''}
|
|
FROM ${fc} t
|
|
LEFT JOIN pf.log l ON l.id = t.pf_logid
|
|
WHERE ${excl}
|
|
GROUP BY 1
|
|
ORDER BY bool_or(t.pf_iter = 'baseline') DESC, min(l.id)
|
|
`);
|
|
res.json({ fc_table: fc, exists: true, value_col: valueCol, units_col: unitsCol, rows: result.rows });
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// update version name, description, exclude_iters, or the fallback display
|
|
// names.
|
|
//
|
|
// bucket_order is deliberately not settable: the bucket column order is the
|
|
// text in pf.log.bucket now, so a stored order would be a second answer to
|
|
// the same question, and a silent one -- nothing reads it.
|
|
//
|
|
// The three name columns are flag-and-value pairs rather than COALESCE:
|
|
// clearing one back to the built-in means writing null, which COALESCE on
|
|
// the value alone cannot tell from "not mentioned".
|
|
router.put('/versions/:id', async (req, res) => {
|
|
const { name, description, exclude_iters,
|
|
adjustment_segment, adjustment_bucket, unlabeled_load } = req.body;
|
|
const set = (v) => (v === undefined ? null : (String(v).trim() || null));
|
|
try {
|
|
const result = await pool.query(`
|
|
UPDATE pf.version SET
|
|
name = COALESCE($2, name),
|
|
description = COALESCE($3, description),
|
|
exclude_iters = COALESCE($4, exclude_iters),
|
|
adjustment_segment = CASE WHEN $5::bool THEN $6::text ELSE adjustment_segment END,
|
|
adjustment_bucket = CASE WHEN $7::bool THEN $8::text ELSE adjustment_bucket END,
|
|
unlabeled_load = CASE WHEN $9::bool THEN $10::text ELSE unlabeled_load END
|
|
WHERE id = $1
|
|
RETURNING *
|
|
`, [
|
|
req.params.id,
|
|
name || null,
|
|
description || null,
|
|
exclude_iters ? JSON.stringify(exclude_iters) : null,
|
|
adjustment_segment !== undefined, set(adjustment_segment),
|
|
adjustment_bucket !== undefined, set(adjustment_bucket),
|
|
unlabeled_load !== undefined, set(unlabeled_load)
|
|
]);
|
|
if (result.rows.length === 0) {
|
|
return res.status(404).json({ error: 'Version not found' });
|
|
}
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// close a version — blocks further edits
|
|
router.post('/versions/:id/close', async (req, res) => {
|
|
const pf_user = sessionUser(req);
|
|
try {
|
|
const result = await pool.query(`
|
|
UPDATE pf.version
|
|
SET status = 'closed', closed_at = now(), closed_by = $2
|
|
WHERE id = $1 AND status = 'open'
|
|
RETURNING *
|
|
`, [req.params.id, pf_user || null]);
|
|
if (result.rows.length === 0) {
|
|
return res.status(404).json({ error: 'Version not found or already closed' });
|
|
}
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// reopen a closed version
|
|
router.post('/versions/:id/reopen', async (req, res) => {
|
|
try {
|
|
const result = await pool.query(`
|
|
UPDATE pf.version
|
|
SET status = 'open', closed_at = NULL, closed_by = NULL
|
|
WHERE id = $1 AND status = 'closed'
|
|
RETURNING *
|
|
`, [req.params.id]);
|
|
if (result.rows.length === 0) {
|
|
return res.status(404).json({ error: 'Version not found or already open' });
|
|
}
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
}
|
|
});
|
|
|
|
// delete a version — drops forecast table then deletes version record
|
|
// log entries are removed by ON DELETE CASCADE on pf.log.version_id
|
|
router.delete('/versions/:id', async (req, res) => {
|
|
const versionId = parseInt(req.params.id);
|
|
const client = await pool.connect();
|
|
try {
|
|
const verResult = await client.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 === 0) {
|
|
return res.status(404).json({ error: 'Version not found' });
|
|
}
|
|
const { tname } = verResult.rows[0];
|
|
const table = fcTable(tname, versionId);
|
|
|
|
await client.query('BEGIN');
|
|
await client.query(`DROP TABLE IF EXISTS ${table}`);
|
|
await client.query(`DELETE FROM pf.version WHERE id = $1`, [versionId]);
|
|
await client.query('COMMIT');
|
|
|
|
res.json({ message: 'Version deleted', fc_table: table });
|
|
} catch (err) {
|
|
await client.query('ROLLBACK');
|
|
console.error(err);
|
|
res.status(500).json({ error: err.message });
|
|
} finally {
|
|
client.release();
|
|
}
|
|
});
|
|
|
|
return router;
|
|
};
|