/*
This is native in Oracle.
In PostgreSQL we must build it manually.

2) Table Structures
Target table (with constraints)
*/
DROP TABLE IF EXISTS dwh_account CASCADE;

CREATE TABLE dwh_account (
    acc_id   INT PRIMARY KEY,
    balance  NUMERIC(12,2) CHECK (balance >= 0),
    status   CHAR(1) NOT NULL
);

--Staging table (raw data)
DROP TABLE IF EXISTS stg_account;

CREATE TABLE stg_account (
    acc_id   INT,
    balance  NUMERIC(12,2),
    status   CHAR(1)
);

--Reject table (manual log)
DROP TABLE IF EXISTS reject_account;

CREATE TABLE reject_account (
    acc_id     INT,
    balance    NUMERIC(12,2),
    status     CHAR(1),
    error_text TEXT,
    reject_ts  TIMESTAMPTZ DEFAULT now()
);


TRUNCATE stg_account;

INSERT INTO stg_account(acc_id, balance, status) VALUES
(1001, 1000, 'A'),   -- valid
(1002, -50,  'A'),   -- CHECK violation
(1003,  500, NULL),  -- NOT NULL violation
(1001,  300, 'A'),   -- duplicate PK
(1004,  200, 'B');   -- valid

--Insert only valid rows
INSERT INTO dwh_account(acc_id, balance, status)
SELECT acc_id, balance, status
FROM stg_account
WHERE balance >= 0
  AND status IS NOT NULL;

--Log rejected rows by rules
INSERT INTO reject_account(acc_id, balance, status, error_text)
SELECT acc_id, balance, status,
       CASE
         WHEN balance < 0 THEN 'Rejected: balance < 0'
         WHEN status IS NULL THEN 'Rejected: status is NULL'
         ELSE 'Rejected by rule'
       END
FROM stg_account
WHERE balance < 0
   OR status IS NULL;

/*
Fast and set-based
But does not handle duplicates or unexpected errors
Approach B – Handling Duplicates with ON CONFLICT
This covers primary key violations.
*/
INSERT INTO dwh_account(acc_id, balance, status)
SELECT acc_id, balance, status
FROM stg_account
WHERE balance >= 0
  AND status IS NOT NULL
ON CONFLICT (acc_id) DO NOTHING;

--Log duplicates
INSERT INTO reject_account(acc_id, balance, status, error_text)
SELECT s.acc_id, s.balance, s.status, 'Rejected: duplicate acc_id'
FROM stg_account s
WHERE EXISTS (
    SELECT 1
    FROM dwh_account d
    WHERE d.acc_id = s.acc_id
);


--Procedure
CREATE OR REPLACE PROCEDURE load_accounts(p_batch_id INT)
LANGUAGE plpgsql
AS $$
DECLARE
    r RECORD;
BEGIN
    FOR r IN SELECT * FROM stg_account LOOP
        BEGIN
            INSERT INTO dwh_account(acc_id, balance, status)
            VALUES (r.acc_id, r.balance, r.status);

        EXCEPTION WHEN others THEN
            INSERT INTO reject_account(
                acc_id,
                balance,
                status,
                error_text
            )
            VALUES (
                r.acc_id,
                r.balance,
                r.status,
                'BATCH=' || p_batch_id || ' ERROR=' || SQLERRM
            );
        END;
    END LOOP;
END;
$$;

--Run the load
CALL load_accounts(1);