-- 047_audit_partitioning.sql
--
-- Makes the audit schema survive thousands of stores.
--
-- Measured on the dev database: interface_log is ~1.5 KB/row, change_log
-- ~650 B/row. Extrapolated to 1,000 active stores that is roughly 1.5 GB/day of
-- interface log, 130 MB/day of change log and ~900 MB/day of activity — call it
-- 170 GB of steady state under the retention policy, on a box that also serves
-- the live database.
--
-- Three things follow from that:
--
--  1. RETENTION MUST BE A PARTITION DROP. Deleting 45 GB of expired rows with
--     DELETE rewrites the table's visibility map, bloats it, and triggers a
--     vacuum storm on the same spindle as live traffic. Dropping a monthly
--     partition is O(1) and returns the space immediately.
--
--  2. INDEXES ARE A WRITE TAX. Every index on change_log is maintained on every
--     audited write, and pg_stat_user_indexes shows several with zero scans.
--     They are dropped here; the ones that remain are the ones the admin and
--     merchant views actually issue.
--
--  3. TIME COLUMNS WANT BRIN, NOT BTREE. These tables are append-only and
--     physically time-ordered, which is precisely BRIN's case: a few kilobytes
--     instead of the ~1.6 GB a btree on 47 GB of timestamps would cost.
--
-- Safe to re-run: a table that is already partitioned is left alone.
--
-- NOTE FOR PRODUCTION: the copy below is a single INSERT ... SELECT, which is
-- fine at dev size and NOT fine at 45 GB. On a large table, create the
-- partitioned table alongside, copy in month-sized chunks, then swap. This
-- migration is written for the current data volume and says so rather than
-- pretending to be online.

BEGIN;

-- ─────────────────────────────────────────────────────────────────────────────
-- Partition helper (referenced by AuditRetentionService.ensureFuturePartitions)
-- ─────────────────────────────────────────────────────────────────────────────

CREATE OR REPLACE FUNCTION audit.ensure_partition(p_table TEXT, p_month DATE)
RETURNS VOID
LANGUAGE plpgsql
AS $fn$
DECLARE
  v_start DATE := date_trunc('month', p_month)::DATE;
  v_end   DATE := (date_trunc('month', p_month) + interval '1 month')::DATE;
  v_name  TEXT := format('%s_%s', p_table, to_char(v_start, 'YYYY_MM'));
BEGIN
  IF EXISTS (
    SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
     WHERE n.nspname = 'audit' AND c.relname = v_name
  ) THEN
    RETURN;
  END IF;

  EXECUTE format(
    'CREATE TABLE audit.%I PARTITION OF audit.%I FOR VALUES FROM (%L) TO (%L)',
    v_name, p_table, v_start, v_end
  );
END;
$fn$;

-- ─────────────────────────────────────────────────────────────────────────────
-- Convert one table to monthly range partitioning
-- ─────────────────────────────────────────────────────────────────────────────

CREATE OR REPLACE FUNCTION audit.partition_table(
  p_table   TEXT,
  p_time_col TEXT,
  p_pk_col  TEXT
) RETURNS VOID
LANGUAGE plpgsql
AS $fn$
DECLARE
  v_kind   CHAR;
  v_min    DATE;
  v_max    DATE;
  v_cursor DATE;
BEGIN
  SELECT c.relkind INTO v_kind
    FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
   WHERE n.nspname = 'audit' AND c.relname = p_table;

  IF v_kind IS NULL THEN
    RAISE NOTICE 'audit.partition_table: %.% missing, skipping', 'audit', p_table;
    RETURN;
  END IF;

  IF v_kind = 'p' THEN
    RAISE NOTICE 'audit.% is already partitioned', p_table;
    RETURN;
  END IF;

  EXECUTE format('ALTER TABLE audit.%I RENAME TO %I', p_table, p_table || '_legacy');

  -- The partition key must be part of the primary key, so the PK becomes
  -- (id, time). Nothing looks these rows up by bare id — the audit views query
  -- by store, batch, record or time — so this costs nothing in practice.
  EXECUTE format(
    'CREATE TABLE audit.%I (LIKE audit.%I INCLUDING DEFAULTS INCLUDING CONSTRAINTS)
       PARTITION BY RANGE (%I)',
    p_table, p_table || '_legacy', p_time_col
  );

  EXECUTE format(
    'ALTER TABLE audit.%I ADD PRIMARY KEY (%I, %I)',
    p_table, p_pk_col, p_time_col
  );

  EXECUTE format('SELECT min(%I)::date, max(%I)::date FROM audit.%I',
                 p_time_col, p_time_col, p_table || '_legacy')
     INTO v_min, v_max;

  v_min := COALESCE(date_trunc('month', v_min)::DATE, date_trunc('month', now())::DATE);
  v_max := COALESCE(date_trunc('month', v_max)::DATE, date_trunc('month', now())::DATE);

  v_cursor := v_min;
  WHILE v_cursor <= v_max + interval '2 months' LOOP
    PERFORM audit.ensure_partition(p_table, v_cursor);
    v_cursor := (v_cursor + interval '1 month')::DATE;
  END LOOP;

  -- A row outside every range would otherwise fail to INSERT — and an audit
  -- write must never be the thing that breaks a user's request. The default
  -- partition catches clock skew and back-dated writes instead.
  EXECUTE format(
    'CREATE TABLE audit.%I PARTITION OF audit.%I DEFAULT',
    p_table || '_default', p_table
  );

  EXECUTE format('INSERT INTO audit.%I SELECT * FROM audit.%I', p_table, p_table || '_legacy');
  EXECUTE format('DROP TABLE audit.%I', p_table || '_legacy');
END;
$fn$;

-- ─────────────────────────────────────────────────────────────────────────────
-- Convert
-- ─────────────────────────────────────────────────────────────────────────────

SELECT audit.partition_table('change_log',    'changedOn',   'id');
SELECT audit.partition_table('interface_log', 'requestedOn', 'id');
SELECT audit.partition_table('activity_log',  'receivedAt',  'id');
SELECT audit.partition_table('change_batch',  'startedAt',   'batchId');

-- ─────────────────────────────────────────────────────────────────────────────
-- Indexes: only what the views actually issue
-- ─────────────────────────────────────────────────────────────────────────────
--
-- Created on the partitioned parents, so Postgres creates and drops the matching
-- per-partition index automatically as months come and go.

-- change_log — read by batch (expand an action), by record (history timeline),
-- and by store+time (the flat list).
CREATE INDEX IF NOT EXISTS "change_log_batchId_idx"
  ON audit.change_log ("batchId");
CREATE INDEX IF NOT EXISTS "change_log_record_idx"
  ON audit.change_log ("schemaName", "tableName", "recordId");
CREATE INDEX IF NOT EXISTS "change_log_store_time_idx"
  ON audit.change_log ("storeId", "changedOn" DESC);
-- Append-only and time-ordered: BRIN costs kilobytes where a btree costs GB.
CREATE INDEX IF NOT EXISTS "change_log_changedOn_brin"
  ON audit.change_log USING BRIN ("changedOn") WITH (pages_per_range = 32);
-- Correlation view only; partial so it costs nothing for the rows that have none.
CREATE INDEX IF NOT EXISTS "change_log_correlation_idx"
  ON audit.change_log ("correlationId") WHERE "correlationId" IS NOT NULL;

-- interface_log — read by store+time, by status bucket, and by correlation.
CREATE INDEX IF NOT EXISTS "interface_log_store_time_idx"
  ON audit.interface_log ("storeId", "requestedOn" DESC);
CREATE INDEX IF NOT EXISTS "interface_log_requestedOn_brin"
  ON audit.interface_log USING BRIN ("requestedOn") WITH (pages_per_range = 32);
-- Failures are the needle in this haystack, and they are a tiny fraction of it.
CREATE INDEX IF NOT EXISTS "interface_log_failures_idx"
  ON audit.interface_log ("storeId", "requestedOn" DESC) WHERE "statusCode" >= 400;
CREATE INDEX IF NOT EXISTS "interface_log_correlation_idx"
  ON audit.interface_log ("correlationId") WHERE "correlationId" IS NOT NULL;

-- activity_log — store+time for the feed, session/event for analysis.
CREATE INDEX IF NOT EXISTS "activity_log_store_time_idx"
  ON audit.activity_log ("storeId", "receivedAt" DESC);
CREATE INDEX IF NOT EXISTS "activity_log_receivedAt_brin"
  ON audit.activity_log USING BRIN ("receivedAt") WITH (pages_per_range = 32);
CREATE INDEX IF NOT EXISTS "activity_log_session_idx"
  ON audit.activity_log ("sessionId") WHERE "sessionId" IS NOT NULL;
CREATE INDEX IF NOT EXISTS "activity_log_event_idx"
  ON audit.activity_log ("storeId", "eventName");

-- change_batch — the merchant/admin list reads this and almost nothing else.
CREATE INDEX IF NOT EXISTS "change_batch_store_time_idx"
  ON audit.change_batch ("storeId", "startedAt" DESC);
CREATE INDEX IF NOT EXISTS "change_batch_actor_idx"
  ON audit.change_batch ("storeId", "actorId", "startedAt" DESC);
-- Drives the reconciler's watermark: only unreconciled headers, so the index
-- stays tiny however large the table grows.
CREATE INDEX IF NOT EXISTS "change_batch_unreconciled_idx"
  ON audit.change_batch ("startedAt") WHERE "rowCount" = 0;

COMMIT;