Data flow

-- Trigger applied to both channel_audit and payments_audit
CREATE OR REPLACE FUNCTION audit_immutable_trigger()
RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'audit tables are append-only; % is not allowed', TG_OP;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER prevent_audit_mutation
BEFORE UPDATE OR DELETE ON channel_audit
FOR EACH ROW EXECUTE FUNCTION audit_immutable_trigger();

CREATE TRIGGER prevent_audit_mutation
BEFORE UPDATE OR DELETE ON payments_audit
FOR EACH ROW EXECUTE FUNCTION audit_immutable_trigger();

CREATE TABLE channel_audit (
  id           ULID PRIMARY KEY,
  user_id      ULID,
  actor_role   TEXT,              -- JOB_SEEKER, TRADER, EMPLOYER, LENDER, SYSTEM
  channel      TEXT,              -- VOICE, WHATSAPP_TEXT, WHATSAPP_CALL, SMS, WEB, API
  event_kind   TEXT,              -- the specific event taxonomy
  event_data   JSONB,             -- channel-specific payload
  request_id   ULID,              -- propagated request ID
  occurred_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (id),
  -- partitioned by month for retention scaling
);

CREATE TABLE payments_audit (
  id              ULID PRIMARY KEY,
  user_id         ULID NOT NULL,
  actor_role      TEXT NOT NULL,
  motion_kind     TEXT NOT NULL,       -- WALLET_CREDIT, WALLET_DEBIT, ESCROW_HOLD, etc.
  amount_kobo     BIGINT NOT NULL,     -- always kobo, never naira
  direction       TEXT NOT NULL,       -- IN or OUT
  source          TEXT,                -- WALLET, MANDATE, VA_INFLOW, PAYOUT, SAVINGS, BILL, ESCROW
  counterparty    TEXT,                -- Job Seeker / Employer / external bank / OPay
  external_ref    TEXT,                -- OPay transaction reference (when applicable)
  request_id      ULID,
  verified_at     TIMESTAMPTZ,         -- when the reconcile guard verified (NULL if not from OPay)
  occurred_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (id)
);

SELECT
  u.id as user_id,
  -- Volume features
  COUNT(DISTINCT a.id) FILTER (WHERE a.status = 'COMPLETED' AND a.completed_at > now() - interval '90 days') as completed_90d,
  AVG(r.rating) FILTER (WHERE r.created_at > now() - interval '90 days') as avg_review_90d,
  -- Cashflow features
  SUM(p.amount_kobo) FILTER (WHERE p.direction = 'IN' AND p.occurred_at > now() - interval '90 days') as inflow_90d_kobo,
  COUNT(DISTINCT p.counterparty) FILTER (WHERE p.direction = 'IN' AND p.occurred_at > now() - interval '90 days') as senders_90d,
  -- Channel diversity
  COUNT(DISTINCT ca.channel) FILTER (WHERE ca.occurred_at > now() - interval '30 days') as channels_30d,
  -- Tenure
  EXTRACT(EPOCH FROM (now() - u.created_at)) / 86400 as tenure_days
FROM identity.users u
LEFT JOIN jobs.applications a ON a.user_id = u.id
LEFT JOIN jobs.reviews r ON r.reviewed_user_id = u.id
LEFT JOIN payments_audit p ON p.user_id = u.id
LEFT JOIN channel_audit ca ON ca.user_id = u.id
WHERE u.id = $1
GROUP BY u.id;

-- The partner-api-snapshot view (pseudocode)
SELECT
  u.id,
  u.actor_role,
  u.standing_tier,
  u.standing_score,
  -- Aggregated cashflow over the consent's ledger window
  SUM(p.amount_kobo) FILTER (WHERE p.direction = 'IN') as total_inflows_kobo,
  SUM(p.amount_kobo) FILTER (WHERE p.direction = 'OUT') as total_outflows_kobo,
  COUNT(*) as transaction_count,
  COUNT(DISTINCT p.counterparty) as unique_counterparties,
  -- KYC summary
  k.tier as kyc_tier,
  k.bvn_verified,
  k.nin_verified,
  k.address_verified
FROM identity.users u
LEFT JOIN payments_audit p ON p.user_id = u.id
  AND p.occurred_at > now() - interval '90 days'  -- consent's ledger window
LEFT JOIN identity.kyc_records k ON k.user_id = u.id
WHERE u.id = $1
GROUP BY u.id, u.actor_role, u.standing_tier, u.standing_score, k.tier, k.bvn_verified, k.nin_verified, k.address_verified;