feat: Add historical time slider with live/historical mode switching and Grafana sync

database/init.sql

@@ -278,19 +278,22 @@ CREATE INDEX IF NOT EXISTS idx_cml_data_covering
 -- requested time range exceeds 3 days, reducing the scanned row count
 -- by ~360x (10-second raw data → 1-hour buckets).
 -- ---------------------------------------------------------------------------
+-- record_count (migration 014) is used by materialize_cml_stats_snapshot()
+-- below to compute completeness percentages for the historical time slider.
 CREATE MATERIALIZED VIEW cml_data_1h
 WITH (timescaledb.continuous) AS
 SELECT
     time_bucket('1 hour', time) AS bucket,
     user_id,
     cml_id,
     sublink_id,
-    MIN(rsl)  AS rsl_min,
-    MAX(rsl)  AS rsl_max,
-    AVG(rsl)  AS rsl_avg,
-    MIN(tsl)  AS tsl_min,
-    MAX(tsl)  AS tsl_max,
-    AVG(tsl)  AS tsl_avg
+    MIN(rsl)     AS rsl_min,
+    MAX(rsl)     AS rsl_max,
+    AVG(rsl)     AS rsl_avg,
+    MIN(tsl)     AS tsl_min,
+    MAX(tsl)     AS tsl_max,
+    AVG(tsl)     AS tsl_avg,
+    COUNT(*)     AS record_count
 FROM cml_data
 GROUP BY bucket, user_id, cml_id, sublink_id
 WITH NO DATA;
@@ -469,4 +472,180 @@ CREATE POLICY user_isolation ON file_processing_log
 GRANT SELECT, INSERT ON file_processing_log TO demo_openmrg, demo_orange_cameroun;
 GRANT SELECT            ON file_processing_log TO webserver_role;
 GRANT USAGE ON SEQUENCE file_processing_log_id_seq TO demo_openmrg, demo_orange_cameroun;
+
+-- ============================================================
+-- Historical CML stats snapshots (migration 015)
+-- Stores hourly snapshots of cml_stats for the historical time slider
+-- on the realtime map. Supports provisional (current partial hour) and
+-- definitive (completed hours) rows.
+-- ============================================================
+CREATE TABLE IF NOT EXISTS cml_stats_history (
+    snapshot_time             TIMESTAMPTZ NOT NULL,
+    cml_id                    TEXT        NOT NULL,
+    user_id                   TEXT        NOT NULL,
+    completeness_percent_6h   REAL,
+    total_records_6h          BIGINT,
+    valid_records_6h          BIGINT,
+    mean_rsl_6h               REAL,
+    stddev_rsl_6h             REAL,
+    completeness_percent_1h   REAL,
+    mean_rsl_1h               REAL,
+    stddev_rsl_1h             REAL,
+    last_rsl                  REAL,
+    is_provisional            BOOLEAN NOT NULL DEFAULT FALSE,
+    PRIMARY KEY (snapshot_time, cml_id, user_id)
+);
+
+-- Create hypertable with 7-day chunks
+SELECT create_hypertable(
+    'cml_stats_history', 'snapshot_time',
+    chunk_time_interval => INTERVAL '7 days',
+    if_not_exists => TRUE
+);
+
+-- Enable compression for old snapshots (>7 days)
+ALTER TABLE cml_stats_history
+    SET (timescaledb.compress,
+         timescaledb.compress_segmentby = 'user_id, cml_id',
+         timescaledb.compress_orderby   = 'snapshot_time DESC');
+
+SELECT add_compression_policy('cml_stats_history', INTERVAL '7 days');
+
+-- Materialization function: writes definitive snapshot from cml_data_1h
+CREATE OR REPLACE FUNCTION materialize_cml_stats_snapshot(
+    p_at_time   TIMESTAMPTZ,   -- must be hour-truncated by caller
+    p_user_id   TEXT
+) RETURNS INT                  -- number of rows upserted
+LANGUAGE plpgsql SECURITY DEFINER
+SET search_path = public
+AS $$
+DECLARE
+    v_rows INT;
+BEGIN
+    -- Validate caller is a known user
+    IF NOT EXISTS (SELECT 1 FROM cml_metadata WHERE user_id = p_user_id LIMIT 1) THEN
+        RAISE EXCEPTION 'Unknown user: %', p_user_id;
+    END IF;
+
+    INSERT INTO cml_stats_history (
+        snapshot_time, cml_id, user_id,
+        completeness_percent_6h, total_records_6h, valid_records_6h,
+        mean_rsl_6h, stddev_rsl_6h,
+        completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h,
+        last_rsl, is_provisional
+    )
+    SELECT
+        p_at_time,
+        m.cml_id,
+        p_user_id,
+        -- 6-hour window: sum the 6 hourly buckets ending at p_at_time
+        -- (h6 already aggregates across those buckets internally, so no
+        -- further SUM/AVG or GROUP BY is needed at this level)
+        ROUND(
+            100.0 * h6.rsl_count::numeric
+                  / NULLIF(h6.record_count, 0),
+            2) AS completeness_percent_6h,
+        h6.record_count                          AS total_records_6h,
+        h6.rsl_count                              AS valid_records_6h,
+        ROUND(h6.rsl_avg::numeric, 2)            AS mean_rsl_6h,
+        ROUND(h6.rsl_stddev::numeric, 2)         AS stddev_rsl_6h,
+        -- 1-hour window: the single bucket containing p_at_time
+        ROUND(
+            100.0 * h1.rsl_count::numeric
+                  / NULLIF(h1.record_count, 0),
+            2) AS completeness_percent_1h,
+        ROUND(h1.rsl_avg::numeric, 2)            AS mean_rsl_1h,
+        ROUND(h1.rsl_stddev::numeric, 2)         AS stddev_rsl_1h,
+        h1.rsl_max                               AS last_rsl,
+        FALSE                                    AS is_provisional
+    FROM (SELECT DISTINCT ON (cml_id) * FROM cml_metadata WHERE user_id = p_user_id) m
+    -- 6h subquery from the 1h continuous aggregate
+    LEFT JOIN LATERAL (
+        SELECT
+            SUM(record_count)                                                              AS record_count,
+            SUM(CASE WHEN rsl_max IS NOT NULL AND rsl_max::text != 'NaN' THEN record_count ELSE 0 END) AS rsl_count,
+            AVG(NULLIF(rsl_avg, 'NaN'::float))                                             AS rsl_avg,
+            AVG(NULLIF(rsl_max, 'NaN'::float) - NULLIF(rsl_min, 'NaN'::float))            AS rsl_stddev
+        FROM cml_data_1h
+        WHERE user_id    = p_user_id
+          AND cml_id     = m.cml_id
+          AND bucket     >= p_at_time - INTERVAL '6 hours'
+          AND bucket     <  p_at_time
+    ) h6 ON TRUE
+    -- 1h subquery: aggregate across all sublinks for the bucket immediately before p_at_time
+    LEFT JOIN LATERAL (
+        SELECT
+            SUM(record_count)                                                              AS record_count,
+            SUM(CASE WHEN rsl_max IS NOT NULL AND rsl_max::text != 'NaN' THEN record_count ELSE 0 END) AS rsl_count,
+            AVG(NULLIF(rsl_avg, 'NaN'::float))                                             AS rsl_avg,
+            MAX(NULLIF(rsl_max, 'NaN'::float))                                             AS rsl_max,
+            MIN(NULLIF(rsl_min, 'NaN'::float))                                             AS rsl_min,
+            AVG(NULLIF(rsl_max, 'NaN'::float) - NULLIF(rsl_min, 'NaN'::float))            AS rsl_stddev
+        FROM cml_data_1h
+        WHERE user_id = p_user_id
+          AND cml_id  = m.cml_id
+          AND bucket  = p_at_time - INTERVAL '1 hour'
+    ) h1 ON TRUE
+    WHERE m.user_id = p_user_id
+    ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET
+        completeness_percent_6h = EXCLUDED.completeness_percent_6h,
+        total_records_6h        = EXCLUDED.total_records_6h,
+        valid_records_6h        = EXCLUDED.valid_records_6h,
+        mean_rsl_6h             = EXCLUDED.mean_rsl_6h,
+        stddev_rsl_6h           = EXCLUDED.stddev_rsl_6h,
+        completeness_percent_1h = EXCLUDED.completeness_percent_1h,
+        mean_rsl_1h             = EXCLUDED.mean_rsl_1h,
+        stddev_rsl_1h           = EXCLUDED.stddev_rsl_1h,
+        last_rsl                = EXCLUDED.last_rsl,
+        is_provisional          = FALSE;
+
+    GET DIAGNOSTICS v_rows = ROW_COUNT;
+    RETURN v_rows;
+END;
+$$;
+
+-- Read function for webserver / Grafana
+CREATE OR REPLACE FUNCTION get_cml_stats_at(
+    p_at_time   TIMESTAMPTZ,
+    p_user_id   TEXT
+) RETURNS TABLE (
+    cml_id                   TEXT,
+    completeness_percent_6h  REAL,
+    total_records_6h         BIGINT,
+    valid_records_6h         BIGINT,
+    mean_rsl_6h              REAL,
+    stddev_rsl_6h            REAL,
+    completeness_percent_1h  REAL,
+    mean_rsl_1h              REAL,
+    stddev_rsl_1h            REAL,
+    last_rsl                 REAL,
+    is_provisional           BOOLEAN
+) LANGUAGE sql STABLE SECURITY DEFINER
+SET search_path = public
+AS $$
+    SELECT
+        h.cml_id,
+        h.completeness_percent_6h,
+        h.total_records_6h,
+        h.valid_records_6h,
+        h.mean_rsl_6h,
+        h.stddev_rsl_6h,
+        h.completeness_percent_1h,
+        h.mean_rsl_1h,
+        h.stddev_rsl_1h,
+        h.last_rsl,
+        h.is_provisional
+    FROM cml_stats_history h
+    WHERE h.user_id       = p_user_id
+      AND h.snapshot_time = date_trunc('hour', p_at_time)
+    ORDER BY h.cml_id;
+$$;
+
+GRANT SELECT, INSERT, UPDATE ON cml_stats_history TO demo_openmrg, demo_orange_cameroun;
+GRANT EXECUTE ON FUNCTION materialize_cml_stats_snapshot(TIMESTAMPTZ, TEXT)
+    TO demo_openmrg, demo_orange_cameroun;
+
+GRANT SELECT ON cml_stats_history TO webserver_role;
+GRANT EXECUTE ON FUNCTION get_cml_stats_at(TIMESTAMPTZ, TEXT)
+    TO webserver_role;
 GRANT SELECT ON cml_data_1h_secure TO webserver_role;

database/migrations/014_add_record_count_to_cml_data_1h.sql

@@ -0,0 +1,68 @@
+-- Migration 014: Add record_count to cml_data_1h continuous aggregate
+--
+-- Prerequisite for the historical time slider (see the following
+-- cml_stats_history migration). materialize_cml_stats_snapshot() computes
+-- completeness_percent_6h / completeness_percent_1h from the number of raw
+-- samples in each hourly bucket, which requires a record_count column that
+-- cml_data_1h has never had (it was only ever needed for Grafana's raw
+-- RSL/TSL line charts, which use MIN/MAX/AVG, not counts).
+--
+-- Continuous aggregates cannot have a column added via ALTER; the view must
+-- be dropped and recreated. This is non-destructive: TimescaleDB
+-- re-materialises from the underlying raw cml_data, matching the pattern
+-- already used in migration 003_update_aggregate_user_id.sql.
+--
+-- Dropping cml_data_1h CASCADEs to its continuous aggregate policy and to
+-- the cml_data_1h_secure security-barrier view (and that view's grants), so
+-- all three must be recreated here.
+--
+-- A brief gap in Grafana's hourly-aggregate data is expected while the
+-- refresh policy backfills the view (~1 refresh cycle, up to 1 hour).
+-- Queries that fall in the gap automatically fall through to raw cml_data.
+--
+-- Apply with:
+--   docker compose exec -T database psql -U myuser -d mydatabase \
+--     < database/migrations/014_add_record_count_to_cml_data_1h.sql
+
+-- Step 1: Remove the old view and its dependent policy, secure view, and grants.
+DROP MATERIALIZED VIEW IF EXISTS cml_data_1h CASCADE;
+
+-- Step 2: Recreate with record_count added.
+CREATE MATERIALIZED VIEW cml_data_1h
+WITH (timescaledb.continuous) AS
+SELECT
+    time_bucket('1 hour', time) AS bucket,
+    user_id,
+    cml_id,
+    sublink_id,
+    MIN(rsl)     AS rsl_min,
+    MAX(rsl)     AS rsl_max,
+    AVG(rsl)     AS rsl_avg,
+    MIN(tsl)     AS tsl_min,
+    MAX(tsl)     AS tsl_max,
+    AVG(tsl)     AS tsl_avg,
+    COUNT(*)     AS record_count
+FROM cml_data
+GROUP BY bucket, user_id, cml_id, sublink_id
+WITH NO DATA;
+
+-- Step 3: Restore the refresh policy (same parameters as before).
+SELECT add_continuous_aggregate_policy('cml_data_1h',
+    start_offset      => INTERVAL '2 days',
+    end_offset        => INTERVAL '1 hour',
+    schedule_interval => INTERVAL '1 hour'
+);
+
+-- Step 4: Recreate the security-barrier view dropped by CASCADE above.
+CREATE VIEW cml_data_1h_secure WITH (security_barrier) AS
+SELECT * FROM cml_data_1h
+WHERE user_id = current_user;
+
+-- Step 5: Restore grants dropped by CASCADE above.
+GRANT SELECT ON cml_data_1h_secure TO demo_openmrg, demo_orange_cameroun;
+GRANT SELECT ON cml_data_1h        TO webserver_role;
+GRANT SELECT ON cml_data_1h_secure TO webserver_role;
+
+-- Step 6: Optional — trigger an immediate backfill rather than waiting for
+--         the next scheduled refresh.  Remove the leading '--' to enable.
+-- CALL refresh_continuous_aggregate('cml_data_1h', NULL, NULL);