-- 123-data-quality-monitoring.sql
-- Data-quality monitoring layer (read-only) from the TIP data-quality audit 2026-07-05.
--
-- NON-DESTRUCTIVE: creates four SELECT-only views. No table, column, row or
-- constraint changes. Safe to apply on a live DB and idempotent (re-runnable).
--
--   v_scraper_health      freshness + breadth per (source vendor, observation table)
--   v_stock_completeness  % of stock fields actually extracted, per source vendor
--   v_coverage            catalog composition + reach_label completeness per manufacturer
--   v_data_quality_alerts one row per OPEN data-quality issue -> the task list
--                         (goal: this view returns zero rows once everything is fixed)
--
-- Freshness thresholds (calibrated against observed scraper cadence 2026-07-06):
--   fresh  = newest row within  7 days
--   stale  = newest row within 30 days
--   dead   = newest row older than 30 days  (regression: had data, then stopped)

-- Drop in reverse-dependency order so the file is fully re-runnable.
DROP VIEW IF EXISTS v_data_quality_alerts;
DROP VIEW IF EXISTS v_scraper_health;
DROP VIEW IF EXISTS v_stock_completeness;
DROP VIEW IF EXISTS v_coverage;

-- ---------------------------------------------------------------------------
-- v_scraper_health: is each scraper still delivering, and how broad is it?
-- One row per (source vendor, observation table). Separating the tables is
-- essential: e.g. ATGBICS price is fresh while ATGBICS stock is dead.
-- ---------------------------------------------------------------------------
CREATE VIEW v_scraper_health AS
WITH obs AS (
  SELECT source_vendor_id, 'stock_observations'::text AS source_table,
         "time", transceiver_id
  FROM stock_observations
  UNION ALL
  SELECT source_vendor_id, 'price_observations'::text AS source_table,
         "time", transceiver_id
  FROM price_observations
)
SELECT v.name  AS vendor,
       v.slug  AS vendor_slug,
       o.source_table,
       count(*)                                                          AS total_rows,
       count(*) FILTER (WHERE o."time" >= now() - interval '30 days')    AS rows_30d,
       count(DISTINCT o.transceiver_id)                                  AS distinct_transceivers,
       max(o."time")::date                                              AS last_observation,
       (now()::date - max(o."time")::date)                              AS days_since_last,
       CASE
         WHEN max(o."time") >= now() - interval '7 days'  THEN 'fresh'
         WHEN max(o."time") >= now() - interval '30 days' THEN 'stale'
         ELSE 'dead'
       END                                                              AS status
FROM obs o
JOIN vendors v ON v.id = o.source_vendor_id
GROUP BY v.id, v.name, v.slug, o.source_table;

-- ---------------------------------------------------------------------------
-- v_stock_completeness: rows are being written, but are the fields populated?
-- Catches "scraper runs but extraction is a no-op" (e.g. QSFPTEK: 75k rows,
-- 0% units_sold / warehouse / delivery).
-- ---------------------------------------------------------------------------
CREATE VIEW v_stock_completeness AS
SELECT v.name AS vendor,
       v.slug AS vendor_slug,
       count(*)              AS stock_rows,
       max(s."time")::date  AS last_obs,
       round(100.0 * count(s.units_sold)     / count(*), 1) AS pct_units_sold,
       round(100.0 * count(*) FILTER (
               WHERE s.warehouse_de_qty IS NOT NULL
                  OR s.warehouse_global_qty IS NOT NULL)    / count(*), 1) AS pct_warehouse_qty,
       round(100.0 * count(*) FILTER (
               WHERE s.warehouse_de_delivery_date IS NOT NULL
                  OR s.warehouse_global_delivery_date IS NOT NULL
                  OR s.backorder_estimated_date IS NOT NULL) / count(*), 1) AS pct_delivery_date,
       round(100.0 * count(s.lead_time_days) / count(*), 1) AS pct_lead_time
FROM stock_observations s
JOIN vendors v ON v.id = s.source_vendor_id
GROUP BY v.id, v.name, v.slug;

-- ---------------------------------------------------------------------------
-- v_coverage: catalog composition and reach_label completeness per manufacturer
-- (transceivers.vendor_id = the brand, distinct from the scraping source vendor).
-- ---------------------------------------------------------------------------
CREATE VIEW v_coverage AS
SELECT v.name AS vendor,
       v.slug AS vendor_slug,
       count(*)                                                    AS transceivers,
       count(*) FILTER (WHERE t.product_type = 'module')           AS modules,
       count(*) FILTER (WHERE t.product_type IN ('dac','aoc','aec')) AS cables,
       count(*) FILTER (WHERE t.reach_label IS NULL)               AS reach_label_null,
       round(100.0 * count(*) FILTER (WHERE t.reach_label IS NULL) / count(*), 1) AS pct_reach_null
FROM transceivers t
JOIN vendors v ON v.id = t.vendor_id
GROUP BY v.id, v.name, v.slug;

-- ---------------------------------------------------------------------------
-- v_data_quality_alerts: the actionable task list. Each rule fires only while
-- the underlying problem is open, so the view drains to empty as things get
-- fixed. severity_rank orders critical(1) > high(2) > warning(3).
-- ---------------------------------------------------------------------------
CREATE VIEW v_data_quality_alerts AS
WITH alerts AS (
  -- 1) Dead scraper: had data, stopped delivering > 30 days ago.
  SELECT 'critical'::text AS severity, 1 AS severity_rank,
         'scraper_dead'::text AS category,
         vendor || ' / ' || source_table AS entity,
         format('no new rows for %s days (last %s); %s rows total',
                days_since_last, last_observation, total_rows) AS detail
  FROM v_scraper_health
  WHERE status = 'dead'

  UNION ALL
  -- 2) Stale scraper: 8-30 days since last row.
  SELECT 'warning', 3, 'scraper_stale',
         vendor || ' / ' || source_table,
         format('no new rows for %s days (last %s)', days_since_last, last_observation)
  FROM v_scraper_health
  WHERE status = 'stale'

  UNION ALL
  -- 3) Stock rows written but extraction is empty (scraper otherwise alive).
  SELECT 'high', 2, 'stock_extraction_empty',
         vendor || ' / stock_observations',
         format('%s rows but units_sold=%s%%, warehouse=%s%%, delivery=%s%% populated',
                stock_rows, pct_units_sold, pct_warehouse_qty, pct_delivery_date)
  FROM v_stock_completeness
  WHERE stock_rows >= 500
    AND pct_units_sold = 0
    AND pct_warehouse_qty = 0
    AND last_obs >= now()::date - 30

  UNION ALL
  -- 4) Fleet-wide delivery-date / lead-time gap -> lead_time_days not derivable.
  SELECT 'high', 2, 'delivery_date_gap',
         'ALL stock scrapers',
         format('delivery_date on %s%% of stock rows, lead_time on %s%% -> lead_time not derivable',
                round(100.0 * count(*) FILTER (
                    WHERE warehouse_de_delivery_date IS NOT NULL
                       OR warehouse_global_delivery_date IS NOT NULL) / nullif(count(*),0), 2),
                round(100.0 * count(lead_time_days) / nullif(count(*),0), 2))
  FROM stock_observations
  HAVING round(100.0 * count(*) FILTER (
             WHERE warehouse_de_delivery_date IS NOT NULL
                OR warehouse_global_delivery_date IS NOT NULL) / nullif(count(*),0), 2) < 5

  UNION ALL
  -- 5) reach_label incomplete on a sizeable catalog.
  SELECT 'warning', 3, 'reach_label_incomplete',
         vendor,
         format('%s%% of %s transceivers have NULL reach_label', pct_reach_null, transceivers)
  FROM v_coverage
  WHERE transceivers >= 100 AND pct_reach_null >= 20

  UNION ALL
  -- 6) Future-dated news -> ingest guard candidate.
  SELECT 'warning', 3, 'future_dated_news',
         'news_articles',
         format('%s article(s) dated in the future (max %s) — ingest guard recommended',
                count(*), max(published_at)::date)
  FROM news_articles
  WHERE published_at > now() + interval '2 days'
  HAVING count(*) > 0
)
SELECT severity, severity_rank, category, entity, detail
FROM alerts
ORDER BY severity_rank, category, entity;