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, bucket_order } = req.body; try { // bucket_order is a flag-and-value pair rather than COALESCE: an empty // array is a meaningful value (no ordering), and COALESCE could not tell // it from "not mentioned". const result = await pool.query(` UPDATE pf.version SET name = COALESCE($2, name), description = COALESCE($3, description), exclude_iters = COALESCE($4, exclude_iters), bucket_order = CASE WHEN $5::bool THEN $6::jsonb ELSE bucket_order END WHERE id = $1 RETURNING * `, [ req.params.id, name || null, description || null, exclude_iters ? JSON.stringify(exclude_iters) : null, bucket_order !== undefined, bucket_order === undefined ? null : JSON.stringify(bucket_order || []) ]); 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; };