pf_app/routes/operations.js

const express = require('express');
const { tableFromArrays, tableToIPC } = require('apache-arrow');
const { applyTokens, buildWhere, buildWhereAny, buildExcludeClause, buildExcludePredicate, buildSetClause, esc } = require('../lib/sql_generator');
const { fcTable } = require('../lib/utils');

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

    async function runSQL(sql, client) {
        console.log('--- SQL ---\n', sql, '\n--- END SQL ---');
        return (client || pool).query(sql);
    }

    // accept either the legacy single `slice` object or the newer `slices` array,
    // and drop any empty entries so an empty selection can never widen to TRUE
    function normalizeSlices(body) {
        const raw = Array.isArray(body.slices) && body.slices.length ? body.slices : [body.slice];
        return raw.filter(s => s && typeof s === 'object' && Object.keys(s).length > 0);
    }

    // Stamp the tag onto the log entry the operation just created.
    // Done as a follow-up UPDATE rather than inside the generated SQL: those
    // templates live in pf.sql per source, so adding a {{tag}} token would strand
    // every source that has not re-run "Generate SQL".
    async function tagLog(client, rows, tag) {
        const clean = (tag || '').trim();
        if (!clean) return null;
        const ids = [...new Set(rows.map(r => r.pf_logid).filter(id => id != null))];
        if (ids.length === 0) return null;
        await client.query(`UPDATE pf.log SET tag = $1 WHERE id = ANY($2::bigint[])`, [clean, ids]);
        return clean;
    }

    // A slice is only meaningful if at least one of its keys is a filterable
    // column. buildWhere silently drops unknown keys, so {"typo": "x"} would
    // otherwise reduce to TRUE and apply the operation to the whole version.
    // Refuse rather than let a malformed selection rewrite every row.
    function assertSelective(slices, ctx) {
        const allowed = new Set(ctx.filterCols);
        slices.forEach((sl, i) => {
            const hits = Object.keys(sl).filter(k => allowed.has(k));
            if (hits.length === 0) {
                const err = new Error(
                    `Slice ${i + 1} does not name any filterable column ` +
                    `(${JSON.stringify(sl)}). Expected one of: ${ctx.filterCols.join(', ')}.`
                );
                err.status = 400;
                throw err;
            }
        });
    }

    // echo back what the caller asked for, for the audit log
    function pickIntent(body) {
        const keys = ['mode', 'target_basis', 'value_incr', 'units_incr', 'value_pct', 'units_pct', 'pct',
                      'target_value', 'target_units', 'target_price'];
        const out = {};
        for (const k of keys) if (body[k] !== undefined && body[k] !== null && body[k] !== '') out[k] = body[k];
        return out;
    }

    // Totals for a WHERE clause, split into the rows operations can change and the
    // rows they cannot. Excluded iters (typically 'reference') are still visible in
    // the pivot, so their contribution has to be reported rather than dropped —
    // otherwise a target set against what the grid shows lands somewhere else.
    async function sliceTotals(client, ctx, whereClause, excludeClause) {
        const pred = buildExcludePredicate(ctx.version.exclude_iters);
        const agg = (col, filter) => col ? `sum(${col}) FILTER (WHERE ${filter})` : 'NULL';
        const v = ctx.valueCol ? `"${ctx.valueCol}"` : null;
        const u = ctx.unitsCol ? `"${ctx.unitsCol}"` : null;

        const r = await client.query(`
            SELECT ${agg(v, pred)}              AS total_value,
                   ${agg(u, pred)}              AS total_units,
                   ${agg(v ? `abs(${v})` : null, pred)} AS abs_value,
                   ${agg(u ? `abs(${u})` : null, pred)} AS abs_units,
                   ${agg(v, `NOT (${pred})`)}   AS excl_value,
                   ${agg(u, `NOT (${pred})`)}   AS excl_units
            FROM ${ctx.table} WHERE ${whereClause}
        `);
        const n = (x) => parseFloat(r.rows[0][x]) || 0;
        return {
            value:     n('total_value'), units:     n('total_units'),
            absValue:  n('abs_value'),   absUnits:  n('abs_units'),
            exclValue: n('excl_value'),  exclUnits: n('excl_units'),
        };
    }

    // Resolve each measure independently into the increment the scale SQL expects.
    // Value and units each accept exactly one of: an absolute target, a change
    // amount, or a percentage — whichever the caller sent. They are resolved
    // separately so a target on one measure and a percentage on the other can be
    // submitted together. Everything is measured against the totals of *this*
    // WHERE clause, which is what makes apply_mode 'each' land per slice.
    async function resolveIncrs(client, ctx, whereClause, excludeClause, body) {
        const num = (v) => (v === undefined || v === null || v === '') ? null : parseFloat(v);

        const tValue = num(body.target_value);
        const tUnits = num(body.target_units);
        const tPrice = num(body.target_price);
        const vIncr  = num(body.value_incr);
        const uIncr  = num(body.units_incr);
        let   vPct   = num(body.value_pct);
        let   uPct   = num(body.units_pct);

        // legacy shape: a single `pct` flag meaning "the increments are percentages"
        if (body.pct) {
            if (vPct === null && vIncr !== null) vPct = vIncr;
            if (uPct === null && uIncr !== null) uPct = uIncr;
        }
        const legacyPct = !!body.pct;

        const anyInput = [tValue, tUnits, tPrice, vIncr, uIncr, vPct, uPct].some(v => v !== null);
        if (!anyInput) return { value: 0, units: 0 };

        const totals = await sliceTotals(client, ctx, whereClause, excludeClause);

        // What the number is measured against:
        //   'adjustable' — only the rows this operation can write (the default, and
        //                  what every earlier version of this API did)
        //   'selected'   — everything the pivot shows for the slice, excluded rows
        //                  included. Those rows cannot move, so reaching the target
        //                  means the adjustable rows absorb the whole difference.
        const basis = body.target_basis === 'selected' ? 'selected' : 'adjustable';
        const fixedValue = basis === 'selected' ? totals.exclValue : 0;
        const fixedUnits = basis === 'selected' ? totals.exclUnits : 0;

        // one measure: target wins, then percentage, then a plain change amount
        const resolve = (target, pct, incr, current, fixed) => {
            // subtract the immovable part: current + incr + fixed === target
            if (target !== null) return (target - fixed) - current;
            // a percentage of the basis, which may include the immovable part
            if (pct    !== null) return (current + fixed) * pct / 100;
            if (incr   !== null && !legacyPct) return incr;
            return 0;
        };

        let value = resolve(tValue, vPct, vIncr, totals.value, fixedValue);
        const units = resolve(tUnits, uPct, uIncr, totals.units, fixedUnits);

        // a price target holds units constant: new value = price x current units.
        // An explicit value target outranks it.
        if (tPrice !== null && tValue === null) {
            value = (tPrice * (totals.units + fixedUnits)) - (totals.value + fixedValue);
        }

        // the scale SQL divides by the slice total; with no rows there is
        // nothing to prorate across and the increment would vanish anyway
        if (totals.value === 0 && totals.units === 0) return { value: 0, units: 0 };

        // Refuse to prorate across a pool that nets to ~zero. Each row's new value is
        // (row / total) * increment, so as the net approaches zero the multiplier
        // explodes and rows fly apart in opposite directions to hit the target — a
        // mathematically faithful, practically useless result. Selecting slices that
        // offset each other is the usual cause, and 'each' handles that correctly.
        assertProratable(totals, value, units);

        return { value: round(value, 6), units: round(units, 6) };
    }

    // a pool is proratable only if its net is a meaningful fraction of its gross
    const NET_TO_GROSS_FLOOR = 0.01;

    function assertProratable(totals, value, units) {
        const check = (net, gross, incr, label) => {
            if (!incr) return;
            if (gross === 0) return;
            if (Math.abs(net) >= gross * NET_TO_GROSS_FLOOR) return;
            const err = new Error(
                `Cannot prorate ${label} across this selection: the rows net to ` +
                `${net.toFixed(2)} against a gross of ${gross.toFixed(2)}, so they very ` +
                `nearly cancel out. Scaling to a target would push them to extreme ` +
                `opposite values. Use "Each" to scale every slice on its own, or narrow ` +
                `the selection so it does not mix offsetting rows.`
            );
            err.status = 400;
            throw err;
        };
        check(totals.value, totals.absValue, value, 'value');
        check(totals.units, totals.absUnits, units, 'units');
    }

    function round(n, dp) {
        if (!isFinite(n)) return 0;
        const f = Math.pow(10, dp);
        return Math.round(n * f) / f;
    }

    // fetch everything needed to execute an operation:
    // version + source info, col_meta, fc_table name, stored SQL
    async function getContext(versionId, operation) {
        const verResult = await pool.query(`
            SELECT v.*, 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
        `, [versionId]);
        if (verResult.rows.length === 0) {
            const err = new Error('Version not found'); err.status = 404; throw err;
        }
        const version = verResult.rows[0];

        const colResult = await pool.query(
            `SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`,
            [version.source_id]
        );
        const colMeta  = colResult.rows;
        const dimCols  = colMeta.filter(c => c.role === 'dimension').map(c => c.cname);
        const dateCols = colMeta.filter(c => c.role === 'date').map(c => c.cname);
        const valueCol = colMeta.find(c => c.role === 'value')?.cname;
        const unitsCol = colMeta.find(c => c.role === 'units')?.cname;

        const sqlResult = await pool.query(
            `SELECT sql FROM pf.sql WHERE source_id = $1 AND operation = $2`,
            [version.source_id, operation]
        );
        if (sqlResult.rows.length === 0) {
            const err = new Error(`No generated SQL for operation "${operation}" — run generate-sql first`);
            err.status = 400; throw err;
        }

        return {
            version,
            table: fcTable(version.tname, version.id),
            colMeta,
            dimCols,
            dateCols,
            filterCols: [...dimCols, ...dateCols],
            valueCol,
            unitsCol,
            sql: sqlResult.rows[0].sql
        };
    }

    function guardOpen(version, res) {
        if (version.status === 'closed') {
            res.status(403).json({ error: 'Version is closed' });
            return false;
        }
        return true;
    }

    // stream all rows for a version as Arrow IPC (all iters including reference)
    router.get('/versions/:id/data', async (req, res) => {
        const versionId = parseInt(req.params.id);
        let client, committed = false;
        try {
            const verResult = await pool.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) {
                const err = new Error('Version not found'); err.status = 404; throw err;
            }
            const tbl = fcTable(verResult.rows[0].tname, versionId);

            const { rows: [{ count }] } = await pool.query(`SELECT COUNT(*) FROM ${tbl}`);
            const rowCount = parseInt(count);

            res.setHeader('Content-Type', 'application/vnd.apache.arrow.stream');
            res.setHeader('X-Row-Count', String(rowCount));

            if (rowCount === 0) { res.end(); return; }

            client = await pool.connect();
            await client.query('BEGIN');
            await client.query(`
                DECLARE pf_cur CURSOR FOR
                SELECT * FROM ${tbl}
            `);

            // Accumulate into column arrays (not row objects) to avoid allocating one JS
            // object per row — cuts peak heap by ~3-5× for large datasets.
            // Still emits a single Arrow record batch so Perspective WASM never sees
            // dictionary REPLACEMENT messages (which crash its Arrow reader).
            let colArrays = null;
            while (true) {
                const { rows } = await client.query('FETCH 10000 FROM pf_cur');
                if (!rows.length) break;
                if (!colArrays) {
                    colArrays = Object.fromEntries(Object.keys(rows[0]).map(k => [k, []]));
                }
                for (const row of rows) {
                    for (const k of Object.keys(colArrays)) colArrays[k].push(row[k]);
                }
            }
            await client.query('COMMIT');
            committed = true;

            const buf = tableToIPC(tableFromArrays(colArrays || {}), 'stream');
            res.setHeader('Content-Length', String(buf.byteLength));
            res.end(Buffer.from(buf.buffer, buf.byteOffset, buf.byteLength));
        } catch (err) {
            console.error(err);
            if (!res.headersSent) res.status(err.status || 500).json({ error: err.message });
            else res.destroy();
        } finally {
            if (client) {
                if (!committed) try { await client.query('ROLLBACK'); } catch {}
                client.release();
            }
        }
    });

    // load baseline rows from source table — additive, no delete
    router.post('/versions/:id/baseline', async (req, res) => {
        const { where_clause, date_offset, pf_user, note, filters, raw_where } = req.body;
        const dateOffset    = date_offset  || '0 days';
        const filterClause  = (raw_where || where_clause || '').trim() || 'TRUE';
        try {
            const ctx = await getContext(parseInt(req.params.id), 'baseline');
            if (!guardOpen(ctx.version, res)) return;
            const paramsJson = JSON.stringify({
                where_clause: filterClause,
                date_offset:  dateOffset,
                ...(raw_where ? { raw_where } : (filters ? { filters } : {}))
            });
            const sql = applyTokens(ctx.sql, {
                fc_table:      ctx.table,
                version_id:    ctx.version.id,
                pf_user:       esc(pf_user || ''),
                note:          esc(note    || ''),
                params:        esc(paramsJson),
                filter_clause: filterClause,
                date_offset:   esc(dateOffset)
            });

            const result = await runSQL(sql);
            res.json(result.rows[0]);
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // edit a baseline or reference segment in place — only allowed before any
    // scale/recode/clone has been applied on this version, since those would
    // have been calibrated against the old segment's totals.
    router.put('/versions/:id/baseline/:logid', async (req, res) => {
        const versionId = parseInt(req.params.id);
        const logid     = parseInt(req.params.logid);
        const { where_clause, date_offset, pf_user, note, filters, raw_where } = req.body;
        const dateOffset   = date_offset  || '0 days';
        const filterClause = (raw_where || where_clause || '').trim() || 'TRUE';

        const client = await pool.connect();
        try {
            const logResult = await client.query(
                `SELECT * FROM pf.log WHERE id = $1 AND version_id = $2`,
                [logid, versionId]
            );
            if (logResult.rows.length === 0) {
                return res.status(404).json({ error: 'Log entry not found' });
            }
            const oldLog = logResult.rows[0];
            if (!['baseline', 'reference'].includes(oldLog.operation)) {
                return res.status(400).json({ error: 'Only baseline or reference segments can be edited' });
            }

            const opsResult = await client.query(
                `SELECT COUNT(*)::int AS n FROM pf.log
                 WHERE version_id = $1 AND operation IN ('scale', 'recode', 'clone')`,
                [versionId]
            );
            if (opsResult.rows[0].n > 0) {
                return res.status(409).json({
                    error: 'Cannot edit segments after forecast operations have been applied. Undo the operations first.'
                });
            }

            const ctx = await getContext(versionId, oldLog.operation);
            if (!guardOpen(ctx.version, res)) return;

            const paramsJson = JSON.stringify({
                where_clause: filterClause,
                date_offset:  dateOffset,
                ...(raw_where ? { raw_where } : (filters ? { filters } : {}))
            });
            const sql = applyTokens(ctx.sql, {
                fc_table:      ctx.table,
                version_id:    ctx.version.id,
                pf_user:       esc(pf_user || ''),
                note:          esc(note    || ''),
                params:        esc(paramsJson),
                filter_clause: filterClause,
                date_offset:   esc(dateOffset)
            });

            await client.query('BEGIN');
            const delRows = await client.query(
                `DELETE FROM ${ctx.table} WHERE pf_logid = $1 RETURNING pf_id`,
                [logid]
            );
            await client.query(`DELETE FROM pf.log WHERE id = $1`, [logid]);
            const insResult = await client.query(sql);
            await client.query('COMMIT');

            res.json({
                rows_deleted: delRows.rowCount,
                pf_ids:       delRows.rows.map(r => r.pf_id),
                rows_affected: insResult.rows[0]?.rows_affected ?? 0
            });
        } catch (err) {
            try { await client.query('ROLLBACK'); } catch {}
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        } finally {
            client.release();
        }
    });

    // delete all baseline rows and log entries for a version
    router.delete('/versions/:id/baseline', async (req, res) => {
        const versionId = parseInt(req.params.id);
        try {
            const ctx = await getContext(versionId, 'baseline');
            if (!guardOpen(ctx.version, res)) return;
            const client = await pool.connect();
            try {
                await client.query('BEGIN');
                const delRows = await client.query(
                    `DELETE FROM ${ctx.table} WHERE pf_iter = 'baseline' RETURNING pf_id`
                );
                const delLog = await client.query(
                    `DELETE FROM pf.log WHERE version_id = $1 AND operation = 'baseline'`,
                    [versionId]
                );
                await client.query('COMMIT');
                res.json({
                    rows_deleted: delRows.rowCount,
                    log_entries_deleted: delLog.rowCount,
                    pf_ids: delRows.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 });
        }
    });

    // load reference rows from source table (additive — does not clear prior reference rows)
    router.post('/versions/:id/reference', async (req, res) => {
        const { where_clause, date_offset, pf_user, note, filters, raw_where } = req.body;
        const dateOffset   = date_offset || '0 days';
        const filterClause = (raw_where || where_clause || '').trim() || 'TRUE';
        try {
            const ctx = await getContext(parseInt(req.params.id), 'reference');
            if (!guardOpen(ctx.version, res)) return;
            const paramsJson = JSON.stringify({
                where_clause: filterClause,
                date_offset:  dateOffset,
                ...(raw_where ? { raw_where } : (filters ? { filters } : {}))
            });
            const sql = applyTokens(ctx.sql, {
                fc_table:      ctx.table,
                version_id:    ctx.version.id,
                pf_user:       esc(pf_user || ''),
                note:          esc(note    || ''),
                params:        esc(paramsJson),
                filter_clause: filterClause,
                date_offset:   esc(dateOffset)
            });

            const result = await runSQL(sql);
            res.json(result.rows[0]);
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // scale one or more slices — adjust value and/or units toward an absolute
    // target or by an increment. With several slices selected, apply_mode decides
    // whether they are treated as one pool ('prorate') or independently ('each').
    router.post('/versions/:id/scale', async (req, res) => {
        const { pf_user, note, apply_mode } = req.body;
        const slices = normalizeSlices(req.body);
        if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });

        const applyMode = apply_mode === 'each' ? 'each' : 'prorate';

        try {
            const ctx = await getContext(parseInt(req.params.id), 'scale');
            if (!guardOpen(ctx.version, res)) return;
            assertSelective(slices, ctx);

            const excludeClause = buildExcludeClause(ctx.version.exclude_iters);

            // 'prorate' pools every slice into one WHERE and lets the SQL's
            // sum() OVER () distribute the increment across the whole pool.
            // 'each' runs the same statement once per slice, so every slice
            // reaches the target on its own and gets its own log entry.
            const units = applyMode === 'each'
                ? slices.map(sl => ({ slices: [sl], where: buildWhere(sl, ctx.filterCols) }))
                : [{ slices, where: buildWhereAny(slices, ctx.filterCols) }];

            const client = await pool.connect();
            let committed = false;
            try {
                await client.query('BEGIN');
                const allRows = [];
                let applied = 0;
                const skipped = [];

                for (const unit of units) {
                    const incr = await resolveIncrs(client, ctx, unit.where, excludeClause, req.body);
                    // no rows, or already at the target — nothing to write for this slice
                    if (incr.value === 0 && incr.units === 0) { skipped.push(...unit.slices); continue; }
                    applied++;

                    const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
                    const sql = applyTokens(ctx.sql, {
                        fc_table:       ctx.table,
                        version_id:     ctx.version.id,
                        pf_user:        esc(pf_user || ''),
                        note:           esc(note    || ''),
                        params:         esc(JSON.stringify({
                            slices: unit.slices,
                            apply_mode: applyMode,
                            ...pickIntent(req.body),
                            resolved: { value_incr: incr.value, units_incr: incr.units }
                        })),
                        slice:          esc(JSON.stringify(loggedSlice)),
                        where_clause:   unit.where,
                        exclude_clause: excludeClause,
                        value_incr:     incr.value,
                        units_incr:     incr.units
                    });
                    const result = await runSQL(sql, client);
                    await tagLog(client, result.rows, req.body.tag);
                    allRows.push(...result.rows);
                }

                if (allRows.length === 0) {
                    await client.query('ROLLBACK');
                    return res.status(400).json({
                        error: 'Nothing to scale — the target matches the current total, or the increment is zero'
                    });
                }

                await client.query('COMMIT');
                committed = true;
                const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'scale' }));
                res.json({
                    rows,
                    rows_affected:  rows.length,
                    slices_applied: applied,
                    ...(skipped.length ? { slices_skipped: skipped } : {})
                });
            } finally {
                if (!committed) try { await client.query('ROLLBACK'); } catch {}
                client.release();
            }
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // recode dimension values on one or more slices
    // inserts negative rows to zero out the original, positive rows with new dimension values
    router.post('/versions/:id/recode', async (req, res) => {
        const { pf_user, note, set, apply_mode } = req.body;
        const slices = normalizeSlices(req.body);
        if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });
        if (!set || Object.keys(set).length === 0) return res.status(400).json({ error: 'set is required' });

        try {
            const ctx = await getContext(parseInt(req.params.id), 'recode');
            if (!guardOpen(ctx.version, res)) return;
            assertSelective(slices, ctx);

            const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
            const setClause     = buildSetClause(ctx.dimCols, set);
            const units         = sliceUnits(slices, ctx, apply_mode);

            const client = await pool.connect();
            let committed = false;
            try {
                await client.query('BEGIN');
                const allRows = [];
                for (const unit of units) {
                    const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
                    const sql = applyTokens(ctx.sql, {
                        fc_table:       ctx.table,
                        version_id:     ctx.version.id,
                        pf_user:        esc(pf_user || ''),
                        note:           esc(note    || ''),
                        params:         esc(JSON.stringify({ slices: unit.slices, set, apply_mode: unit.mode })),
                        slice:          esc(JSON.stringify(loggedSlice)),
                        where_clause:   unit.where,
                        exclude_clause: excludeClause,
                        set_clause:     setClause
                    });
                    const result = await runSQL(sql, client);
                    await tagLog(client, result.rows, req.body.tag);
                    allRows.push(...result.rows);
                }
                await client.query('COMMIT');
                committed = true;
                const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'recode' }));
                res.json({ rows, rows_affected: rows.length, slices_applied: units.length });
            } finally {
                if (!committed) try { await client.query('ROLLBACK'); } catch {}
                client.release();
            }
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // clone one or more slices as new business under new dimension values
    // does not offset the original slice
    router.post('/versions/:id/clone', async (req, res) => {
        const { pf_user, note, set, scale, apply_mode } = req.body;
        const slices = normalizeSlices(req.body);
        if (slices.length === 0) return res.status(400).json({ error: 'slice is required' });
        if (!set || Object.keys(set).length === 0) return res.status(400).json({ error: 'set is required' });

        try {
            const ctx = await getContext(parseInt(req.params.id), 'clone');
            if (!guardOpen(ctx.version, res)) return;
            assertSelective(slices, ctx);

            const scaleFactor   = (scale != null) ? parseFloat(scale) : 1.0;
            const excludeClause = buildExcludeClause(ctx.version.exclude_iters);
            const setClause     = buildSetClause(ctx.dimCols, set);
            const units         = sliceUnits(slices, ctx, apply_mode);

            const client = await pool.connect();
            let committed = false;
            try {
                await client.query('BEGIN');
                const allRows = [];
                for (const unit of units) {
                    const loggedSlice = unit.slices.length === 1 ? unit.slices[0] : unit.slices;
                    const sql = applyTokens(ctx.sql, {
                        fc_table:       ctx.table,
                        version_id:     ctx.version.id,
                        pf_user:        esc(pf_user || ''),
                        note:           esc(note    || ''),
                        params:         esc(JSON.stringify({ slices: unit.slices, set, scale: scaleFactor, apply_mode: unit.mode })),
                        slice:          esc(JSON.stringify(loggedSlice)),
                        where_clause:   unit.where,
                        exclude_clause: excludeClause,
                        set_clause:     setClause,
                        scale_factor:   scaleFactor
                    });
                    const result = await runSQL(sql, client);
                    await tagLog(client, result.rows, req.body.tag);
                    allRows.push(...result.rows);
                }
                await client.query('COMMIT');
                committed = true;
                const rows = allRows.map(r => ({ ...r, pf_note: note || null, pf_op: 'clone' }));
                res.json({ rows, rows_affected: rows.length, slices_applied: units.length });
            } finally {
                if (!committed) try { await client.query('ROLLBACK'); } catch {}
                client.release();
            }
        } catch (err) {
            console.error(err);
            res.status(err.status || 500).json({ error: err.message });
        }
    });

    // log routes (GET /versions/:id/log, DELETE /log/:logid, PATCH /log/:logid)
    // live in routes/log.js — see that file.

    return router;
};