pf_app/routes/perspective.js

// SPIKE (Option C): server-side DuckDB virtual-server backend for Perspective.
//
// The browser runs Perspective's VirtualServer + GenericSQLVirtualServerModel
// (WASM) which turns each pivot interaction into a SQL string. That SQL is
// POSTed here and run against a persistent in-process DuckDB that has the
// forecast table materialized from Postgres. Only the generated SQL and the
// aggregated (viewport-sized) Arrow result cross the wire — never all rows.
//
// This is a measurement spike, not production wiring: one global DuckDB
// connection, materialize-once per version (stale after writes until /init
// refresh). See pf_perspective_options.md option C.

const express = require('express');

let duckdb = null;
try { duckdb = require('duckdb'); } catch { /* optional dep */ }

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

    let dbPromise = null;            // resolves to { db, conn }
    const materialized = new Map();  // versionId -> { duckName, rows, ms }

    function getDb() {
        if (dbPromise) return dbPromise;
        dbPromise = new Promise((resolve, reject) => {
            const db = new duckdb.Database(':memory:');
            const conn = db.connect();
            const PG = `host=${process.env.DB_HOST} port=${process.env.DB_PORT} `
                + `dbname=${process.env.DB_NAME} user=${process.env.DB_USER} password=${process.env.DB_PASSWORD}`;
            conn.exec(
                `INSTALL postgres; LOAD postgres; `
                + `INSTALL arrow FROM community; LOAD arrow; `
                + `ATTACH '${PG}' AS pg (TYPE postgres, READ_ONLY);`,
                (err) => err ? reject(err) : resolve({ db, conn })
            );
        });
        return dbPromise;
    }

    const allP   = (conn, sql) => new Promise((res, rej) => conn.all(sql, (e, r) => e ? rej(e) : res(r)));
    const execP  = (conn, sql) => new Promise((res, rej) => conn.exec(sql, (e)    => e ? rej(e) : res()));
    const arrowP = (conn, sql) => new Promise((res, rej) => conn.arrowIPCAll(sql, (e, r) => e ? rej(e) : res(r)));

    // bigint-safe JSON for the small metadata queries (schema/size/min-max)
    const sendJson = (res, rows) =>
        res.type('json').send(JSON.stringify(rows, (k, v) => typeof v === 'bigint' ? Number(v) : v));

    async function materialize(conn, versionId, refresh) {
        if (materialized.has(versionId) && !refresh) return materialized.get(versionId);
        const v = await pool.query(
            `SELECT s.tname FROM pf.version v JOIN pf.source s ON s.id = v.source_id WHERE v.id = $1`,
            [versionId]
        );
        if (!v.rows.length) throw new Error(`version ${versionId} not found`);
        const tname    = v.rows[0].tname;
        const fcTable  = `fc_${tname}_${versionId}`;
        const duckName = `fc_${versionId}`;
        const t0 = Date.now();
        if (refresh) await execP(conn, `DROP TABLE IF EXISTS "${duckName}";`);
        await execP(conn, `CREATE TABLE IF NOT EXISTS "${duckName}" AS SELECT * FROM pg."pf"."${fcTable}";`);
        const cnt = await allP(conn, `SELECT count(*) AS c FROM "${duckName}";`);
        const info = { duckName, rows: Number(cnt[0].c), ms: Date.now() - t0 };
        materialized.set(versionId, info);
        return info;
    }

    // Materialize a version's forecast table into DuckDB and report the hosted id.
    router.post('/perspective/init', async (req, res) => {
        if (!duckdb) return res.status(501).json({ error: 'duckdb node binding not installed' });
        try {
            const versionId = parseInt(req.body.versionId);
            if (!versionId) return res.status(400).json({ error: 'versionId required' });
            const { conn } = await getDb();
            const info = await materialize(conn, versionId, !!req.body.refresh);
            res.json({ tableId: `memory.${info.duckName}`, rows: info.rows, materialize_ms: info.ms });
        } catch (err) {
            console.error('[perspective/init]', err.message);
            res.status(500).json({ error: err.message });
        }
    });

    // Execute one Perspective-generated SQL statement. arrow=true → Arrow IPC bytes.
    router.post('/perspective/sql', async (req, res) => {
        if (!duckdb) return res.status(501).json({ error: 'duckdb node binding not installed' });
        const { sql, arrow } = req.body || {};
        if (!sql) return res.status(400).json({ error: 'sql required' });
        try {
            const { conn } = await getDb();
            if (process.env.PSP_DEBUG_SQL) console.log(`[psp sql${arrow ? ' arrow' : ''}]`, String(sql).replace(/\s+/g, ' ').slice(0, 220));
            if (arrow) {
                const bufs = await arrowP(conn, sql);
                const out  = Buffer.concat(bufs, bufs.reduce((a, b) => a + b.length, 0));
                res.type('application/octet-stream').send(out);
            } else {
                sendJson(res, await allP(conn, sql));
            }
        } catch (err) {
            console.error('[perspective/sql]', err.message, '\n  SQL:', String(sql).slice(0, 300));
            res.status(500).json({ error: err.message });
        }
    });

    return router;
};