Paul Trowbridge
d2ba944d41
Named layouts lived in localStorage: invisible to anyone else, gone on the next machine, and nothing to publish from. The one server-side layout was pf.source.default_layout -- a single anonymous blob any account could overwrite for every account, which is a published layout with no owner. pf.layout replaces both. A layout is named, owned, and either private or published; published ones are listed by everyone on the forecast and writable only by their owner or an admin, the same rule pf.log already uses for its entries. Scope is the version, since that is the entry point, with version_id NULL for the source-wide default a new version inherits. Applying is never restricted -- Save is withheld on a layout that is not yours, Save as forks it -- because the guarantee wanted is that a published layout cannot be changed out from under people, not that it cannot be adapted. The toolbar's flat chip row becomes one Layout menu: Published and Mine, rename/publish/default/delete shown only where they would be allowed, and a dirty dot computed by comparing the live config against the one the pivot was applied from, since restore() fires the change event itself. Existing localStorage lists are lifted into pf.layout on first load. PUT /sources/:id/default-layout is removed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
497 lines
21 KiB
JavaScript
497 lines
21 KiB
JavaScript
const express = require('express');
|
|
const { generateSQL, buildTerritoryClause } = require('../lib/sql_generator');
|
|
const { RELATION_COLUMNS_SQL } = require('../lib/utils');
|
|
const { sessionTerritory } = require('../lib/auth');
|
|
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 the source's real columns
|
|
await client.query(`
|
|
INSERT INTO pf.col_meta (source_id, cname, role, opos)
|
|
SELECT $3, column_name, 'dimension', ordinal_position
|
|
FROM (${RELATION_COLUMNS_SQL}) c
|
|
ORDER BY ordinal_position
|
|
ON CONFLICT (source_id, cname) DO NOTHING
|
|
`, [schema, tname, source.id]);
|
|
|
|
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' });
|
|
}
|
|
// Exactly one per source: the scope is a single IN list against a single
|
|
// column, and two flagged would silently mean whichever one a .find()
|
|
// reached first -- the trap is_key already fell into (see CLAUDE.md).
|
|
const territoryCols = cols.filter(c => c.is_territory).map(c => c.cname);
|
|
if (territoryCols.length > 1) {
|
|
return res.status(400).json({
|
|
error: `Only one column can be the territory. Flagged: ${territoryCols.join(', ')}`
|
|
});
|
|
}
|
|
|
|
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, is_territory, opos)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
|
|
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,
|
|
is_territory = EXCLUDED.is_territory,
|
|
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.is_territory || 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`;
|
|
|
|
// Completion reads the *source* table, which no territory scope has
|
|
// touched -- so without this a scoped account could enumerate every
|
|
// customer, part and rep in the business from a dropdown, having
|
|
// been shown none of their rows.
|
|
const terrRow = (await pool.query(
|
|
`SELECT cname FROM pf.col_meta WHERE source_id = $1 AND is_territory LIMIT 1`,
|
|
[req.params.id]
|
|
)).rows[0];
|
|
const terrClause = buildTerritoryClause(sessionTerritory(req), terrRow?.cname || null);
|
|
if (terrClause) filter += ` AND ${terrClause}`;
|
|
|
|
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 });
|
|
}
|
|
});
|
|
|
|
// PUT /sources/:id/default-layout is gone. It wrote pf.source.default_layout,
|
|
// one anonymous blob per source that any account could overwrite for every
|
|
// other account -- a published layout with no owner. Its successor is
|
|
// pf.layout: named, owned, and writable only by its owner or an admin. The
|
|
// old column is left in place, already migrated into pf.layout by
|
|
// setup_sql/01_schema.sql, and read by nothing.
|
|
|
|
// 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;
|
|
};
|