pf_app/routes/versions.js

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, 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;
};