broad-announce/migrations/clickhouse_schema.sql

-- 007_clickhouse.up.sql
-- M7: ClickHouse schema for the deliveries archive.
-- M8: + deliveries_dlq_archive (DLQ) with 2y TTL.
--
-- ClickHouse is columnar and prefers wide tables with
-- per-column compression. The schema mirrors the Postgres
-- `deliveries` table one-to-one for v1; a future M7.5 can
-- strip unused fields (e.g. drop the JSONB payload blob
-- if we never query into it in CH).
--
-- The DDL below is designed to be runnable from the
-- archiverd's first-run setup hook. It's also what the
-- M7 smoke executes by hand via curl http://clickhouse:8123/.
--
-- Note on types:
--   - `id` is BIGINT in Postgres (BIGSERIAL). In CH we
--     keep BIGINT — CH's UInt64 is the closest match,
--     but for an archive that doesn't reference id
--     anywhere else, BIGINT is fine.
--   - `payload` is JSONB in Postgres. We store it as a
--     CH String. The Go side marshals to JSON text
--     before INSERT. CH's JSON type is still experimental
--     in 24.x; String is the safe choice.
--   - `status` is a small enum ('pending','sent','failed','dlq').
--     In CH we use LowCardinality(String) — the column
--     takes ~0.5 bytes per row for the dictionary.

CREATE DATABASE IF NOT EXISTS ba_archive;

-- ── M7: live deliveries archive ─────────────────────────────────
-- 1-year retention. Rows that age out of the Timescale
-- 7d hot window land here; after 1y CH drops the
-- partition. Operators can override via
--   ALTER TABLE ba_archive.deliveries_archive MODIFY TTL ...
CREATE TABLE IF NOT EXISTS ba_archive.deliveries_archive (
    id              BIGINT,
    alert_id        String,
    company_id      String,
    individual_id   String,
    channel         LowCardinality(String),
    target          String,
    status          LowCardinality(String),
    attempts        UInt32,
    last_error      String,
    payload         String,  -- raw JSON
    created_at      DateTime64(3, 'UTC'),
    sent_at         Nullable(DateTime64(3, 'UTC')),
    next_attempt_at Nullable(DateTime64(3, 'UTC')),
    -- The source postgres row's last-update marker. We
    -- set this to the wall-clock time at insert. Useful
    -- for "show me everything archived in the last 24h".
    archived_at     DateTime DEFAULT now()
) ENGINE = MergeTree
  PARTITION BY toYYYYMM(created_at)
  ORDER BY (company_id, created_at, id)
  TTL toDateTime(created_at) + INTERVAL 365 DAY;

-- A MATERIALIZED VIEW that aggregates per-company,
-- per-day delivery counts. Used by M9 dashboards.
CREATE MATERIALIZED VIEW IF NOT EXISTS
    ba_archive.deliveries_per_company_daily_mv
ENGINE = SummingMergeTree
  PARTITION BY toYYYYMM(day)
  ORDER BY (company_id, day, channel, status)
AS
SELECT
    company_id,
    toDate(created_at) AS day,
    channel,
    status,
    count()           AS n
FROM ba_archive.deliveries_archive
GROUP BY company_id, day, channel, status;

-- ── M8: DLQ archive ────────────────────────────────────────────
-- 2-year retention (vs 1y for live). DLQ entries are
-- forensic data — operators want them around longer
-- when triaging a regression. Mirrors the live
-- deliveries_archive column shape minus the retry-
-- scheduling columns (sent_at, next_attempt_at) which
-- the DLQ doesn't carry.
CREATE TABLE IF NOT EXISTS ba_archive.deliveries_dlq_archive (
    id              BIGINT,
    alert_id        String,
    company_id      String,
    individual_id   String,
    channel         LowCardinality(String),
    target          String,
    status          LowCardinality(String),
    attempts        UInt32,
    last_error      String,
    payload         String,  -- raw JSON, the original NATS envelope
    created_at      DateTime64(3, 'UTC'),
    archived_at     DateTime DEFAULT now()
) ENGINE = MergeTree
  PARTITION BY toYYYYMM(created_at)
  ORDER BY (company_id, created_at, id)
  TTL toDateTime(created_at) + INTERVAL 730 DAY;

-- Per-company daily DLQ counts. Used by the M9
-- observability stack to surface DLQ rate per tenant.
CREATE MATERIALIZED VIEW IF NOT EXISTS
    ba_archive.deliveries_dlq_per_company_daily_mv
ENGINE = SummingMergeTree
  PARTITION BY toYYYYMM(day)
  ORDER BY (company_id, day, channel)
AS
SELECT
    company_id,
    toDate(created_at) AS day,
    channel,
    count()           AS n
FROM ba_archive.deliveries_dlq_archive
GROUP BY company_id, day, channel;