dataflow/database/functions.sql

--
-- Dataflow Functions
-- Simple, clear functions for import and transformation
--

SET search_path TO dataflow, public;

------------------------------------------------------
-- Function: import_records
-- Import data with automatic deduplication
------------------------------------------------------
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';

------------------------------------------------------
-- Function: get_import_log
-- Return import history for a source
------------------------------------------------------
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';

------------------------------------------------------
-- Function: get_all_import_logs
-- Return import history across all sources
------------------------------------------------------
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';

------------------------------------------------------
-- Function: delete_import
-- Delete all records from a specific import and remove the log entry
------------------------------------------------------
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';

------------------------------------------------------
-- Aggregate: jsonb_concat_obj
-- Merge JSONB objects across rows (later rows win on key conflicts)
-- Usage: jsonb_concat_obj(col ORDER BY sequence)
------------------------------------------------------
CREATE OR REPLACE FUNCTION dataflow.jsonb_merge(a JSONB, b JSONB)
RETURNS JSONB AS $$
    SELECT COALESCE(a, '{}') || COALESCE(b, '{}')
$$ LANGUAGE sql IMMUTABLE;

DROP AGGREGATE IF EXISTS dataflow.jsonb_concat_obj(JSONB);
CREATE AGGREGATE dataflow.jsonb_concat_obj(JSONB) (
    sfunc    = dataflow.jsonb_merge,
    stype    = JSONB,
    initcond = '{}'
);

------------------------------------------------------
-- Function: apply_transformations
-- Apply all transformation rules to records (set-based)
------------------------------------------------------
DROP FUNCTION IF EXISTS apply_transformations(TEXT, INTEGER[]);
CREATE OR REPLACE FUNCTION apply_transformations(
    p_source_name TEXT,
    p_record_ids  INTEGER[] DEFAULT NULL,  -- NULL = all eligible records
    p_overwrite   BOOLEAN   DEFAULT FALSE  -- FALSE = skip already-transformed, TRUE = overwrite all
) RETURNS JSON AS $$
DECLARE
    v_seq   INT;
    v_count INT := 0;
BEGIN
    -- Accumulator: one row per qualifying record, additions built up across sequence steps.
    -- Each sequence step reads from data || additions so later rules can reference earlier outputs.
    CREATE TEMP TABLE _xform_acc ON COMMIT DROP AS
    SELECT id, data, '{}'::jsonb AS additions
    FROM dataflow.records
    WHERE source_name = p_source_name
      AND (p_overwrite OR transformed IS NULL)
      AND (p_record_ids IS NULL OR id = ANY(p_record_ids));

    -- Process one sequence group at a time, in order.
    -- Rules at sequence N can read fields written by rules at sequence < N.
    FOR v_seq IN
        SELECT DISTINCT sequence
        FROM dataflow.rules
        WHERE source_name = p_source_name AND enabled = true
        ORDER BY sequence
    LOOP
        WITH
        -- Current view of each record: original data merged with accumulated outputs so far
        current AS (
            SELECT id, data || additions AS current_data
            FROM _xform_acc
        ),
        -- Fan out one row per regex match for rules at this sequence level
        rx AS (
            SELECT
                c.id,
                r.name          AS rule_name,
                r.sequence,
                r.output_field,
                r.retain,
                r.function_type,
                COALESCE(mt.rn, rp.rn, 1) AS result_number,
                CASE WHEN array_length(mt.mt, 1) = 1 THEN to_jsonb(mt.mt[1]) ELSE to_jsonb(mt.mt) END AS match_val,
                to_jsonb(rp.rp) AS replace_val
            FROM dataflow.rules r
            INNER JOIN current c ON (c.current_data ? r.field)
            LEFT JOIN LATERAL regexp_matches(c.current_data ->> r.field, r.pattern, r.flags)
                WITH ORDINALITY AS mt(mt, rn) ON r.function_type = 'extract'
            LEFT JOIN LATERAL regexp_replace(c.current_data ->> r.field, r.pattern, r.replace_value, r.flags)
                WITH ORDINALITY AS rp(rp, rn) ON r.function_type = 'replace'
            WHERE r.source_name = p_source_name
              AND r.sequence = v_seq
              AND r.enabled = true
        ),
        agg_matches AS (
            SELECT
                id, rule_name, sequence, output_field, retain, function_type,
                CASE function_type
                    WHEN 'replace' THEN jsonb_agg(replace_val) -> 0
                    ELSE
                        CASE WHEN max(result_number) = 1
                            THEN jsonb_agg(match_val ORDER BY result_number) -> 0
                            ELSE jsonb_agg(match_val ORDER BY result_number)
                        END
                END AS extracted
            FROM rx
            GROUP BY id, rule_name, sequence, output_field, retain, function_type
        ),
        linked AS (
            SELECT
                a.id, a.sequence, a.output_field, a.retain, a.extracted,
                m.output AS mapped
            FROM agg_matches a
            LEFT JOIN dataflow.mappings m ON
                m.source_name = p_source_name
                AND m.rule_name = a.rule_name
                AND m.input_value = a.extracted
            WHERE a.extracted IS NOT NULL
        ),
        rule_output AS (
            SELECT id, sequence,
                CASE
                    WHEN mapped IS NOT NULL THEN
                        mapped || CASE WHEN retain THEN jsonb_build_object(output_field, extracted) ELSE '{}'::jsonb END
                    ELSE
                        jsonb_build_object(output_field, extracted)
                END AS output
            FROM linked
        ),
        seq_additions AS (
            SELECT id, dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
            FROM rule_output
            GROUP BY id
        )
        UPDATE _xform_acc acc
        SET additions = additions || COALESCE(sa.additions, '{}'::jsonb)
        FROM seq_additions sa
        WHERE acc.id = sa.id;
    END LOOP;

    -- Write final result: original data + all accumulated rule outputs + any manual overrides
    WITH updated AS (
        UPDATE dataflow.records rec
        SET transformed    = rec.data || acc.additions || COALESCE(rec.overrides, '{}'::jsonb),
            transformed_at = CURRENT_TIMESTAMP
        FROM _xform_acc acc
        WHERE rec.id = acc.id
        RETURNING rec.id
    )
    SELECT count(*) INTO v_count FROM updated;

    RETURN json_build_object('success', true, 'transformed', v_count);
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records. Rules are processed in sequence order — a rule at sequence N can read fields written by rules at sequence < N (chaining).';

------------------------------------------------------
-- Function: get_all_values
-- All extracted values (mapped + unmapped) with counts and mapping output
------------------------------------------------------
DROP FUNCTION IF EXISTS get_all_values(TEXT, TEXT);
CREATE FUNCTION get_all_values(
    p_source_name TEXT,
    p_rule_name TEXT DEFAULT NULL
) RETURNS TABLE (
    rule_name TEXT,
    output_field TEXT,
    source_field TEXT,
    extracted_value JSONB,
    record_count BIGINT,
    sample JSONB,
    mapping_id INTEGER,
    output JSONB,
    is_mapped BOOLEAN
) AS $$
BEGIN
    RETURN QUERY
    WITH extracted AS (
        SELECT
            r.name AS rule_name,
            r.output_field,
            r.field AS source_field,
            rec.transformed->r.output_field AS extracted_value,
            rec.data AS record_data,
            row_number() OVER (
                PARTITION BY r.name, rec.transformed->r.output_field
                ORDER BY rec.id
            ) AS rn
        FROM dataflow.records rec
        CROSS JOIN dataflow.rules r
        WHERE
            rec.source_name = p_source_name
            AND r.source_name = p_source_name
            AND rec.transformed IS NOT NULL
            AND rec.transformed ? r.output_field
            AND (p_rule_name IS NULL OR r.name = p_rule_name)
            AND rec.data ? r.field
    ),
    aggregated AS (
        SELECT
            e.rule_name,
            e.output_field,
            e.source_field,
            e.extracted_value,
            count(*) AS record_count,
            jsonb_agg(e.record_data ORDER BY e.rn) FILTER (WHERE e.rn <= 5) AS sample
        FROM extracted e
        GROUP BY e.rule_name, e.output_field, e.source_field, e.extracted_value
    )
    SELECT
        a.rule_name,
        a.output_field,
        a.source_field,
        a.extracted_value,
        a.record_count,
        a.sample,
        m.id AS mapping_id,
        m.output,
        (m.id IS NOT NULL) AS is_mapped
    FROM aggregated a
    LEFT JOIN dataflow.mappings m ON
        m.source_name = p_source_name
        AND m.rule_name = a.rule_name
        AND m.input_value = a.extracted_value
    ORDER BY a.record_count DESC;
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION get_all_values IS 'All extracted values with record counts and mapping output (single query for All tab)';

------------------------------------------------------
-- Function: get_unmapped_values
-- Find extracted values that need mappings
------------------------------------------------------
DROP FUNCTION IF EXISTS get_unmapped_values(TEXT, TEXT);
CREATE FUNCTION get_unmapped_values(
    p_source_name TEXT,
    p_rule_name TEXT DEFAULT NULL
) RETURNS TABLE (
    rule_name TEXT,
    output_field TEXT,
    source_field TEXT,
    extracted_value JSONB,
    record_count BIGINT,
    sample JSONB
) AS $$
BEGIN
    RETURN QUERY
    WITH extracted AS (
        SELECT
            r.name AS rule_name,
            r.output_field,
            r.field AS source_field,
            rec.transformed->r.output_field AS extracted_value,
            rec.data AS record_data,
            row_number() OVER (
                PARTITION BY r.name, rec.transformed->r.output_field
                ORDER BY rec.id
            ) AS rn
        FROM
            dataflow.records rec
            CROSS JOIN dataflow.rules r
        WHERE
            rec.source_name = p_source_name
            AND r.source_name = p_source_name
            AND rec.transformed IS NOT NULL
            AND rec.transformed ? r.output_field
            AND (p_rule_name IS NULL OR r.name = p_rule_name)
            AND rec.data ? r.field
    )
    SELECT
        e.rule_name,
        e.output_field,
        e.source_field,
        e.extracted_value,
        count(*) AS record_count,
        jsonb_agg(e.record_data ORDER BY e.rn) FILTER (WHERE e.rn <= 5) AS sample
    FROM extracted e
    WHERE NOT EXISTS (
        SELECT 1 FROM dataflow.mappings m
        WHERE m.source_name = p_source_name
          AND m.rule_name = e.rule_name
          AND m.input_value = e.extracted_value
    )
    GROUP BY e.rule_name, e.output_field, e.source_field, e.extracted_value
    ORDER BY count(*) DESC;
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION get_unmapped_values IS 'Find extracted values that need mappings defined';

------------------------------------------------------
-- Function: reprocess_records
-- Clear and reapply transformations
------------------------------------------------------
CREATE OR REPLACE FUNCTION reprocess_records(p_source_name TEXT)
RETURNS JSON AS $$
    -- Overwrite all records directly — no clear step, mirrors TPS srce_map_overwrite
    SELECT dataflow.apply_transformations(p_source_name, NULL, TRUE)
$$ LANGUAGE sql;

COMMENT ON FUNCTION reprocess_records IS 'Clear and reapply all transformations for a source';

------------------------------------------------------
-- Function: generate_source_view
-- Build a typed flat view in dfv schema
------------------------------------------------------
CREATE OR REPLACE FUNCTION generate_source_view(p_source_name TEXT)
RETURNS JSON AS $$
DECLARE
    v_config  JSONB;
    v_fields  JSONB;
    v_field   JSONB;
    v_cols    TEXT := '';
    v_sql     TEXT;
    v_view    TEXT;
BEGIN
    SELECT config INTO v_config
    FROM dataflow.sources
    WHERE name = p_source_name;

    IF v_config IS NULL OR NOT (v_config ? 'fields') OR jsonb_array_length(v_config->'fields') = 0 THEN
        RETURN json_build_object('success', false, 'error', 'No schema fields defined for this source');
    END IF;

    v_fields := v_config->'fields';

    FOR v_field IN SELECT * FROM jsonb_array_elements(v_fields)
    LOOP
        IF v_cols != '' THEN v_cols := v_cols || ', '; END IF;

        IF v_field->>'expression' IS NOT NULL THEN
            -- Computed expression: substitute {fieldname} refs with (transformed->>'fieldname')::type
            -- e.g. "{Amount} * {sign}" → "(transformed->>'Amount')::numeric * (transformed->>'sign')::numeric"
            DECLARE
                v_expr    TEXT := v_field->>'expression';
                v_ref     TEXT;
                v_cast    TEXT := COALESCE(NULLIF(v_field->>'type', ''), 'numeric');
            BEGIN
                WHILE v_expr ~ '\{[^}]+\}' LOOP
                    v_ref  := substring(v_expr FROM '\{([^}]+)\}');
                    v_expr := replace(v_expr, '{' || v_ref || '}',
                                      format('(transformed->>%L)::numeric', v_ref));
                END LOOP;
                v_cols := v_cols || format('%s AS %I', v_expr, v_field->>'name');
            END;
        ELSE
            CASE v_field->>'type'
                WHEN 'date' THEN
                    v_cols := v_cols || format('(transformed->>%L)::date AS %I',
                        v_field->>'name', v_field->>'name');
                WHEN 'numeric' THEN
                    v_cols := v_cols || format('(transformed->>%L)::numeric AS %I',
                        v_field->>'name', v_field->>'name');
                ELSE
                    v_cols := v_cols || format('transformed->>%L AS %I',
                        v_field->>'name', v_field->>'name');
            END CASE;
        END IF;
    END LOOP;

    CREATE SCHEMA IF NOT EXISTS dfv;

    v_view := 'dfv.' || quote_ident(p_source_name);

    EXECUTE format('DROP VIEW IF EXISTS %s', v_view);

    v_sql := format(
        'CREATE VIEW %s AS SELECT %s FROM dataflow.records WHERE source_name = %L AND transformed IS NOT NULL',
        v_view, v_cols, p_source_name
    );

    EXECUTE v_sql;

    RETURN json_build_object('success', true, 'view', v_view, 'sql', v_sql);
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION generate_source_view IS 'Generate a typed flat view in dfv schema from source config.fields';

------------------------------------------------------
-- Summary
------------------------------------------------------
-- Functions: 4 simple, focused functions
--   1. import_records - Import with deduplication
--   2. apply_transformations - Apply rules and mappings
--   3. get_unmapped_values - Find values needing mappings
--   4. reprocess_records - Re-transform all records
--
-- Each function does ONE thing clearly
-- No complex nested CTEs
-- Easy to understand and debug
------------------------------------------------------