-- 009_kafka_lag.sql
-- Kafka consumer-group lag history for the OpenBMP ingestion path, written by
-- the kafka-lag-monitor service every ~30s. Backs the Kafka Lag dashboard so
-- the ingestion path can be sanity-checked: watch lag spike during a BGP
-- convergence storm and drain again, and confirm the consumer member count
-- when psql-app is scaled out.

CREATE TABLE IF NOT EXISTS kafka_consumer_lag (
    ts         timestamptz  NOT NULL DEFAULT now(),
    group_id   varchar(128) NOT NULL,
    topic      varchar(200) NOT NULL,
    partition  integer      NOT NULL,
    committed  bigint,
    log_end    bigint,
    lag        bigint
);

CREATE TABLE IF NOT EXISTS kafka_consumer_members (
    ts         timestamptz  NOT NULL DEFAULT now(),
    group_id   varchar(128) NOT NULL,
    members    integer
);

SELECT create_hypertable('kafka_consumer_lag',     'ts', if_not_exists => TRUE);
SELECT create_hypertable('kafka_consumer_members', 'ts', if_not_exists => TRUE);

CREATE INDEX IF NOT EXISTS idx_kafka_lag_group_ts
    ON kafka_consumer_lag (group_id, ts DESC);
CREATE INDEX IF NOT EXISTS idx_kafka_members_group_ts
    ON kafka_consumer_members (group_id, ts DESC);

-- Bound history growth (~250k rows/day across ~88 partitions at 30s). Wrapped
-- so the script still succeeds if the TimescaleDB job scheduler is unavailable.
DO $$ BEGIN
  PERFORM add_retention_policy('kafka_consumer_lag',     INTERVAL '14 days');
  PERFORM add_retention_policy('kafka_consumer_members', INTERVAL '14 days');
EXCEPTION WHEN OTHERS THEN
  RAISE NOTICE 'kafka lag retention policy not added: %', SQLERRM;
END $$;