-- @query configure_autoload
SET autoload_known_extensions = false;
-- @query configure_external_access
SET enable_external_access = false;
-- @query create_event_ledger
CREATE TABLE IF NOT EXISTS event_ledger (
  event_id VARCHAR PRIMARY KEY,
  payload_sha256 VARCHAR NOT NULL,
  order_id VARCHAR NOT NULL,
  source_version BIGINT NOT NULL,
  source_updated_at VARCHAR NOT NULL,
  order_date DATE NOT NULL,
  channel VARCHAR NOT NULL,
  amount_cents BIGINT NOT NULL
);
-- @query create_order_state
CREATE TABLE IF NOT EXISTS order_state (
  order_id VARCHAR PRIMARY KEY,
  source_event_id VARCHAR NOT NULL,
  source_version BIGINT NOT NULL,
  source_updated_at VARCHAR NOT NULL,
  order_date DATE NOT NULL,
  channel VARCHAR NOT NULL,
  amount_cents BIGINT NOT NULL
);
-- @query create_orders_fact
CREATE TABLE IF NOT EXISTS orders_fact (
  order_id VARCHAR PRIMARY KEY,
  source_event_id VARCHAR NOT NULL,
  source_version BIGINT NOT NULL,
  source_updated_at VARCHAR NOT NULL,
  order_date DATE NOT NULL,
  channel VARCHAR NOT NULL,
  amount_cents BIGINT NOT NULL
);
-- @query create_daily_mart
CREATE OR REPLACE VIEW daily_channel_mart AS
SELECT order_date, channel, COUNT(*) AS order_count, SUM(amount_cents) AS amount_cents
FROM orders_fact
GROUP BY order_date, channel;
-- @query get_event_ledger
SELECT event_id, payload_sha256 FROM event_ledger ORDER BY event_id;
-- @query get_order_state
SELECT order_id, source_event_id, source_version, source_updated_at,
       CAST(order_date AS VARCHAR) AS order_date, channel, amount_cents
FROM order_state ORDER BY order_id;
-- @query begin_transaction
BEGIN TRANSACTION;
-- @query insert_event_ledger
INSERT INTO event_ledger VALUES (?, ?, ?, ?, ?, CAST(? AS DATE), ?, ?);
-- @query upsert_order_state
INSERT INTO order_state VALUES (?, ?, ?, ?, CAST(? AS DATE), ?, ?)
ON CONFLICT (order_id) DO UPDATE SET
  source_event_id = excluded.source_event_id,
  source_version = excluded.source_version,
  source_updated_at = excluded.source_updated_at,
  order_date = excluded.order_date,
  channel = excluded.channel,
  amount_cents = excluded.amount_cents
WHERE excluded.source_version > order_state.source_version
   OR (excluded.source_version = order_state.source_version AND excluded.source_updated_at > order_state.source_updated_at)
   OR (excluded.source_version = order_state.source_version AND excluded.source_updated_at = order_state.source_updated_at AND excluded.source_event_id > order_state.source_event_id);
-- @query upsert_order_fact
INSERT INTO orders_fact VALUES (?, ?, ?, ?, CAST(? AS DATE), ?, ?)
ON CONFLICT (order_id) DO UPDATE SET
  source_event_id = excluded.source_event_id,
  source_version = excluded.source_version,
  source_updated_at = excluded.source_updated_at,
  order_date = excluded.order_date,
  channel = excluded.channel,
  amount_cents = excluded.amount_cents
WHERE excluded.source_version > orders_fact.source_version
   OR (excluded.source_version = orders_fact.source_version AND excluded.source_updated_at > orders_fact.source_updated_at)
   OR (excluded.source_version = orders_fact.source_version AND excluded.source_updated_at = orders_fact.source_updated_at AND excluded.source_event_id > orders_fact.source_event_id);
-- @query query_orders_fact
SELECT order_id, source_event_id, source_version, source_updated_at,
       CAST(order_date AS VARCHAR) AS order_date, channel, amount_cents
FROM orders_fact ORDER BY order_id;
-- @query query_daily_mart
SELECT CAST(order_date AS VARCHAR) AS order_date, channel, order_count, amount_cents
FROM daily_channel_mart ORDER BY order_date, channel;
-- @query check_state_fact
SELECT COUNT(*) AS mismatch_count
FROM order_state AS state
FULL OUTER JOIN orders_fact AS fact USING (order_id)
WHERE state.source_event_id IS DISTINCT FROM fact.source_event_id
   OR state.source_version IS DISTINCT FROM fact.source_version
   OR state.source_updated_at IS DISTINCT FROM fact.source_updated_at
   OR state.order_date IS DISTINCT FROM fact.order_date
   OR state.channel IS DISTINCT FROM fact.channel
   OR state.amount_cents IS DISTINCT FROM fact.amount_cents;
-- @query check_daily_mart
WITH fact_groups AS (
  SELECT order_date, channel, COUNT(*) AS order_count, SUM(amount_cents) AS amount_cents
  FROM orders_fact GROUP BY order_date, channel
), mismatches AS (
  (SELECT * FROM fact_groups EXCEPT SELECT * FROM daily_channel_mart)
  UNION ALL
  (SELECT * FROM daily_channel_mart EXCEPT SELECT * FROM fact_groups)
)
SELECT COUNT(*) AS mismatch_count FROM mismatches;
-- @query check_nonnegative
SELECT COUNT(*) AS invalid_count FROM orders_fact WHERE amount_cents < 0;
-- @query commit_transaction
COMMIT;
-- @query rollback_transaction
ROLLBACK;