const express = require('express'); const { generateSQL } = require('../lib/sql_generator'); const { sessionUser } = require('../lib/auth'); module.exports = function(pool) { const router = express.Router(); // list all registered sources router.get('/sources', async (req, res) => { try { const result = await pool.query( `SELECT * FROM pf.source ORDER BY schema, tname` ); res.json(result.rows); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // register a source table // auto-populates col_meta from information_schema with role='ignore' router.post('/sources', async (req, res) => { const { schema, tname, label } = req.body; const created_by = sessionUser(req); if (!schema || !tname) { return res.status(400).json({ error: 'schema and tname are required' }); } if (!/^\w+$/.test(schema) || !/^\w+$/.test(tname)) { return res.status(400).json({ error: 'Invalid schema or table name' }); } const client = await pool.connect(); try { await client.query('BEGIN'); const src = await client.query( `INSERT INTO pf.source (schema, tname, label, created_by) VALUES ($1, $2, $3, $4) RETURNING *`, [schema, tname, label || null, created_by || null] ); const source = src.rows[0]; // seed col_meta from information_schema await client.query(` INSERT INTO pf.col_meta (source_id, cname, role, opos) SELECT $1, column_name, 'dimension', ordinal_position FROM information_schema.columns WHERE table_schema = $2 AND table_name = $3 ORDER BY ordinal_position ON CONFLICT (source_id, cname) DO NOTHING `, [source.id, schema, tname]); await client.query('COMMIT'); res.status(201).json(source); } catch (err) { await client.query('ROLLBACK'); console.error(err); if (err.code === '23505') { return res.status(409).json({ error: 'Source already registered' }); } res.status(500).json({ error: err.message }); } finally { client.release(); } }); // get col_meta for a source router.get('/sources/:id/cols', async (req, res) => { try { const result = await pool.query( `SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [req.params.id] ); res.json(result.rows); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // save col_meta — accepts full array, upserts each row router.put('/sources/:id/cols', async (req, res) => { const sourceId = parseInt(req.params.id); const cols = req.body; if (!Array.isArray(cols)) { return res.status(400).json({ error: 'body must be an array' }); } const client = await pool.connect(); try { await client.query('BEGIN'); for (const col of cols) { await client.query(` INSERT INTO pf.col_meta (source_id, cname, label, role, is_key, dim_group, dim_period_col, in_grain, opos) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (source_id, cname) DO UPDATE SET label = EXCLUDED.label, role = EXCLUDED.role, is_key = EXCLUDED.is_key, dim_group = EXCLUDED.dim_group, dim_period_col = EXCLUDED.dim_period_col, in_grain = EXCLUDED.in_grain, opos = EXCLUDED.opos `, [ sourceId, col.cname, col.label || null, col.role || 'ignore', col.is_key || false, col.dim_group || null, col.dim_period_col || null, col.in_grain || false, col.opos || null ]); } await client.query('COMMIT'); const result = await pool.query( `SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [sourceId] ); res.json(result.rows); } catch (err) { await client.query('ROLLBACK'); console.error(err); res.status(500).json({ error: err.message }); } finally { client.release(); } }); // generate SQL for all operations from current col_meta and store in pf.sql router.post('/sources/:id/generate-sql', async (req, res) => { const sourceId = parseInt(req.params.id); try { const srcResult = await pool.query( `SELECT * FROM pf.source WHERE id = $1`, [sourceId] ); if (srcResult.rows.length === 0) { return res.status(404).json({ error: 'Source not found' }); } const colResult = await pool.query( `SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [sourceId] ); // validate required roles const colMeta = colResult.rows; const roles = new Set(colMeta.map(c => c.role)); const missing = ['value', 'date'].filter(r => !roles.has(r)); if (missing.length > 0) { return res.status(400).json({ error: `col_meta is missing required roles: ${missing.join(', ')}` }); } if (!colMeta.some(c => c.role === 'dimension')) { return res.status(400).json({ error: 'col_meta has no dimension columns' }); } const sqls = generateSQL(srcResult.rows[0], colMeta); const client = await pool.connect(); try { await client.query('BEGIN'); for (const [operation, sql] of Object.entries(sqls)) { await client.query(` INSERT INTO pf.sql (source_id, operation, sql, generated_at) VALUES ($1, $2, $3, now()) ON CONFLICT (source_id, operation) DO UPDATE SET sql = EXCLUDED.sql, generated_at = EXCLUDED.generated_at `, [sourceId, operation, sql]); } // drop operations this generation no longer produces — e.g. get_agg // after the grain has been cleared, which would otherwise leave a // stale template the load path would still pick up await client.query( `DELETE FROM pf.sql WHERE source_id = $1 AND operation <> ALL($2::text[])`, [sourceId, Object.keys(sqls)] ); await client.query('COMMIT'); } catch (err) { await client.query('ROLLBACK'); throw err; } finally { client.release(); } res.json({ message: 'SQL generated', operations: Object.keys(sqls) }); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // view generated SQL for a source (for inspection / debug) router.get('/sources/:id/sql', async (req, res) => { try { const result = await pool.query( `SELECT operation, sql, generated_at FROM pf.sql WHERE source_id = $1 ORDER BY operation`, [req.params.id] ); res.json(result.rows); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // get distinct values for a key column (used to populate operation panel dropdowns) router.get('/sources/:id/values/:col', async (req, res) => { const col = req.params.col; try { const srcResult = await pool.query( `SELECT schema, tname FROM pf.source WHERE id = $1`, [req.params.id] ); if (srcResult.rows.length === 0) return res.status(404).json({ error: 'Source not found' }); // validate col is a key dimension on this source const metaResult = await pool.query( `SELECT 1 FROM pf.col_meta WHERE source_id = $1 AND cname = $2 AND is_key = true`, [req.params.id, col] ); if (metaResult.rows.length === 0) { return res.status(400).json({ error: `"${col}" is not a key column` }); } // ?q= narrows, ?limit= caps. A key column can be very wide -- part on // osm_skinny has 11,290 distinct values -- so returning the lot to fill a // completion list is both a slow query and a large response for a control // that can only usefully show a handful. const { schema, tname } = srcResult.rows[0]; const q = (req.query.q || '').trim(); const limit = Math.min(parseInt(req.query.limit) || 5000, 5000); const params = []; let filter = `WHERE "${col}" IS NOT NULL`; if (q) { params.push(`%${q}%`); filter += ` AND "${col}"::text ILIKE $${params.length}`; } params.push(limit); const result = await pool.query( `SELECT DISTINCT "${col}"::text AS val FROM ${schema}.${tname} ${filter} ORDER BY 1 LIMIT $${params.length}`, params ); res.json(result.rows.map(r => r.val)); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // Resolve a dim_group to its key column and siblings, or explain why it cannot be. async function resolveGroup(sourceId, group) { const { rows: meta } = await pool.query( `SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [sourceId]); const members = meta.filter(c => c.dim_group === group); if (!members.length) { const err = new Error(`No columns are grouped as "${group}" on this source`); err.status = 404; throw err; } const keyCol = members.find(c => c.is_key); if (!keyCol) { const err = new Error( `Group "${group}" has no is_key column, so its members have nothing to be keyed on`); err.status = 400; throw err; } return { keyCol, siblings: members.filter(c => c.cname !== keyCol.cname), // recency column: the source's primary date, the same one the generator // treats as the date for loads dateCol: meta.find(c => c.role === 'date')?.cname || null, }; } // The member list for a group, as one array. Small enough to send whole -- // 11,290 parts on osm_skinny -- so the client holds it and filters locally // instead of querying per keystroke. router.get('/sources/:id/dim/:group', async (req, res) => { try { const sourceId = parseInt(req.params.id); const { keyCol, siblings } = await resolveGroup(sourceId, req.params.group); const includeInactive = req.query.all === '1'; const { rows } = await pool.query(` SELECT key_value, attrs, is_active, source_seen FROM pf.dim_member WHERE source_id = $1 AND dim_group = $2 ${includeInactive ? '' : 'AND is_active'} ORDER BY key_value `, [sourceId, req.params.group]); res.json({ group: req.params.group, key_col: keyCol.cname, siblings: siblings.map(c => c.cname), members: rows, }); } catch (err) { console.error(err); res.status(err.status || 500).json({ error: err.message }); } }); // Rebuild a group's members from the source. A merge, not a replace: curation // (a member deactivated by hand, or added before it ever sold) has to survive a // refresh, so absent members are marked source_seen = false rather than deleted. // // Slow by nature -- it reads the whole source, which for a view over a // transaction table is millions of rows -- so it is a deliberate action rather // than something that happens on a page load. router.post('/sources/:id/dim/:group/refresh', async (req, res) => { const sourceId = parseInt(req.params.id); const group = req.params.group; try { const srcResult = await pool.query( `SELECT schema, tname FROM pf.source WHERE id = $1`, [sourceId]); if (!srcResult.rows.length) return res.status(404).json({ error: 'Source not found' }); const { schema, tname } = srcResult.rows[0]; const { keyCol, siblings, dateCol } = await resolveGroup(sourceId, group); if (!siblings.length) { return res.status(400).json({ error: `Group "${group}" has no sibling columns to store` }); } const q = (n) => `"${n}"`; const attrs = siblings.map(c => `'${c.cname}', s.${q(c.cname)}::text`).join(', '); // A key can carry more than one attribute set across history -- 11,290 // parts against 13,662 combinations on osm_skinny. Take the most recent // by the source's date column, which is the live definition. const recency = dateCol ? `s.${q(dateCol)} DESC NULLS LAST` : `1`; const started = Date.now(); // An explicit stamp rather than now(): inside a transaction now() is the // transaction's start time, so "refreshed in this run" and "refreshed in // a run that began at the same instant" would be indistinguishable. const runAt = new Date(); const client = await pool.connect(); try { await client.query('BEGIN'); const { rows: [{ n }] } = await client.query(` WITH ranked AS ( SELECT s.${q(keyCol.cname)}::text AS key_value, jsonb_build_object(${attrs}) AS attrs, row_number() OVER ( PARTITION BY s.${q(keyCol.cname)} ORDER BY ${recency} ) AS rn FROM ${q(schema)}.${q(tname)} s WHERE s.${q(keyCol.cname)} IS NOT NULL ) ,upserted AS ( INSERT INTO pf.dim_member (source_id, dim_group, key_value, attrs, refreshed_at, source_seen) SELECT $1, $2, key_value, attrs, $3, true FROM ranked WHERE rn = 1 ON CONFLICT (source_id, dim_group, key_value) DO UPDATE SET attrs = EXCLUDED.attrs, source_seen = true, refreshed_at = $3, updated_at = now() RETURNING 1 ) SELECT count(*)::int AS n FROM upserted `, [sourceId, group, runAt]); // anything this run did not touch is no longer in the source const { rowCount: dropped } = await client.query(` UPDATE pf.dim_member SET source_seen = false, updated_at = now() WHERE source_id = $1 AND dim_group = $2 AND source_seen AND refreshed_at IS DISTINCT FROM $3 `, [sourceId, group, runAt]); await client.query('COMMIT'); res.json({ group, members: n, no_longer_in_source: dropped, ms: Date.now() - started }); } 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 }); } }); // given a key column value, look up sibling dim_group column values from source // returns { sibling_col: value, ... } if exactly one match, null if none or ambiguous router.get('/sources/:id/lookup', async (req, res) => { const { col, value } = req.query; if (!col || value == null || value === '') return res.json(null); try { const [srcResult, metaResult] = await Promise.all([ pool.query(`SELECT schema, tname FROM pf.source WHERE id = $1`, [req.params.id]), pool.query(`SELECT * FROM pf.col_meta WHERE source_id = $1 ORDER BY opos`, [req.params.id]) ]); if (srcResult.rows.length === 0) return res.status(404).json({ error: 'Source not found' }); const keyCol = metaResult.rows.find(c => c.cname === col && c.is_key && c.dim_group); if (!keyCol) return res.json(null); const siblings = metaResult.rows.filter(c => c.dim_group === keyCol.dim_group && c.cname !== col ); if (!siblings.length) return res.json(null); const { schema, tname } = srcResult.rows[0]; const sibCols = siblings.map(c => `"${c.cname}"`).join(', '); const result = await pool.query( `SELECT DISTINCT ${sibCols} FROM "${schema}"."${tname}" WHERE "${col}" = $1 LIMIT 2`, [value] ); if (result.rows.length !== 1) return res.json(null); res.json(result.rows[0]); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // set or clear the default Perspective layout for a source. // Body: a Perspective view config (group_by, split_by, columns, plugin_config, …). // Pass null or {} to clear. router.put('/sources/:id/default-layout', async (req, res) => { try { const layout = req.body && Object.keys(req.body).length > 0 ? req.body : null; const result = await pool.query( `UPDATE pf.source SET default_layout = $1 WHERE id = $2 RETURNING *`, [layout, req.params.id] ); if (result.rows.length === 0) return res.status(404).json({ error: 'Source not found' }); res.json(result.rows[0]); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); // deregister a source — does not drop existing forecast tables router.get('/dim-period/cols', async (req, res) => { try { const result = await pool.query(` SELECT column_name FROM information_schema.columns WHERE table_schema = 'pf' AND table_name = 'dim_period' AND column_name NOT IN ('sdat', 'edat', 'drange', 'ndays') ORDER BY ordinal_position `); res.json(result.rows.map(r => r.column_name)); } catch (err) { console.error(err); res.status(500).json({ error: err.message }); } }); router.delete('/sources/:id', async (req, res) => { try { const result = await pool.query( `DELETE FROM pf.source WHERE id = $1 RETURNING *`, [req.params.id] ); if (result.rows.length === 0) { return res.status(404).json({ error: 'Source not found' }); } res.json({ message: 'Source deregistered', source: result.rows[0] }); } catch (err) { console.error(err); if (err.code === '23001' || err.code === '23503') { return res.status(409).json({ error: 'Source has existing versions — delete them first.' }); } res.status(500).json({ error: err.message }); } }); return router; };