-- Canonical production re-derivation for
-- /research/nursing-home-stars-vs-actual-harm.
--
-- Execute this one WITH statement through
-- scripts/research/study-figure-audit/sql.ts::executeReadOnly. That helper
-- pins Supabase project rfabugntpwelbpbexgcz and wraps the statement in:
--   BEGIN TRANSACTION READ ONLY;
--   SET LOCAL statement_timeout = '120s';
--   ...this statement...
--   ROLLBACK;
--
-- The three-year period is anchored to the maximum CMS processing date, then
-- applied to survey_date. This distinction matters: the source file contains
-- three inspection cycles, which span more than three literal years.
WITH source_versions AS MATERIALIZED (
  SELECT
    ps.id,
    ps.source_name,
    ps.source_url,
    ps.source_publisher,
    ps.retrieved_at,
    ps.upstream_modified_at,
    ps.content_sha256,
    substring(
      ps.source_name FROM '([0-9]{4}-[0-9]{2}-[0-9]{2})$'
    )::date AS source_release_date
  FROM public.provenance_sources AS ps
  WHERE ps.source_name LIKE 'CMS Nursing Home Provider Information %'
    AND EXISTS (
      SELECT 1
      FROM public.provenance_claims AS pc
      WHERE pc.source_id = ps.id
        AND pc.subject_type = 'facility'
        AND pc.metric = 'nursing_overall_rating'
        AND pc.superseded_at IS NULL
    )
),
latest_period AS (
  SELECT
    max(
      to_char(source_release_date, 'YYYY') || 'q' ||
      extract(quarter FROM source_release_date)::integer
    ) AS claim_period
  FROM source_versions
),
period_sources AS MATERIALIZED (
  SELECT sv.*
  FROM source_versions AS sv
  CROSS JOIN latest_period AS lp
  WHERE to_char(sv.source_release_date, 'YYYY') || 'q' ||
        extract(quarter FROM sv.source_release_date)::integer = lp.claim_period
),
rating_claims AS MATERIALIZED (
  SELECT pc.*
  FROM public.provenance_claims AS pc
  JOIN period_sources AS ps ON ps.id = pc.source_id
  CROSS JOIN latest_period AS lp
  WHERE pc.subject_type = 'facility'
    AND pc.superseded_at IS NULL
    AND right(pc.claim_key, length(lp.claim_period) + 1) = '.' || lp.claim_period
),
rating_subjects AS MATERIALIZED (
  SELECT DISTINCT subject_id AS ccn
  FROM rating_claims
),
rating_pivot AS MATERIALIZED (
  SELECT
    subject_id AS ccn,
    max(value_numeric) FILTER (
      WHERE metric = 'nursing_overall_rating'
    )::integer AS overall_rating,
    max(value_numeric) FILTER (
      WHERE metric = 'nursing_health_inspection_rating'
    )::integer AS health_inspection_rating,
    max(value_numeric) FILTER (
      WHERE metric = 'nursing_staffing_rating'
    )::integer AS staffing_rating,
    max(value_numeric) FILTER (
      WHERE metric = 'nursing_quality_measure_rating'
    )::integer AS qm_rating
  FROM rating_claims
  WHERE metric IN (
    'nursing_overall_rating',
    'nursing_health_inspection_rating',
    'nursing_staffing_rating',
    'nursing_quality_measure_rating'
  )
  GROUP BY subject_id
),
ratings AS MATERIALIZED (
  SELECT
    rs.ccn,
    rp.overall_rating,
    rp.health_inspection_rating,
    rp.staffing_rating,
    rp.qm_rating
  FROM rating_subjects AS rs
  LEFT JOIN rating_pivot AS rp USING (ccn)
),
window_parameters AS MATERIALIZED (
  SELECT
    max(processing_date)::date AS window_end,
    (max(processing_date) - interval '3 years')::date AS window_start
  FROM public.nh_health_deficiencies
),
window_deficiencies AS MATERIALIZED (
  SELECT d.*
  FROM public.nh_health_deficiencies AS d
  CROSS JOIN window_parameters AS wp
  WHERE d.survey_date BETWEEN wp.window_start AND wp.window_end
),
g_plus_rows AS MATERIALIZED (
  SELECT *
  FROM window_deficiencies
  WHERE scope_severity_code IN ('G', 'H', 'I', 'J', 'K', 'L')
),
harm_by_ccn AS MATERIALIZED (
  SELECT
    ccn,
    count(*)::bigint AS g_plus_count,
    count(*) FILTER (
      WHERE coalesce(citation_under_idr, 'N') <> 'Y'
    )::bigint AS non_idr_g_plus_count,
    max(survey_date)::date AS latest_g_plus_survey_date,
    max(scope_severity_code) AS worst_severity
  FROM g_plus_rows
  GROUP BY ccn
),
latest_facility_identity AS MATERIALIZED (
  SELECT DISTINCT ON (ccn)
    ccn,
    provider_name,
    city_town,
    state
  FROM public.nh_health_deficiencies
  ORDER BY
    ccn,
    processing_date DESC NULLS LAST,
    survey_date DESC NULLS LAST,
    id DESC
),
population AS MATERIALIZED (
  SELECT
    r.*,
    coalesce(h.g_plus_count, 0)::bigint AS g_plus_count,
    coalesce(h.non_idr_g_plus_count, 0)::bigint AS non_idr_g_plus_count,
    h.latest_g_plus_survey_date,
    h.worst_severity,
    i.provider_name,
    i.city_town,
    i.state
  FROM ratings AS r
  LEFT JOIN harm_by_ccn AS h USING (ccn)
  LEFT JOIN latest_facility_identity AS i USING (ccn)
),
star_bands AS (
  SELECT
    stars.star,
    count(p.*) FILTER (WHERE p.overall_rating = stars.star)::bigint AS total_count,
    count(p.*) FILTER (
      WHERE p.overall_rating = stars.star AND p.g_plus_count > 0
    )::bigint AS g_plus_count
  FROM generate_series(1, 5) AS stars(star)
  LEFT JOIN population AS p ON p.overall_rating = stars.star
  GROUP BY stars.star
),
top_100 AS (
  SELECT *
  FROM population
  WHERE overall_rating >= 4
    AND g_plus_count > 0
  ORDER BY g_plus_count DESC, overall_rating DESC, ccn ASC
  LIMIT 100
),
state_counts AS (
  SELECT state, count(*)::bigint AS facility_count
  FROM population
  WHERE overall_rating >= 4
    AND g_plus_count > 0
    AND state IS NOT NULL
  GROUP BY state
),
year_counts AS (
  SELECT
    extract(year FROM survey_date)::integer AS survey_year,
    count(*)::bigint AS citation_count,
    count(DISTINCT ccn)::bigint AS facility_count
  FROM g_plus_rows
  GROUP BY extract(year FROM survey_date)::integer
),
full_file_harm AS MATERIALIZED (
  SELECT ccn, count(*)::bigint AS g_plus_count
  FROM public.nh_health_deficiencies
  WHERE scope_severity_code IN ('G', 'H', 'I', 'J', 'K', 'L')
  GROUP BY ccn
),
full_file_population AS (
  SELECT
    r.ccn,
    r.overall_rating,
    coalesce(f.g_plus_count, 0)::bigint AS g_plus_count
  FROM ratings AS r
  LEFT JOIN full_file_harm AS f USING (ccn)
),
duplicate_metrics AS (
  SELECT ccn, metric, count(*) AS row_count
  FROM (
    SELECT subject_id AS ccn, metric
    FROM rating_claims
    WHERE metric IN (
      'nursing_overall_rating',
      'nursing_health_inspection_rating',
      'nursing_staffing_rating',
      'nursing_quality_measure_rating'
    )
  ) AS metric_rows
  GROUP BY ccn, metric
  HAVING count(*) <> 1
)
SELECT jsonb_build_object(
  'projectRef', 'rfabugntpwelbpbexgcz',
  'observedAt', current_date,
  'source', (
    SELECT jsonb_build_object(
      'claimPeriod', lp.claim_period,
      -- CMS resource-file URLs are content-addressed and can expire. Keep the
      -- selected production row's release, retrieval, and digest metadata, but
      -- publish the stable, live PDC landing pages for both joined datasets.
      'names', jsonb_build_array(
        'CMS Care Compare — Nursing Home Provider Information (4pq5-n9py)',
        'CMS Care Compare — Nursing Home Health Deficiencies (r5ix-sfxw)'
      ),
      'urls', jsonb_build_array(
        'https://data.cms.gov/provider-data/dataset/4pq5-n9py',
        'https://data.cms.gov/provider-data/dataset/r5ix-sfxw'
      ),
      'publishers', jsonb_build_array(
        'Centers for Medicare & Medicaid Services',
        'Centers for Medicare & Medicaid Services'
      ),
      'provenanceSourceNames',
        jsonb_agg(ps.source_name ORDER BY ps.source_name, ps.id),
      'ratingReleaseDate', max(ps.source_release_date),
      'ratingRetrievedAt', max(ps.retrieved_at),
      'ratingUpstreamModifiedAt', max(ps.upstream_modified_at),
      'ratingContentSha256',
        jsonb_agg(ps.content_sha256 ORDER BY ps.source_name, ps.id),
      'windowStart', wp.window_start,
      'windowEnd', wp.window_end
    )
    FROM period_sources AS ps
    CROSS JOIN latest_period AS lp
    CROSS JOIN window_parameters AS wp
    GROUP BY lp.claim_period, wp.window_start, wp.window_end
  ),
  'integrity', jsonb_build_object(
    'ratingUniverse', (SELECT count(*) FROM rating_subjects),
    'ratedFacilities', (
      SELECT count(*) FROM ratings WHERE overall_rating IS NOT NULL
    ),
    'unratedFacilities', (
      SELECT count(*) FROM ratings WHERE overall_rating IS NULL
    ),
    'duplicateCcnMetrics', (SELECT count(*) FROM duplicate_metrics),
    'invalidOverallRatings', (
      SELECT count(*)
      FROM ratings
      WHERE overall_rating IS NOT NULL
        AND overall_rating NOT BETWEEN 1 AND 5
    ),
    'deficiencyRowsAllFile', (
      SELECT count(*) FROM public.nh_health_deficiencies
    ),
    'deficiencyFacilitiesAllFile', (
      SELECT count(DISTINCT ccn) FROM public.nh_health_deficiencies
    ),
    'surveyDateMinAllFile', (
      SELECT min(survey_date) FROM public.nh_health_deficiencies
    ),
    'surveyDateMaxAllFile', (
      SELECT max(survey_date) FROM public.nh_health_deficiencies
    ),
    'gPlusCitationsWindow', (SELECT count(*) FROM g_plus_rows),
    'gPlusFacilitiesWindow', (SELECT count(DISTINCT ccn) FROM g_plus_rows),
    'gPlusCitationsJoinedToRatedFacilities', (
      SELECT coalesce(sum(g_plus_count), 0)
      FROM population
      WHERE overall_rating IS NOT NULL
    )
  ),
  'headline', jsonb_build_object(
    'totalHighStarFacilities', (
      SELECT count(*) FROM population WHERE overall_rating >= 4
    ),
    'highStarWithHarm', (
      SELECT count(*)
      FROM population
      WHERE overall_rating >= 4 AND g_plus_count > 0
    ),
    'fourStarWithHarm', (
      SELECT count(*)
      FROM population
      WHERE overall_rating = 4 AND g_plus_count > 0
    ),
    'fiveStarWithHarm', (
      SELECT count(*)
      FROM population
      WHERE overall_rating = 5 AND g_plus_count > 0
    ),
    'pctHighStarWithHarm', (
      SELECT round(
        100.0 * count(*) FILTER (WHERE g_plus_count > 0) /
        nullif(count(*), 0),
        2
      )
      FROM population
      WHERE overall_rating >= 4
    ),
    'highStarWithNonIdrHarm', (
      SELECT count(*)
      FROM population
      WHERE overall_rating >= 4 AND non_idr_g_plus_count > 0
    ),
    'pctHighStarWithNonIdrHarm', (
      SELECT round(
        100.0 * count(*) FILTER (WHERE non_idr_g_plus_count > 0) /
        nullif(count(*), 0),
        2
      )
      FROM population
      WHERE overall_rating >= 4
    ),
    'totalGPlusCitations', (SELECT count(*) FROM g_plus_rows),
    'lowStarZeroHarm', (
      SELECT count(*)
      FROM population
      WHERE overall_rating IN (1, 2) AND g_plus_count = 0
    )
  ),
  'fullInspectionCycleSensitivity', jsonb_build_object(
    'surveyDateMin', (
      SELECT min(survey_date)
      FROM public.nh_health_deficiencies
      WHERE scope_severity_code IN ('G', 'H', 'I', 'J', 'K', 'L')
    ),
    'surveyDateMax', (
      SELECT max(survey_date)
      FROM public.nh_health_deficiencies
      WHERE scope_severity_code IN ('G', 'H', 'I', 'J', 'K', 'L')
    ),
    'highStarWithHarm', (
      SELECT count(*)
      FROM full_file_population
      WHERE overall_rating >= 4 AND g_plus_count > 0
    ),
    'pctHighStarWithHarm', (
      SELECT round(
        100.0 * count(*) FILTER (WHERE g_plus_count > 0) /
        nullif(count(*), 0),
        2
      )
      FROM full_file_population
      WHERE overall_rating >= 4
    )
  ),
  'rateByStarBand', (
    SELECT jsonb_agg(
      jsonb_build_object(
        'star', star,
        'gPlusCount', g_plus_count,
        'totalCount', total_count,
        'ratePct', round(100.0 * g_plus_count / nullif(total_count, 0), 2)
      )
      ORDER BY star
    )
    FROM star_bands
  ),
  'avgSubRatings', jsonb_build_object(
    'allFacilities', (
      SELECT jsonb_build_object(
        'hi', round(avg(health_inspection_rating), 2),
        'st', round(avg(staffing_rating), 2),
        'qm', round(avg(qm_rating), 2)
      )
      FROM population
      WHERE overall_rating IS NOT NULL
    ),
    'allGPlusFacilities', (
      SELECT jsonb_build_object(
        'hi', round(avg(health_inspection_rating), 2),
        'st', round(avg(staffing_rating), 2),
        'qm', round(avg(qm_rating), 2)
      )
      FROM population
      WHERE overall_rating IS NOT NULL AND g_plus_count > 0
    ),
    'highStarGPlusFacilities', (
      SELECT jsonb_build_object(
        'hi', round(avg(health_inspection_rating), 2),
        'st', round(avg(staffing_rating), 2),
        'qm', round(avg(qm_rating), 2)
      )
      FROM population
      WHERE overall_rating >= 4 AND g_plus_count > 0
    )
  ),
  'top100HighStarHarm', (
    SELECT jsonb_agg(
      jsonb_build_object(
        'ccn', ccn,
        'facilityName', provider_name,
        'city', city_town,
        'state', state,
        'overallRating', overall_rating,
        'healthInspectionRating', health_inspection_rating,
        'staffingRating', staffing_rating,
        'qmRating', qm_rating,
        'gPlusCount', g_plus_count,
        'latestGPlusSurveyDate', latest_g_plus_survey_date,
        'worstSeverity', worst_severity
      )
      ORDER BY g_plus_count DESC, overall_rating DESC, ccn ASC
    )
    FROM top_100
  ),
  'topStatesByHighStarHarm', (
    SELECT jsonb_agg(
      jsonb_build_object('state', state, 'count', facility_count)
      ORDER BY facility_count DESC, state ASC
    )
    FROM state_counts
  ),
  'yearBySurveyDate', (
    SELECT jsonb_agg(
      jsonb_build_object(
        'year', survey_year,
        'citations', citation_count,
        'facilities', facility_count
      )
      ORDER BY survey_year
    )
    FROM year_counts
  )
) AS result;
