pf_app/routes/log.js

const express = require('express');
const { grainOf } = require('../lib/sql_generator');
const { sessionUser } = require('../lib/auth');
const { fcTable } = require('../lib/utils');

module.exports = function(pool) {
    const router = express.Router();

    // list log entries for a version, newest first, with row counts and value/units totals
    router.get('/versions/:id/log', async (req, res) => {
        const versionId = parseInt(req.params.id);
        try {
            const verResult = await pool.query(
                `SELECT v.*, 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];
            const table = fcTable(tname, versionId);

            const colMeta = await pool.query(
                `SELECT cname, role FROM pf.col_meta WHERE source_id = $1 AND role IN ('value', 'units')`,
                [source_id]
            );
            const valueCol = colMeta.rows.find(c => c.role === 'value')?.cname;
            const unitsCol = colMeta.rows.find(c => c.role === 'units')?.cname;

            const aggCols = [
                `count(f.pf_id)::int AS row_count`,
                valueCol ? `sum(f."${valueCol}")::float8 AS value_total` : `NULL::float8 AS value_total`,
                unitsCol ? `sum(f."${unitsCol}")::float8 AS units_total` : `NULL::float8 AS units_total`
            ].join(', ');

            // The totals are stamped onto the entry when it is written, so the
            // normal read is a scan of a few dozen log rows rather than a join
            // against millions of forecast rows.
            //
            // ?recount=1 does it the old way. Stored totals are fixed at write
            // time and cannot drift on their own, but nothing stops someone
            // deleting forecast rows by hand, and a stored figure has no way to
            // notice. This is the way back -- and the backfill for entries
            // written before the columns existed.
            const recount = req.query.recount === '1' || req.query.recount === 'true';
            const stamped = !recount && (await pool.query(
                `SELECT count(*)::int AS n FROM pf.log
                 WHERE version_id = $1 AND row_count IS NULL`, [versionId]
            )).rows[0].n === 0;

            // ?kind=adjustments drops the baseline and reference entries. That is not
            // only about what gets listed: the aggregate below joins the whole
            // forecast table, and on a real version the load entries own almost
            // every row of it -- 2.5M against a few thousand for the adjustments.
            // Filtering in the WHERE keeps them out of the join rather than
            // totalling them and discarding the answer.
            const adjustmentsOnly = req.query.kind === 'adjustments';
            const opFilter = adjustmentsOnly
                ? `AND l.operation NOT IN ('baseline', 'reference')`
                : '';

            const result = stamped
                ? await pool.query(`
                    SELECT l.*,
                           $2::text AS value_col,
                           $3::text AS units_col
                    FROM pf.log l
                    WHERE l.version_id = $1
                    ${opFilter}
                    ORDER BY l.id DESC
                `, [versionId, valueCol || null, unitsCol || null])
                : await pool.query(`
                    SELECT l.*, ${aggCols},
                           $2::text AS value_col,
                           $3::text AS units_col
                    FROM pf.log l
                    LEFT JOIN ${table} f ON f.pf_logid = l.id
                    WHERE l.version_id = $1
                    ${opFilter}
                    GROUP BY l.id
                    ORDER BY l.id DESC
                `, [versionId, valueCol || null, unitsCol || null]);

            // A recount is also a repair: write back what it found, so the next
            // read is cheap again and the stored figure matches the rows.
            if (recount) {
                for (const r of result.rows) {
                    await pool.query(
                        `UPDATE pf.log SET row_count = $2, value_total = $3, units_total = $4,
                                           measure_cols = $5::jsonb
                         WHERE id = $1`,
                        [r.id, r.row_count, r.value_total, r.units_total,
                         JSON.stringify({ value: valueCol || null, units: unitsCol || null })]
                    );
                }
            }
            // The statement is kilobytes per entry and the list is opened to scan,
            // not to read SQL. It stays in the row for the debug endpoint below.
            res.json(result.rows.map(({ sql_text, ...r }) => ({
                ...r, has_sql: !!sql_text,
            })));
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // Everything about one entry, for when the rows look wrong: what was asked
    // for (params), what it ran against (env), and the statement that actually
    // executed with territory and scope resolved into it (sql_text).
    //
    // Both the intent and the SQL, because the translation between them is
    // exactly what is in doubt when a result is surprising.
    router.get('/log/:logid/debug', async (req, res) => {
        const logId = parseInt(req.params.logid);
        try {
            const { rows } = await pool.query(
                `SELECT l.*, v.name AS version_name, s.schema, s.tname
                 FROM pf.log l
                 JOIN pf.version v ON v.id = l.version_id
                 JOIN pf.source s  ON s.id = v.source_id
                 WHERE l.id = $1`, [logId]
            );
            if (!rows.length) return res.status(404).json({ error: 'Log entry not found' });
            res.json(rows[0]);
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // undo a log entry — delete all fc rows with this logid, then delete the log entry
    router.delete('/log/:logid', async (req, res) => {
        const logId = parseInt(req.params.logid);
        try {
            const logResult = await pool.query(`
                SELECT l.*, v.status, s.tname, v.id AS version_id, v.source_id
                FROM pf.log l
                JOIN pf.version v ON v.id = l.version_id
                JOIN pf.source s ON s.id = v.source_id
                WHERE l.id = $1
            `, [logId]);
            if (!logResult.rows.length) return res.status(404).json({ error: 'Log entry not found' });
            const log = logResult.rows[0];
            if (log.status === 'closed') return res.status(403).json({ error: 'Version is closed' });

            // Undo deletes rows wholesale by logid, so it cannot be territory
            // filtered the way a read or a write can -- half-undoing an entry
            // would leave the version in a state nothing describes. Ownership
            // instead: your own entries, or an admin's override. Territory alone
            // would not do it anyway, since two accounts can share one.
            if (!req.session?.user?.is_admin && log.pf_user !== sessionUser(req)) {
                return res.status(403).json({
                    error: `That entry was made by ${log.pf_user || 'someone else'} — only they or an administrator can undo it`
                });
            }
            const table = fcTable(log.tname, log.version_id);

            // In grain mode the client's table is indexed on pf_gkey, so undo has to
            // report the grain keys to remove rather than raw pf_ids. The keys are
            // distinct while rows_deleted still counts the raw rows removed.
            const colMeta = await pool.query(
                `SELECT cname, role, in_grain, opos FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`,
                [log.source_id]
            );
            const grain = grainOf(colMeta.rows);

            const client = await pool.connect();
            try {
                await client.query('BEGIN');
                const deleted = grain
                    ? await client.query(`
                        WITH
                        del AS (
                            DELETE FROM ${table}
                            WHERE pf_logid = $1
                            RETURNING ${grain.groupCols().join(', ')}
                        )
                        SELECT
                             count(*)::int                        AS rows_deleted
                            ,array_agg(DISTINCT ${grain.key()})   AS pf_gkeys
                        FROM del
                      `, [logId])
                    : await client.query(
                        `DELETE FROM ${table} WHERE pf_logid = $1 RETURNING pf_id`, [logId]
                      );
                await client.query('DELETE FROM pf.log WHERE id = $1', [logId]);
                await client.query('COMMIT');
                res.json(grain
                    ? {
                        rows_deleted: deleted.rows[0].rows_deleted,
                        pf_gkeys:     deleted.rows[0].pf_gkeys || []
                      }
                    : {
                        rows_deleted: deleted.rowCount,
                        pf_ids:       deleted.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 });
        }
    });

    // update the note and/or tag on a log entry. Both are annotations — they never
    // affect the forecast rows — so they stay editable after the fact, including on
    // a closed version, where relabelling history is still legitimate.
    router.patch('/log/:logid', async (req, res) => {
        const logId = parseInt(req.params.logid);
        const { note, tag, bucket, label } = req.body;
        if (note === undefined && tag === undefined
            && bucket === undefined && label === undefined) {
            return res.status(400).json({
                error: 'Nothing to update — send note, tag, bucket and/or label'
            });
        }
        // Every entry carries a label and a bucket (pf.log's log_named CHECK), so
        // they can be renamed but not cleared. Refused here for the message; the
        // constraint would refuse it anyway.
        for (const [field, value] of [['label', label], ['bucket', bucket]]) {
            if (value !== undefined && !String(value ?? '').trim()) {
                return res.status(400).json({ error: `The ${field} can be renamed but not cleared` });
            }
        }
        try {
            // Same rule as undo: your own entries, or an admin's. These are
            // annotations, but label and bucket name the pivot's columns for
            // everyone who opens the version, so an unguarded PATCH let any
            // account rename the company's segments -- including on loads whose
            // rows it cannot see.
            const owner = await pool.query(
                `SELECT pf_user FROM pf.log WHERE id = $1`, [logId]
            );
            if (!owner.rows.length) return res.status(404).json({ error: 'Log entry not found' });
            if (!req.session?.user?.is_admin && owner.rows[0].pf_user !== sessionUser(req)) {
                return res.status(403).json({
                    error: `That entry was made by ${owner.rows[0].pf_user || 'someone else'} — only they or an administrator can change it`
                });
            }

            // COALESCE on the flag, not the value: an explicit null or '' must be
            // able to clear a field, which COALESCE on the value alone would ignore
            const result = await pool.query(
                `UPDATE pf.log SET
                     note   = CASE WHEN $2::bool THEN $3::text ELSE note   END,
                     tag    = CASE WHEN $4::bool THEN $5::text ELSE tag    END,
                     bucket = CASE WHEN $6::bool THEN $7::text ELSE bucket END,
                     label  = CASE WHEN $8::bool THEN $9::text ELSE label  END
                 WHERE id = $1 RETURNING *`,
                [
                    logId,
                    note   !== undefined, note   === undefined ? null : (String(note).trim()   || null),
                    tag    !== undefined, tag    === undefined ? null : (String(tag).trim()    || null),
                    bucket !== undefined, bucket === undefined ? null : (String(bucket).trim() || null),
                    label  !== undefined, label  === undefined ? null : (String(label).trim()  || null),
                ]
            );
            if (!result.rows.length) return res.status(404).json({ error: 'Log entry not found' });
            res.json(result.rows[0]);
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    return router;
};