dataflow/database/import.sql

--
-- Import queries
-- CSV import and the import audit trail; SQL for the import/log endpoints in
-- api/routes/sources.js
--

SET search_path TO dataflow, public;

-- ── Import ────────────────────────────────────────────────────────────────────

-- Import records, skipping any whose constraint key already exists in the table.
--
-- Dedup is enforced here, in the CTE — there is no unique constraint on
-- constraint_key and ON CONFLICT must never be used. Within one batch every row
-- inserts even if two rows share a constraint key, because banks legitimately send
-- identical-looking transactions (same date, description, amount) on the same day.
-- The key exists only to stop a re-imported overlapping date range from
-- double-counting rows already in the table.
CREATE OR REPLACE FUNCTION import_records(
    p_source_name TEXT,
    p_data JSONB  -- Array of records
) RETURNS JSON AS $$
DECLARE
    v_constraint_fields TEXT[];
    v_inserted          INTEGER;
    v_duplicates        INTEGER;
    v_log_id            INTEGER;
BEGIN
    SELECT constraint_fields INTO v_constraint_fields
    FROM dataflow.sources
    WHERE name = p_source_name;

    IF v_constraint_fields IS NULL THEN
        RETURN json_build_object(
            'success', false,
            'error', 'Source not found: ' || p_source_name
        );
    END IF;

    WITH
    -- All incoming records with their constraint keys
    pending AS (
        SELECT
            rec.value      AS data,
            rec.ordinality AS seq,
            (SELECT jsonb_object_agg(f, rec.value->>f)
             FROM unnest(v_constraint_fields) AS f) AS constraint_key
        FROM jsonb_array_elements(p_data) WITH ORDINALITY AS rec
    ),
    -- Keys already in the database (excluded)
    existing AS (
        SELECT DISTINCT r.constraint_key
        FROM dataflow.records r
        INNER JOIN pending p ON p.constraint_key = r.constraint_key
        WHERE r.source_name = p_source_name
    ),
    -- Rows whose constraint key is not yet in the database
    new_records AS (
        SELECT p.data, p.constraint_key, p.seq
        FROM pending p
        WHERE NOT EXISTS (SELECT 1 FROM existing e WHERE e.constraint_key = p.constraint_key)
    ),
    -- Write the log entry
    log_entry AS (
        INSERT INTO dataflow.import_log (source_name, records_imported, records_duplicate, info)
        VALUES (
            p_source_name,
            (SELECT count(*) FROM new_records),
            (SELECT count(*) FROM pending) - (SELECT count(*) FROM new_records),
            jsonb_build_object(
                'total',          jsonb_array_length(p_data),
                'inserted_keys',  (SELECT jsonb_agg(constraint_key ORDER BY constraint_key) FROM new_records),
                'excluded_keys',  (SELECT jsonb_agg(constraint_key) FROM existing)
            )
        )
        RETURNING id, records_imported, records_duplicate
    ),
    -- Insert new records
    inserted AS (
        INSERT INTO dataflow.records (source_name, data, constraint_key, import_id)
        SELECT p_source_name, nr.data, nr.constraint_key, (SELECT id FROM log_entry)
        FROM new_records nr
        ORDER BY nr.seq
        RETURNING id
    )
    SELECT le.id, le.records_imported, le.records_duplicate
    INTO   v_log_id, v_inserted, v_duplicates
    FROM   log_entry le;

    RETURN json_build_object(
        'success',    true,
        'imported',   v_inserted,
        'duplicates', v_duplicates,
        'log_id',     v_log_id
    );
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION import_records IS 'Import records with automatic deduplication';

-- ── Audit trail ───────────────────────────────────────────────────────────────

CREATE OR REPLACE FUNCTION get_import_log(p_source_name TEXT)
RETURNS TABLE (
    id              INTEGER,
    source_name     TEXT,
    records_imported INTEGER,
    records_duplicate INTEGER,
    imported_at     TIMESTAMPTZ,
    info            JSONB
) AS $$
    SELECT id, source_name, records_imported, records_duplicate, imported_at, info
    FROM dataflow.import_log
    WHERE source_name = p_source_name
    ORDER BY imported_at DESC;
$$ LANGUAGE sql;

COMMENT ON FUNCTION get_import_log IS 'Return import history for a source, newest first, including inserted/excluded key lists';

CREATE OR REPLACE FUNCTION get_all_import_logs()
RETURNS TABLE (
    id               INTEGER,
    source_name      TEXT,
    records_imported INTEGER,
    records_duplicate INTEGER,
    imported_at      TIMESTAMPTZ,
    info             JSONB
) AS $$
    SELECT id, source_name, records_imported, records_duplicate, imported_at, info
    FROM dataflow.import_log
    ORDER BY imported_at DESC;
$$ LANGUAGE sql;

COMMENT ON FUNCTION get_all_import_logs IS 'Return import history across all sources, newest first';

-- Records are removed by the import_id FK's ON DELETE CASCADE
CREATE OR REPLACE FUNCTION delete_import(p_log_id INTEGER)
RETURNS JSON AS $$
DECLARE
    v_deleted INTEGER;
BEGIN
    IF NOT EXISTS (SELECT 1 FROM dataflow.import_log WHERE id = p_log_id) THEN
        RETURN json_build_object('success', false, 'error', 'Import log entry not found');
    END IF;

    SELECT count(*) INTO v_deleted FROM dataflow.records WHERE import_id = p_log_id;

    -- Cascade handles deleting records via FK ON DELETE CASCADE
    DELETE FROM dataflow.import_log WHERE id = p_log_id;

    RETURN json_build_object(
        'success',         true,
        'records_deleted', v_deleted,
        'log_id',          p_log_id
    );
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION delete_import IS 'Delete all records belonging to an import batch and remove the log entry';