Pointing completion at the source made every keystroke a 76-second query: gs.osm_skinny is a plain view over rlarp.osm_stack, so ILIKE '%1601%' scanned 6.9M rows and read 1.3M buffers off disk, 64s of it I/O. Nothing had called that endpoint before, so the cost only appeared once a debounced input was wired to it. The source is the wrong list anyway. It reaches back over all of history and would offer parts discontinued years ago; the version holds what was actually loaded, which is what the forecast is being written against. So GET /versions/:id/values/:col reads the version's own forecast table -- 2.0s for 11,290 parts on fc_osm_skinny_29 -- and holds the result in memory, keyed on that version's latest pf.log id. Any load, adjustment or undo moves the id and the next request rebuilds, so nothing has to remember to invalidate. Filtering happens over the cached array, so typing costs one small max(id) query. The column name is interpolated into the DISTINCT, so it is checked against col_meta first. The source-side endpoint keeps its new q/limit but no longer has a caller; its ILIKE path against a view is the expensive one and should stay unused. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
411 lines
17 KiB
JavaScript
411 lines
17 KiB
JavaScript
const express = require('express');
|
|
const { fcTable, mapType } = require('../lib/utils');
|
|
const { sessionUser } = require('../lib/auth');
|
|
|
|
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
|
|
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;
|
|
|
|
let rows = null, byIter = [];
|
|
if (exists) {
|
|
const countResult = await pool.query(
|
|
`SELECT pf_iter, count(*)::int AS n FROM ${fc} 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, or exclude_iters
|
|
router.put('/versions/:id', async (req, res) => {
|
|
const { name, description, exclude_iters } = req.body;
|
|
try {
|
|
const result = await pool.query(`
|
|
UPDATE pf.version SET
|
|
name = COALESCE($2, name),
|
|
description = COALESCE($3, description),
|
|
exclude_iters = COALESCE($4, exclude_iters)
|
|
WHERE id = $1
|
|
RETURNING *
|
|
`, [
|
|
req.params.id,
|
|
name || null,
|
|
description || null,
|
|
exclude_iters ? JSON.stringify(exclude_iters) : null
|
|
]);
|
|
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;
|
|
};
|