-- 045_audit_batching_and_triggers.sql
--
-- Replaces the hand-written change-log mechanism with database triggers, and
-- adds the batch grouping the admin change view reads.
--
-- Why triggers rather than application calls: the previous approach had 135
-- hand-written logChange() call sites, of which 2 recorded the before-state and
-- 0 recorded a correlation id. A trigger gets both for free from
-- to_jsonb(OLD)/to_jsonb(NEW), and cannot be bypassed by a seed, a cron
-- service, a migration, or a manual psql edit.
--
-- Idempotent: safe to re-run.

BEGIN;

CREATE SCHEMA IF NOT EXISTS audit;

-- ─────────────────────────────────────────────────────────────────────────────
-- 1. New columns on the existing tables
-- ─────────────────────────────────────────────────────────────────────────────

ALTER TABLE audit.change_log
  ADD COLUMN IF NOT EXISTS "batchId" TEXT,
  ADD COLUMN IF NOT EXISTS "txId"    BIGINT;

ALTER TABLE audit.interface_log
  ADD COLUMN IF NOT EXISTS "batchId"      TEXT,
  ADD COLUMN IF NOT EXISTS "routePattern" TEXT,
  ADD COLUMN IF NOT EXISTS "truncated"    BOOLEAN NOT NULL DEFAULT FALSE,
  ADD COLUMN IF NOT EXISTS "sampleRate"   INTEGER NOT NULL DEFAULT 1;

-- The foreign keys from the audit tables to organizations/stores made every
-- audited write pay two constraint checks and take locks on those tables while
-- under concurrent load. The ids stay as plain columns.
ALTER TABLE audit.change_log
  DROP CONSTRAINT IF EXISTS "change_log_organizationId_fkey",
  DROP CONSTRAINT IF EXISTS "change_log_storeId_fkey";

ALTER TABLE audit.interface_log
  DROP CONSTRAINT IF EXISTS "interface_log_organizationId_fkey",
  DROP CONSTRAINT IF EXISTS "interface_log_storeId_fkey";

-- High-cardinality text index on the write path, serving a query the admin UI
-- does not issue. Dropped.
DROP INDEX IF EXISTS audit."interface_log_method_path_idx";

-- ─────────────────────────────────────────────────────────────────────────────
-- 2. Batch header — one row per user action
-- ─────────────────────────────────────────────────────────────────────────────

CREATE TABLE IF NOT EXISTS audit.change_batch (
  "batchId"        TEXT PRIMARY KEY,
  "organizationId" TEXT,
  "storeId"        TEXT,
  "actorId"        TEXT,
  "actorRole"      TEXT,
  "actorName"      TEXT,
  "ipAddress"      TEXT,
  "userAgent"      TEXT,
  "correlationId"  TEXT,
  "action"         TEXT,
  "summary"        TEXT,
  "route"          TEXT,
  "method"         TEXT,
  "source"         TEXT        NOT NULL DEFAULT 'api',
  "rowCount"       INTEGER     NOT NULL DEFAULT 0,
  "tableCount"     INTEGER     NOT NULL DEFAULT 0,
  "startedAt"      TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
  "endedAt"        TIMESTAMP(3)
);

CREATE INDEX IF NOT EXISTS "change_batch_storeId_startedAt_idx" ON audit.change_batch ("storeId", "startedAt" DESC);
CREATE INDEX IF NOT EXISTS "change_batch_organizationId_idx"    ON audit.change_batch ("organizationId");
CREATE INDEX IF NOT EXISTS "change_batch_actorId_idx"           ON audit.change_batch ("actorId");
CREATE INDEX IF NOT EXISTS "change_batch_correlationId_idx"     ON audit.change_batch ("correlationId");
CREATE INDEX IF NOT EXISTS "change_batch_startedAt_idx"         ON audit.change_batch ("startedAt" DESC);
CREATE INDEX IF NOT EXISTS "change_batch_action_idx"            ON audit.change_batch ("action");

CREATE INDEX IF NOT EXISTS "change_log_batchId_idx"         ON audit.change_log ("batchId");
CREATE INDEX IF NOT EXISTS "change_log_storeId_batchId_idx" ON audit.change_log ("storeId", "batchId");
CREATE INDEX IF NOT EXISTS "change_log_txId_idx"            ON audit.change_log ("txId");
CREATE INDEX IF NOT EXISTS "interface_log_batchId_idx"      ON audit.interface_log ("batchId");
CREATE INDEX IF NOT EXISTS "interface_log_direction_idx"    ON audit.interface_log ("direction");
CREATE INDEX IF NOT EXISTS "interface_log_storeId_requestedOn_idx" ON audit.interface_log ("storeId", "requestedOn" DESC);

-- Deliberately NO foreign key from change_log.batchId to change_batch.
--
-- This was tried and reverted during implementation: with the constraint in
-- place, a trigger firing before the application had written the batch header
-- raised 23503, and because the trigger runs inside the caller's transaction
-- that error rolled back the user's actual business write. An audit trail that
-- can abort a checkout is worse than one with an occasional orphan row.
-- Detail rows and headers are joined in the read query instead, and orphans
-- stay visible grouped by txId.
ALTER TABLE audit.change_log DROP CONSTRAINT IF EXISTS "change_log_batchId_fkey";

-- ─────────────────────────────────────────────────────────────────────────────
-- 3. Activity log — UI clicks and views
-- ─────────────────────────────────────────────────────────────────────────────

CREATE TABLE IF NOT EXISTS audit.activity_log (
  "id"             TEXT PRIMARY KEY,
  "organizationId" TEXT,
  "storeId"        TEXT,
  "userId"         TEXT,
  "anonymousId"    TEXT,
  "sessionId"      TEXT,
  "app"            TEXT        NOT NULL DEFAULT 'web',
  "eventName"      TEXT        NOT NULL,
  "route"          TEXT,
  "targetId"       TEXT,
  "props"          JSONB,
  "occurredAt"     TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
  "receivedAt"     TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
  "userAgent"      TEXT,
  "ipHash"         TEXT,
  "referrer"       TEXT
);

CREATE INDEX IF NOT EXISTS "activity_log_storeId_receivedAt_idx" ON audit.activity_log ("storeId", "receivedAt" DESC);
CREATE INDEX IF NOT EXISTS "activity_log_eventName_idx"          ON audit.activity_log ("eventName");
CREATE INDEX IF NOT EXISTS "activity_log_sessionId_idx"          ON audit.activity_log ("sessionId");
CREATE INDEX IF NOT EXISTS "activity_log_userId_idx"             ON audit.activity_log ("userId");
CREATE INDEX IF NOT EXISTS "activity_log_anonymousId_idx"        ON audit.activity_log ("anonymousId");
CREATE INDEX IF NOT EXISTS "activity_log_receivedAt_idx"         ON audit.activity_log ("receivedAt" DESC);

-- ─────────────────────────────────────────────────────────────────────────────
-- 4. Enrolment registry
-- ─────────────────────────────────────────────────────────────────────────────
--
-- Tier 1 = old + new value, long retention (money and identity).
-- Tier 2 = new value only, high churn / low forensic value.
-- Tier 3 = never audited; listed explicitly rather than implied by absence, so
--          "is this table deliberately unaudited?" has an answer in the data.

CREATE TABLE IF NOT EXISTS audit.audited_tables (
  "schemaName"       TEXT NOT NULL,
  "tableName"        TEXT NOT NULL,
  "tier"             SMALLINT NOT NULL DEFAULT 1,
  "captureOldValue"  BOOLEAN NOT NULL DEFAULT TRUE,
  -- Columns whose change alone must NOT produce a row. Without this, every
  -- heartbeat/counter write generates a change row that says nothing changed —
  -- this is the single biggest volume control on the change log.
  "excludedColumns"  TEXT[] NOT NULL DEFAULT ARRAY['updatedOn','updatedAt','updatedBy','lastSeenAt','viewCount','searchVector']::TEXT[],
  "enabled"          BOOLEAN NOT NULL DEFAULT TRUE,
  "note"             TEXT,
  PRIMARY KEY ("schemaName", "tableName")
);

-- ─────────────────────────────────────────────────────────────────────────────
-- 5. The trigger function
-- ─────────────────────────────────────────────────────────────────────────────

CREATE OR REPLACE FUNCTION audit.fn_log_change() RETURNS TRIGGER
LANGUAGE plpgsql
SECURITY DEFINER
AS $fn$
DECLARE
  v_actor        JSONB;
  v_old          JSONB;
  v_new          JSONB;
  v_changed      TEXT[];
  v_excluded     TEXT[] := COALESCE(TG_ARGV[0], '')::TEXT[];
  v_capture_old  BOOLEAN := COALESCE(TG_ARGV[1], 'true')::BOOLEAN;
  v_record_id    TEXT;
  v_store_id     TEXT;
  v_org_id       TEXT;
  v_operation    TEXT;
BEGIN
  -- Transaction-local, set by the application at the start of the transaction.
  -- A session-level SET would leak one user's identity onto the next request
  -- that borrows the pooled connection; for an audit trail, wrong attribution
  -- is worse than none, so this is deliberately the transaction-scoped form.
  BEGIN
    v_actor := NULLIF(current_setting('audit.actor', TRUE), '')::JSONB;
  EXCEPTION WHEN OTHERS THEN
    v_actor := NULL;
  END;

  IF TG_OP = 'INSERT' THEN
    v_operation := 'create';
    v_new := to_jsonb(NEW);
    v_old := NULL;
  ELSIF TG_OP = 'UPDATE' THEN
    v_operation := 'update';
    v_new := to_jsonb(NEW);
    v_old := to_jsonb(OLD);
  ELSE
    v_operation := 'delete';
    v_new := NULL;
    v_old := to_jsonb(OLD);
  END IF;

  -- Changed columns = keys whose value actually differs, minus the exclusions.
  IF TG_OP = 'UPDATE' THEN
    SELECT COALESCE(array_agg(key ORDER BY key), ARRAY[]::TEXT[])
      INTO v_changed
      FROM jsonb_each(v_new) AS n(key, value)
     WHERE NOT (key = ANY (v_excluded))
       AND n.value IS DISTINCT FROM (v_old -> n.key);

    -- Nothing of substance changed (a touched updatedOn, a bumped counter).
    -- Returning early here is what keeps the change log from filling with rows
    -- that record no change.
    IF array_length(v_changed, 1) IS NULL THEN
      RETURN NULL;
    END IF;
  ELSE
    SELECT COALESCE(array_agg(key ORDER BY key), ARRAY[]::TEXT[])
      INTO v_changed
      FROM jsonb_object_keys(COALESCE(v_new, v_old)) AS k(key)
     WHERE NOT (key = ANY (v_excluded));
  END IF;

  v_record_id := COALESCE(v_new ->> 'id', v_old ->> 'id', '');
  v_store_id  := COALESCE(v_new ->> 'storeId', v_old ->> 'storeId', v_actor ->> 'storeId');
  v_org_id    := COALESCE(v_new ->> 'organizationId', v_old ->> 'organizationId', v_actor ->> 'organizationId');

  INSERT INTO audit.change_log (
    "id", "organizationId", "storeId", "schemaName", "tableName", "recordId",
    "operation", "changedColumns", "oldValue", "newValue", "changedBy",
    "changedOn", "correlationId", "source", "actorRole", "actorName",
    "ipAddress", "batchId", "txId", "createdOn", "updatedOn"
  ) VALUES (
    gen_random_uuid()::TEXT,
    v_org_id,
    v_store_id,
    TG_TABLE_SCHEMA,
    TG_TABLE_NAME,
    v_record_id,
    v_operation,
    to_jsonb(v_changed),
    CASE WHEN v_capture_old THEN v_old ELSE NULL END,
    v_new,
    COALESCE(v_actor ->> 'actorId', 'system'),
    clock_timestamp(),
    v_actor ->> 'correlationId',
    COALESCE(v_actor ->> 'source', 'trigger'),
    v_actor ->> 'actorRole',
    v_actor ->> 'actorName',
    v_actor ->> 'ipAddress',
    v_actor ->> 'batchId',
    pg_current_xact_id()::TEXT::BIGINT,
    clock_timestamp(),
    clock_timestamp()
  );

  RETURN NULL; -- AFTER trigger; return value is ignored
END;
$fn$;

-- ─────────────────────────────────────────────────────────────────────────────
-- 6. Enrol / un-enrol helpers
-- ─────────────────────────────────────────────────────────────────────────────

CREATE OR REPLACE FUNCTION audit.enroll_table(
  p_schema TEXT,
  p_table  TEXT,
  p_tier   SMALLINT DEFAULT 1
) RETURNS VOID
LANGUAGE plpgsql
AS $fn$
DECLARE
  v_excluded    TEXT[];
  v_capture_old BOOLEAN;
  v_trigger     TEXT := format('trg_audit_%s', p_table);
BEGIN
  IF NOT EXISTS (
    SELECT 1 FROM information_schema.tables
     WHERE table_schema = p_schema AND table_name = p_table
  ) THEN
    RAISE NOTICE 'audit.enroll_table: %.% does not exist, skipping', p_schema, p_table;
    RETURN;
  END IF;

  INSERT INTO audit.audited_tables ("schemaName", "tableName", "tier", "captureOldValue")
  VALUES (p_schema, p_table, p_tier, p_tier = 1)
  ON CONFLICT ("schemaName", "tableName") DO UPDATE
    SET "tier" = EXCLUDED."tier",
        "captureOldValue" = EXCLUDED."captureOldValue",
        "enabled" = TRUE;

  SELECT "excludedColumns", "captureOldValue"
    INTO v_excluded, v_capture_old
    FROM audit.audited_tables
   WHERE "schemaName" = p_schema AND "tableName" = p_table;

  EXECUTE format('DROP TRIGGER IF EXISTS %I ON %I.%I', v_trigger, p_schema, p_table);
  EXECUTE format(
    'CREATE TRIGGER %I AFTER INSERT OR UPDATE OR DELETE ON %I.%I
       FOR EACH ROW EXECUTE FUNCTION audit.fn_log_change(%L, %L)',
    v_trigger, p_schema, p_table, v_excluded::TEXT, v_capture_old::TEXT
  );
END;
$fn$;

CREATE OR REPLACE FUNCTION audit.unenroll_table(p_schema TEXT, p_table TEXT)
RETURNS VOID
LANGUAGE plpgsql
AS $fn$
BEGIN
  EXECUTE format('DROP TRIGGER IF EXISTS %I ON %I.%I', format('trg_audit_%s', p_table), p_schema, p_table);
  UPDATE audit.audited_tables SET "enabled" = FALSE
   WHERE "schemaName" = p_schema AND "tableName" = p_table;
END;
$fn$;

-- Re-applies triggers from the registry. Run after changing excludedColumns,
-- and after any `prisma db push` — Prisma does not know about these triggers
-- and a table rewrite drops them.
CREATE OR REPLACE FUNCTION audit.sync_triggers() RETURNS INTEGER
LANGUAGE plpgsql
AS $fn$
DECLARE r RECORD; n INTEGER := 0;
BEGIN
  FOR r IN SELECT "schemaName", "tableName", "tier" FROM audit.audited_tables WHERE "enabled" LOOP
    PERFORM audit.enroll_table(r."schemaName", r."tableName", r."tier");
    n := n + 1;
  END LOOP;
  RETURN n;
END;
$fn$;

-- ─────────────────────────────────────────────────────────────────────────────
-- 7. Tier 1 enrolment — money and identity
-- ─────────────────────────────────────────────────────────────────────────────

DO $$
DECLARE
  t TEXT;
  tier1 TEXT[] := ARRAY[
    -- Money
    'sales.orders', 'sales.order_items', 'sales.order_status_history',
    'sales.payments', 'sales.payment_transactions',
    'sales.refunds', 'sales.refund_items', 'sales.shipments',
    'cart.checkout_sessions',
    -- Identity and access
    'admin.users', 'store.store_staff', 'store.store_staff_permissions',
    'store.stores', 'store.store_domains', 'platform.organizations',
    'platform.roles', 'platform.permissions', 'platform.role_permissions',
    'platform.invite_tokens',
    -- Catalogue and price
    'products.products', 'products.product_variants', 'products.product_images',
    'products.product_options', 'products.product_option_values',
    'products.brands', 'products.collections', 'products.inventory',
    'master.categories',
    -- Stock
    'inventory.inventory_items', 'inventory.inventory_levels',
    'inventory.inventory_movements', 'inventory.locations',
    -- Pricing rules and tax
    'marketing.promotions', 'marketing.promotion_rules', 'marketing.discount_codes',
    'master.tax_slab_profiles',
    -- Store configuration that changes what a customer is charged
    'store.payment_provider_settings', 'store.store_settings', 'store.store_branding',
    'store.shipping_settings', 'store.shipping_rates', 'store.shipping_zones',
    'store.shipping_zone_rates', 'store.shipping_channels',
    'store.shipping_weight_slabs', 'store.shipping_value_slabs',
    'store.shipping_distance_bands',
    'platform.platform_settings', 'platform.email_settings',
    -- Customers
    'customer.customers', 'customer.customer_addresses',
    'customer.customer_store_memberships', 'admin.addresses'
  ];
BEGIN
  FOREACH t IN ARRAY tier1 LOOP
    PERFORM audit.enroll_table(split_part(t, '.', 1), split_part(t, '.', 2), 1::SMALLINT);
  END LOOP;
END $$;

-- Tier 2 — audited, but new-value only. High churn, low forensic value: we
-- want to know a cart changed, not to keep both halves of every quantity bump.
DO $$
DECLARE
  t TEXT;
  tier2 TEXT[] := ARRAY[
    'cart.carts', 'cart.cart_items', 'cart.checkout_line_items',
    'customer.wishlist_items', 'admin.wishlist_items',
    'admin.user_sessions', 'customer.customer_sessions',
    'reviews.reviews'
  ];
BEGIN
  FOREACH t IN ARRAY tier2 LOOP
    PERFORM audit.enroll_table(split_part(t, '.', 1), split_part(t, '.', 2), 2::SMALLINT);
  END LOOP;
END $$;

-- Tier 3 — recorded as deliberately never audited. Auditing the audit tables
-- would recurse; the rest are high-churn derived data with no forensic value.
INSERT INTO audit.audited_tables ("schemaName", "tableName", "tier", "enabled", "note")
VALUES
  ('audit', 'change_log',     3, FALSE, 'the audit table itself — would recurse'),
  ('audit', 'change_batch',   3, FALSE, 'the audit table itself — would recurse'),
  ('audit', 'interface_log',  3, FALSE, 'diagnostics stream, not business data'),
  ('audit', 'activity_log',   3, FALSE, 'analytics stream, highest volume'),
  ('audit', 'audited_tables', 3, FALSE, 'audit configuration')
ON CONFLICT ("schemaName", "tableName") DO NOTHING;

COMMIT;