Paul Trowbridge
c4e7211e6d
Sources with a `teller` block in their config can pull transactions straight from the bank instead of taking a CSV upload. Only the fetch differs — dedupe, logging, and transformation reuse the import path. api/lib/teller.js speaks Teller's mutual-TLS protocol (client cert plus the access token as the HTTP Basic username) and flattens transactions into the shallow map the rule engine expects. Access tokens live in .env, one per enrollment, not in the database that manage.py offers to reset. Pending transactions are skipped by default: their ids change when they post, which would import the same charge twice under two keys. Sources should use ['id'] as constraint_fields — Teller's transaction id makes overlapping pulls free while keeping genuinely repeated charges distinct. Untested against the live API — Teller has no self-serve signup at the moment, so no account to verify against. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01G2HFeU5neCKagTnmA6o9Tu
338 lines
13 KiB
JavaScript
338 lines
13 KiB
JavaScript
/**
|
|
* Sources Routes
|
|
* Manage data sources
|
|
*/
|
|
|
|
const express = require('express');
|
|
const multer = require('multer');
|
|
const { parse } = require('csv-parse/sync');
|
|
const { lit, arr } = require('../lib/sql');
|
|
const teller = require('../lib/teller');
|
|
|
|
const upload = multer({ storage: multer.memoryStorage() });
|
|
|
|
module.exports = (pool) => {
|
|
const router = express.Router();
|
|
|
|
// Global import log (all sources)
|
|
router.get('/import-log', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM get_all_import_logs()`);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// List the accounts behind a Teller enrollment — used to find the account_id
|
|
// to put in a source's config. Declared before /:name so it isn't shadowed.
|
|
router.get('/teller-accounts', async (req, res, next) => {
|
|
try {
|
|
const accounts = await teller.listAccounts(req.query.token_env);
|
|
res.json(accounts);
|
|
} catch (err) {
|
|
if (err instanceof teller.TellerError) return res.status(err.status || 502).json({ error: err.message });
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// List all sources
|
|
router.get('/', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM list_sources()`);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Get single source
|
|
router.get('/:name', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM get_source(${lit(req.params.name)})`);
|
|
if (result.rows.length === 0) return res.status(404).json({ error: 'Source not found' });
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Suggest source definition from CSV
|
|
router.post('/suggest', upload.single('file'), async (req, res, next) => {
|
|
try {
|
|
if (!req.file) return res.status(400).json({ error: 'No file uploaded' });
|
|
|
|
const records = parse(req.file.buffer, { columns: true, skip_empty_lines: true, trim: true });
|
|
if (records.length === 0) return res.status(400).json({ error: 'CSV file is empty' });
|
|
|
|
const ISO_DATE_RE = /^\d{4}-\d{2}-\d{2}(T[\d:.Z+-]+)?$/;
|
|
const sample = records[0];
|
|
const sampleRows = records.slice(0, 50);
|
|
|
|
const fields = Object.keys(sample).map(key => {
|
|
const vals = sampleRows.map(r => r[key]).filter(v => v !== '' && v != null);
|
|
let type = 'text';
|
|
if (vals.length > 0 && vals.every(v => !isNaN(parseFloat(v)) && isFinite(v) && String(v).charAt(0) !== '0')) {
|
|
type = 'numeric';
|
|
} else if (vals.length > 0 && vals.every(v => ISO_DATE_RE.test(String(v)))) {
|
|
type = 'date';
|
|
}
|
|
return { name: key, type };
|
|
});
|
|
|
|
res.json({ name: '', constraint_fields: [], fields, sampleRows });
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Create source
|
|
router.post('/', async (req, res, next) => {
|
|
try {
|
|
const { name, constraint_fields, config, global_picklist } = req.body;
|
|
if (!name || !constraint_fields || !Array.isArray(constraint_fields)) {
|
|
return res.status(400).json({ error: 'Missing required fields: name, constraint_fields (array)' });
|
|
}
|
|
const result = await pool.query(
|
|
`SELECT * FROM create_source(${lit(name)}, ${arr(constraint_fields)}, ${lit(config || {})}, ${lit(global_picklist !== false)})`
|
|
);
|
|
res.status(201).json(result.rows[0]);
|
|
} catch (err) {
|
|
if (err.code === '23505') return res.status(409).json({ error: 'Source already exists' });
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Update source
|
|
router.put('/:name', async (req, res, next) => {
|
|
try {
|
|
const { constraint_fields, config, global_picklist } = req.body;
|
|
const gpVal = global_picklist !== undefined ? lit(global_picklist) : 'NULL';
|
|
const result = await pool.query(
|
|
`SELECT * FROM update_source(${lit(req.params.name)}, ${constraint_fields ? arr(constraint_fields) : 'NULL'}, ${config ? lit(config) : 'NULL'}, ${gpVal})`
|
|
);
|
|
if (result.rows.length === 0) return res.status(404).json({ error: 'Source not found' });
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Delete source
|
|
router.delete('/:name', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM delete_source(${lit(req.params.name)})`);
|
|
if (result.rows.length === 0) return res.status(404).json({ error: 'Source not found' });
|
|
res.json({ success: true, deleted: result.rows[0].delete_source });
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Import CSV data and apply transformations to new records
|
|
router.post('/:name/import', upload.single('file'), async (req, res, next) => {
|
|
try {
|
|
if (!req.file) return res.status(400).json({ error: 'No file uploaded' });
|
|
const records = parse(req.file.buffer, { columns: true, skip_empty_lines: true, trim: true });
|
|
const importResult = await pool.query(
|
|
`SELECT import_records(${lit(req.params.name)}, ${lit(records)}) as result`
|
|
);
|
|
const importData = importResult.rows[0].result;
|
|
|
|
if (!importData.success) return res.json(importData);
|
|
|
|
const transformResult = await pool.query(
|
|
`SELECT apply_transformations(${lit(req.params.name)}) as result`
|
|
);
|
|
const transformData = transformResult.rows[0].result;
|
|
|
|
res.json({ ...importData, transform: transformData });
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Pull transactions from Teller and import them, same as a CSV upload.
|
|
// Safe to re-run: overlapping transactions are skipped by constraint key.
|
|
router.post('/:name/sync', async (req, res, next) => {
|
|
try {
|
|
const sourceResult = await pool.query(`SELECT * FROM get_source(${lit(req.params.name)})`);
|
|
const source = sourceResult.rows[0];
|
|
if (!source || !source.name) return res.status(404).json({ error: 'Source not found' });
|
|
|
|
const cfg = (source.config || {}).teller;
|
|
if (!cfg || !cfg.account_id) {
|
|
return res.status(400).json({
|
|
error: `Source "${req.params.name}" has no teller.account_id in its config`
|
|
});
|
|
}
|
|
|
|
const opts = { ...req.query, ...req.body };
|
|
const { fetched, records } = await teller.fetchTransactions({
|
|
accountId: cfg.account_id,
|
|
tokenEnv: cfg.token_env,
|
|
count: opts.count !== undefined ? parseInt(opts.count) : cfg.count,
|
|
days: opts.days !== undefined ? parseInt(opts.days) : cfg.days,
|
|
includePending: opts.include_pending === true || opts.include_pending === 'true',
|
|
});
|
|
|
|
if (records.length === 0) {
|
|
return res.json({ success: true, fetched, imported: 0, duplicates: 0 });
|
|
}
|
|
|
|
const importResult = await pool.query(
|
|
`SELECT import_records(${lit(req.params.name)}, ${lit(records)}) as result`
|
|
);
|
|
const importData = importResult.rows[0].result;
|
|
|
|
if (!importData.success) return res.json({ ...importData, fetched });
|
|
|
|
const transformResult = await pool.query(
|
|
`SELECT apply_transformations(${lit(req.params.name)}) as result`
|
|
);
|
|
|
|
res.json({ ...importData, fetched, transform: transformResult.rows[0].result });
|
|
} catch (err) {
|
|
if (err instanceof teller.TellerError) return res.status(err.status || 502).json({ error: err.message });
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Get import log
|
|
router.get('/:name/import-log', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM get_import_log(${lit(req.params.name)})`);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Delete an import (removes all records from that batch and the log entry)
|
|
router.delete('/:name/import-log/:id', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(
|
|
`SELECT delete_import(${lit(parseInt(req.params.id))}) as result`
|
|
);
|
|
const data = result.rows[0].result;
|
|
if (!data.success) return res.status(404).json(data);
|
|
res.json(data);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Apply transformations
|
|
router.post('/:name/transform', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT apply_transformations(${lit(req.params.name)}) as result`);
|
|
res.json(result.rows[0].result);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Get all known field names for a source
|
|
router.get('/:name/fields', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM get_source_fields(${lit(req.params.name)})`);
|
|
res.json(result.rows);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Generate output view
|
|
router.post('/:name/view', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT generate_source_view(${lit(req.params.name)}) as result`);
|
|
const data = result.rows[0].result;
|
|
if (data && data.success) {
|
|
await pool.query(`UPDATE dataflow.sources SET view_generated_at = NOW() WHERE name = ${lit(req.params.name)}`);
|
|
}
|
|
res.json(data);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Reprocess all records
|
|
router.post('/:name/reprocess', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT reprocess_records(${lit(req.params.name)}) as result`);
|
|
res.json(result.rows[0].result);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Get statistics
|
|
router.get('/:name/stats', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM get_source_stats(${lit(req.params.name)})`);
|
|
res.json(result.rows[0]);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Get view data (paginated, sortable)
|
|
router.get('/:name/view-data', async (req, res, next) => {
|
|
try {
|
|
const { limit = 100, offset = 0, sort_col, sort_dir, filters } = req.query;
|
|
let parsedFilters = null;
|
|
if (filters) {
|
|
try { parsedFilters = JSON.parse(filters); } catch { /* ignore bad JSON */ }
|
|
}
|
|
const result = await pool.query(
|
|
`SELECT get_view_data(${lit(req.params.name)}, ${lit(parseInt(limit))}, ${lit(parseInt(offset))}, ${lit(sort_col || null)}, ${lit(sort_dir || 'asc')}, ${parsedFilters ? lit(parsedFilters) : 'NULL'}) as result`
|
|
);
|
|
res.json(result.rows[0].result);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// Override keys — distinct field names used in overrides across all records for this source
|
|
router.get('/:name/override-keys', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(
|
|
`SELECT DISTINCT jsonb_object_keys(overrides) AS key
|
|
FROM dataflow.records
|
|
WHERE source_name = ${lit(req.params.name)} AND overrides IS NOT NULL
|
|
ORDER BY key`
|
|
);
|
|
res.json(result.rows.map(r => r.key));
|
|
} catch (err) { next(err); }
|
|
});
|
|
|
|
// Pivot layouts
|
|
router.get('/:name/layouts', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM list_pivot_layouts(${lit(req.params.name)})`);
|
|
res.json(result.rows);
|
|
} catch (err) { next(err); }
|
|
});
|
|
|
|
router.post('/:name/layouts', async (req, res, next) => {
|
|
try {
|
|
const { layout_name, config } = req.body;
|
|
if (!layout_name || !config) return res.status(400).json({ error: 'layout_name and config required' });
|
|
const result = await pool.query(
|
|
`SELECT * FROM save_pivot_layout(${lit(req.params.name)}, ${lit(layout_name)}, ${lit(config)})`
|
|
);
|
|
res.json(result.rows[0]);
|
|
} catch (err) { next(err); }
|
|
});
|
|
|
|
router.delete('/:name/layouts/:id', async (req, res, next) => {
|
|
try {
|
|
const result = await pool.query(`SELECT * FROM delete_pivot_layout(${lit(parseInt(req.params.id))})`);
|
|
if (result.rows.length === 0) return res.status(404).json({ error: 'Layout not found' });
|
|
res.json({ success: true });
|
|
} catch (err) { next(err); }
|
|
});
|
|
|
|
return router;
|
|
};
|