dataflow/database/transform.sql

--
-- Transform queries
-- The rule/mapping engine; SQL for the transform endpoints in api/routes/sources.js,
-- api/routes/rules.js and api/routes/records.js
--
-- Order matters within this file: the aggregate is used by apply_transformations,
-- which in turn is called by reprocess_records.
--

SET search_path TO dataflow, public;

-- ── Merge aggregate ───────────────────────────────────────────────────────────

-- 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 = '{}'
);

-- ── Apply rules and mappings ──────────────────────────────────────────────────

-- Writes only the rule/mapping output into records.transformed. Raw values stay in
-- data and manual edits stay in overrides; readers merge the three layers.
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 $$
WITH
-- All records to process
qualifying AS (
    SELECT id, data
    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))
),
-- Mirror TPS rx: fan out one row per regex match, drive from rules → records
rx AS (
    SELECT
        q.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,
        -- extract: build map_val and retain_val per match (mirrors TPS)
        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 qualifying q ON q.data ? r.field
    LEFT JOIN LATERAL regexp_matches(q.data ->> r.field, r.pattern, r.flags)
        WITH ORDINALITY AS mt(mt, rn) ON r.function_type = 'extract'
    LEFT JOIN LATERAL regexp_replace(q.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.enabled = true
),
-- Aggregate match rows back into one value per (record, rule) — mirrors TPS agg_to_target_items
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
),
-- Join with mappings to find mapped output — mirrors TPS link_map
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
),
-- Build per-rule output JSONB:
--   mapped  → use mapping output; also write output_field if retain = true
--   no map  → write extracted value to output_field
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
),
-- Merge all rule outputs per record in sequence order — mirrors TPS agg_to_id
record_additions AS (
    SELECT
        id,
        dataflow.jsonb_concat_obj(output ORDER BY sequence) AS additions
    FROM rule_output
    GROUP BY id
),
-- Update all qualifying records; records with no rule matches get an empty object
updated AS (
    UPDATE dataflow.records rec
    SET transformed    = COALESCE(ra.additions, '{}'::jsonb),
        transformed_at = CURRENT_TIMESTAMP
    FROM qualifying q
    LEFT JOIN record_additions ra ON ra.id = q.id
    WHERE rec.id = q.id
    RETURNING rec.id
)
SELECT json_build_object('success', true, 'transformed', count(*))
FROM updated
$$ LANGUAGE sql;

COMMENT ON FUNCTION apply_transformations IS 'Apply transformation rules and mappings to records (set-based CTE)';

-- ── Reprocess ─────────────────────────────────────────────────────────────────

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 'Reapply all transformations for a source, overwriting existing values';