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