dataflow/api/routes/records.js

/**
 * Records Routes
 * Query and manage imported records
 */

const express = require('express');
const { lit } = require('../lib/sql');

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

    // List records for a source
    router.get('/source/:source_name', async (req, res, next) => {
        try {
            const { limit = 100, offset = 0, transformed_only } = req.query;
            const result = await pool.query(
                `SELECT * FROM list_records(${lit(req.params.source_name)}, ${lit(parseInt(limit))}, ${lit(parseInt(offset))}, ${lit(transformed_only === 'true')})`
            );
            res.json(result.rows);
        } catch (err) {
            next(err);
        }
    });

    // Get single record
    router.get('/:id', async (req, res, next) => {
        try {
            const result = await pool.query(`SELECT * FROM get_record(${lit(parseInt(req.params.id))})`);
            if (result.rows.length === 0) return res.status(404).json({ error: 'Record not found' });
            res.json(result.rows[0]);
        } catch (err) {
            next(err);
        }
    });

    // Search records
    router.post('/search', async (req, res, next) => {
        try {
            const { source_name, query, limit = 100 } = req.body;
            if (!source_name || !query) {
                return res.status(400).json({ error: 'Missing required fields: source_name, query' });
            }
            const result = await pool.query(
                `SELECT * FROM search_records(${lit(source_name)}, ${lit(query)}, ${lit(parseInt(limit))})`
            );
            res.json(result.rows);
        } catch (err) {
            next(err);
        }
    });

    // Set overrides for all selected records and immediately merge into transformed
    router.post('/bulk-overrides', async (req, res, next) => {
        try {
            const { source_name, record_ids, overrides } = req.body;
            if (!source_name || !record_ids || !Array.isArray(record_ids) || record_ids.length === 0 || !overrides || typeof overrides !== 'object') {
                return res.status(400).json({ error: 'source_name, record_ids array, and overrides object required' });
            }
            const client = await pool.connect();
            try {
                await client.query('BEGIN');
                for (const id of record_ids) {
                    await client.query(
                        `UPDATE dataflow.records 
                         SET overrides = COALESCE(overrides, '{}'::jsonb) || $1,
                            transformed = transformed || $1
                         WHERE id = $2 AND source_name = $3`,
                        [overrides, id, source_name]
                    );
                }
                await client.query('COMMIT');
                res.json({ updated: record_ids.length });
            } catch (err) {
                await client.query('ROLLBACK');
                next(err);
            } finally {
                client.release();
            }
        } catch (err) {
            next(err);
        }
    });

    // Set overrides for a record and immediately merge into transformed
    router.put('/:id/overrides', async (req, res, next) => {
        try {
            const { overrides } = req.body;
            if (!overrides || typeof overrides !== 'object')
                return res.status(400).json({ error: 'overrides object required' });
            const result = await pool.query(
                `SELECT * FROM set_record_overrides(${lit(parseInt(req.params.id))}, ${lit(overrides)})`
            );
            if (result.rows.length === 0) return res.status(404).json({ error: 'Record not found' });
            res.json(result.rows[0]);
        } catch (err) {
            next(err);
        }
    });

    // Clear overrides and reprocess that record to restore computed values
    router.delete('/:id/overrides', async (req, res, next) => {
        try {
            const rec = await pool.query(
                `SELECT * FROM clear_record_overrides(${lit(parseInt(req.params.id))})`
            );
            if (rec.rows.length === 0) return res.status(404).json({ error: 'Record not found' });
            // Reprocess this record so transformed reflects rules/mappings without overrides
            await pool.query(
                `SELECT apply_transformations(${lit(rec.rows[0].source_name)}, ARRAY[${lit(parseInt(req.params.id))}::int], true)`
            );
            const updated = await pool.query(`SELECT * FROM get_record(${lit(parseInt(req.params.id))})`);
            res.json(updated.rows[0]);
        } catch (err) {
            next(err);
        }
    });

    // Delete record
    router.delete('/:id', async (req, res, next) => {
        try {
            const result = await pool.query(`SELECT * FROM delete_record(${lit(parseInt(req.params.id))})`);
            if (result.rows.length === 0) return res.status(404).json({ error: 'Record not found' });
            res.json({ success: true, deleted: result.rows[0].id });
        } catch (err) {
            next(err);
        }
    });

    // Delete all records for a source
    router.delete('/source/:source_name/all', async (req, res, next) => {
        try {
            const result = await pool.query(`SELECT * FROM delete_source_records(${lit(req.params.source_name)})`);
            res.json({ success: true, deleted_count: result.rows[0].deleted_count });
        } catch (err) {
            next(err);
        }
    });

    return router;
};