-- Rectification anonymous aggregate telemetry
-- (docs/tasks/TASK-rectification-telemetry-20260926.md).
--
-- One row per rectification Case, written once, when its range card is first
-- delivered. The row holds numbers and closed enums only. It carries no user,
-- Case or session id, no birth data, no names, no conversation text, no model
-- output, and no timestamp finer than the ISO week (`recorded_week`).
--
-- Dedupe lives in a separate ledger keyed by case_id. The ledger cascades with
-- the Case (and so with account deletion) and has no column that points at a
-- stats row, so a stats row cannot be joined back to a person.
--
-- Access:
--   * both tables enable RLS and grant no table privilege to any runtime role;
--   * the web server writes only through record_rectification_telemetry
--     (SECURITY DEFINER, service_role only);
--   * the admin reads only aggregates through rectification_telemetry_summary
--     (SECURITY DEFINER, admin_runtime only). There is no per-row read path.
--
-- Retention: 180 days. Every write first deletes rows whose week started more
-- than 180 days ago; the summary never reads past 180 days either; and
-- purge_expired_rectification_telemetry() lets an operator run the same delete.
--
-- Backward compatibility: additive only. Two new tables and three new
-- functions; no existing table, column, function or grant changes, so the code
-- that is deployed before this migration keeps working unchanged.

begin;

create table if not exists public.rectification_telemetry (
  id uuid primary key default gen_random_uuid(),
  recorded_week date not null
    default (pg_catalog.date_trunc('week', pg_catalog.timezone('UTC', pg_catalog.now())))::date
    check (extract(isodow from recorded_week) = 1),
  window_radius_minutes integer check (window_radius_minutes between 0 and 720),
  birth_time_source text not null
    check (birth_time_source in ('hospital_record', 'approximate', 'period_only', 'unknown')),
  questions_total integer not null check (questions_total between 0 and 1000),
  questions_targeted integer not null check (questions_targeted between 0 and 1000),
  questions_guided integer not null check (questions_guided between 0 and 1000),
  questions_dated_probe integer not null check (questions_dated_probe between 0 and 1000),
  questions_personality integer not null check (questions_personality between 0 and 1000),
  questions_open integer not null check (questions_open between 0 and 1000),
  experiences_added integer not null check (experiences_added between 0 and 1000),
  range_width_minutes integer check (range_width_minutes between 0 and 1440),
  candidate_count integer not null check (candidate_count between 0 and 1440),
  top_two_gap_points integer check (top_two_gap_points between 0 and 100),
  stop_reason text not null check (stop_reason in (
    'converged',
    'pool_exhausted',
    'user_no_more',
    'round_cap',
    'user_stopped',
    'error'
  )),
  precision_gate_met boolean,
  duration_seconds integer not null check (duration_seconds between 0 and 31536000),
  algorithm_version text check (algorithm_version ~ '^[A-Za-z0-9][A-Za-z0-9._:+-]{0,63}$'),
  policy_version text check (policy_version ~ '^[A-Za-z0-9][A-Za-z0-9._:+-]{0,63}$'),
  skill_version text check (skill_version ~ '^[A-Za-z0-9][A-Za-z0-9._:+-]{0,63}$'),
  check (
    questions_targeted + questions_guided + questions_dated_probe
      + questions_personality + questions_open <= questions_total
  )
);

create index if not exists rectification_telemetry_week_idx
  on public.rectification_telemetry (recorded_week);

create table if not exists public.rectification_telemetry_reported_cases (
  case_id uuid primary key
    references public.agentic_rectification_cases(id) on delete cascade
);

alter table public.rectification_telemetry enable row level security;
alter table public.rectification_telemetry_reported_cases enable row level security;

revoke all on table public.rectification_telemetry from public, anon, authenticated, service_role;
revoke all on table public.rectification_telemetry_reported_cases from public, anon, authenticated, service_role;

do $$
begin
  if exists (select 1 from pg_roles where rolname = 'app_runtime') then
    revoke all on table public.rectification_telemetry from app_runtime;
    revoke all on table public.rectification_telemetry_reported_cases from app_runtime;
  end if;
  if exists (select 1 from pg_roles where rolname = 'admin_runtime') then
    revoke all on table public.rectification_telemetry from admin_runtime;
    revoke all on table public.rectification_telemetry_reported_cases from admin_runtime;
  end if;
end;
$$;

-- ---------------------------------------------------------------------------
-- Write path: server only. The ids are used for the ownership check and the
-- dedupe ledger; neither is written into the stats row.
-- ---------------------------------------------------------------------------

create or replace function public.record_rectification_telemetry(
  p_user_id uuid,
  p_case_id uuid,
  p_window_radius_minutes integer,
  p_birth_time_source text,
  p_questions_total integer,
  p_questions_targeted integer,
  p_questions_guided integer,
  p_questions_dated_probe integer,
  p_questions_personality integer,
  p_questions_open integer,
  p_experiences_added integer,
  p_range_width_minutes integer,
  p_candidate_count integer,
  p_top_two_gap_points integer,
  p_stop_reason text,
  p_precision_gate_met boolean,
  p_duration_seconds integer,
  p_algorithm_version text,
  p_policy_version text,
  p_skill_version text
)
returns boolean
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_marked uuid;
begin
  if p_user_id is null or p_case_id is null then
    raise exception 'rectification_telemetry_invalid' using errcode = '22023';
  end if;
  perform 1 from public.agentic_rectification_cases c
    where c.id = p_case_id and c.user_id = p_user_id;
  if not found then
    raise exception 'rectification_telemetry_case_not_found' using errcode = 'P0002';
  end if;

  delete from public.rectification_telemetry
    where recorded_week < (pg_catalog.timezone('UTC', pg_catalog.now()))::date - 180;

  insert into public.rectification_telemetry_reported_cases (case_id)
    values (p_case_id)
    on conflict (case_id) do nothing
    returning case_id into v_marked;
  if v_marked is null then
    return false;
  end if;

  insert into public.rectification_telemetry (
    window_radius_minutes,
    birth_time_source,
    questions_total,
    questions_targeted,
    questions_guided,
    questions_dated_probe,
    questions_personality,
    questions_open,
    experiences_added,
    range_width_minutes,
    candidate_count,
    top_two_gap_points,
    stop_reason,
    precision_gate_met,
    duration_seconds,
    algorithm_version,
    policy_version,
    skill_version
  ) values (
    p_window_radius_minutes,
    p_birth_time_source,
    p_questions_total,
    p_questions_targeted,
    p_questions_guided,
    p_questions_dated_probe,
    p_questions_personality,
    p_questions_open,
    p_experiences_added,
    p_range_width_minutes,
    p_candidate_count,
    p_top_two_gap_points,
    p_stop_reason,
    p_precision_gate_met,
    p_duration_seconds,
    p_algorithm_version,
    p_policy_version,
    p_skill_version
  );
  return true;
end;
$$;

create or replace function public.purge_expired_rectification_telemetry()
returns integer
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_deleted integer;
begin
  delete from public.rectification_telemetry
    where recorded_week < (pg_catalog.timezone('UTC', pg_catalog.now()))::date - 180;
  get diagnostics v_deleted = row_count;
  return v_deleted;
end;
$$;

-- ---------------------------------------------------------------------------
-- Read path: aggregates only (medians, distributions, weekly trend). Weeks
-- are capped at 26 and rows older than 180 days are never read.
-- ---------------------------------------------------------------------------

create or replace function public.rectification_telemetry_summary(p_weeks integer default 12)
returns jsonb
language sql
stable
security definer
set search_path = ''
as $$
  with bounds as (
    select
      greatest(1, least(coalesce(p_weeks, 12), 26)) as weeks,
      (pg_catalog.timezone('UTC', pg_catalog.now()))::date as today
  ), window_start as (
    select
      b.weeks,
      greatest(
        (pg_catalog.date_trunc('week', b.today::timestamp))::date - (b.weeks - 1) * 7,
        b.today - 180
      ) as since
    from bounds b
  ), recent as (
    select t.*
    from public.rectification_telemetry t, window_start w
    where t.recorded_week >= w.since
  ), width_buckets(key, lo, hi, ord) as (
    values ('0', 0, 0, 1), ('1-5', 1, 5, 2), ('6-10', 6, 10, 3), ('11-20', 11, 20, 4),
      ('21-30', 21, 30, 5), ('31-60', 31, 60, 6), ('61+', 61, null, 7)
  ), question_buckets(key, lo, hi, ord) as (
    values ('0-3', 0, 3, 1), ('4-6', 4, 6, 2), ('7-9', 7, 9, 3), ('10-12', 10, 12, 4),
      ('13-15', 13, 15, 5), ('16+', 16, null, 6)
  ), gap_buckets(key, lo, hi, ord) as (
    values ('0-2', 0, 2, 1), ('3-4', 3, 4, 2), ('5-9', 5, 9, 3), ('10+', 10, null, 4)
  ), stop_reasons(key, ord) as (
    values ('converged', 1), ('pool_exhausted', 2), ('user_no_more', 3),
      ('round_cap', 4), ('user_stopped', 5), ('error', 6)
  ), sources(key, ord) as (
    values ('hospital_record', 1), ('approximate', 2), ('period_only', 3), ('unknown', 4)
  ), weeks as (
    select (pg_catalog.date_trunc('week', b.today::timestamp))::date - g.n * 7 as week
    from bounds b, pg_catalog.generate_series(0, (select weeks from bounds) - 1) as g(n)
  )
  select pg_catalog.jsonb_build_object(
    'weeks', (select weeks from window_start),
    'since', (select since from window_start),
    'retention_days', 180,
    'sessions', (select count(*) from recent),
    'medians', (
      select pg_catalog.jsonb_build_object(
        'range_width_minutes_p50', round((percentile_cont(0.5) within group (order by range_width_minutes))::numeric, 1),
        'range_width_minutes_p90', round((percentile_cont(0.9) within group (order by range_width_minutes))::numeric, 1),
        'questions_total_p50', round((percentile_cont(0.5) within group (order by questions_total))::numeric, 1),
        'questions_total_p90', round((percentile_cont(0.9) within group (order by questions_total))::numeric, 1),
        'experiences_added_p50', round((percentile_cont(0.5) within group (order by experiences_added))::numeric, 1),
        'candidate_count_p50', round((percentile_cont(0.5) within group (order by candidate_count))::numeric, 1),
        'window_radius_minutes_p50', round((percentile_cont(0.5) within group (order by window_radius_minutes))::numeric, 1),
        'duration_seconds_p50', round((percentile_cont(0.5) within group (order by duration_seconds))::numeric, 1),
        'duration_seconds_p90', round((percentile_cont(0.9) within group (order by duration_seconds))::numeric, 1)
      )
      from recent
    ),
    'question_types', (
      select pg_catalog.jsonb_build_object(
        'targeted', round(coalesce(avg(questions_targeted), 0)::numeric, 2),
        'guided', round(coalesce(avg(questions_guided), 0)::numeric, 2),
        'dated_probe', round(coalesce(avg(questions_dated_probe), 0)::numeric, 2),
        'personality', round(coalesce(avg(questions_personality), 0)::numeric, 2),
        'open', round(coalesce(avg(questions_open), 0)::numeric, 2),
        'other', round(coalesce(avg(
          questions_total - questions_targeted - questions_guided
            - questions_dated_probe - questions_personality - questions_open
        ), 0)::numeric, 2),
        'guided_asked_sessions', count(*) filter (where questions_guided > 0)
      )
      from recent
    ),
    'width_distribution', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object('key', x.key, 'count', x.n) order by x.ord)
      from (
        select wb.key, wb.ord, count(r.id) as n
        from width_buckets wb
        left join recent r
          on r.range_width_minutes >= wb.lo
         and (wb.hi is null or r.range_width_minutes <= wb.hi)
        group by wb.key, wb.ord
        union all
        select 'unknown', 99, count(*) from recent where range_width_minutes is null
      ) x
    ),
    'question_distribution', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object('key', x.key, 'count', x.n) order by x.ord)
      from (
        select qb.key, qb.ord, count(r.id) as n
        from question_buckets qb
        left join recent r
          on r.questions_total >= qb.lo
         and (qb.hi is null or r.questions_total <= qb.hi)
        group by qb.key, qb.ord
      ) x
    ),
    'gap_distribution', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object('key', x.key, 'count', x.n) order by x.ord)
      from (
        select gb.key, gb.ord, count(r.id) as n
        from gap_buckets gb
        left join recent r
          on r.top_two_gap_points >= gb.lo
         and (gb.hi is null or r.top_two_gap_points <= gb.hi)
        group by gb.key, gb.ord
        union all
        select 'single', 99, count(*) from recent where top_two_gap_points is null
      ) x
    ),
    'stop_reasons', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object('key', s.key, 'count', (
        select count(*) from recent r where r.stop_reason = s.key
      )) order by s.ord)
      from stop_reasons s
    ),
    'birth_time_sources', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object('key', s.key, 'count', (
        select count(*) from recent r where r.birth_time_source = s.key
      )) order by s.ord)
      from sources s
    ),
    'precision_gate', (
      select pg_catalog.jsonb_build_object(
        'met', count(*) filter (where precision_gate_met is true),
        'not_met', count(*) filter (where precision_gate_met is false),
        'unknown', count(*) filter (where precision_gate_met is null)
      )
      from recent
    ),
    'weekly', (
      select pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object(
        'week', w.week,
        'sessions', (select count(*) from recent r where r.recorded_week = w.week),
        'range_width_minutes_p50', (
          select round((percentile_cont(0.5) within group (order by r.range_width_minutes))::numeric, 1)
          from recent r where r.recorded_week = w.week
        ),
        'questions_total_p50', (
          select round((percentile_cont(0.5) within group (order by r.questions_total))::numeric, 1)
          from recent r where r.recorded_week = w.week
        ),
        'precision_gate_met', (
          select count(*) from recent r where r.recorded_week = w.week and r.precision_gate_met is true
        ),
        'precision_gate_known', (
          select count(*) from recent r where r.recorded_week = w.week and r.precision_gate_met is not null
        )
      ) order by w.week)
      from weeks w
      where w.week >= (select since from window_start) - 6
    ),
    'versions', (
      select coalesce(pg_catalog.jsonb_agg(pg_catalog.jsonb_build_object(
        'algorithm_version', v.algorithm_version,
        'policy_version', v.policy_version,
        'skill_version', v.skill_version,
        'count', v.n
      ) order by v.n desc, v.algorithm_version, v.policy_version, v.skill_version), '[]'::jsonb)
      from (
        select algorithm_version, policy_version, skill_version, count(*) as n
        from recent
        group by algorithm_version, policy_version, skill_version
        order by count(*) desc
        limit 10
      ) v
    )
  );
$$;

revoke all on function public.record_rectification_telemetry(
  uuid, uuid, integer, text, integer, integer, integer, integer, integer, integer,
  integer, integer, integer, integer, text, boolean, integer, text, text, text
) from public, anon, authenticated;
revoke all on function public.purge_expired_rectification_telemetry() from public, anon, authenticated;
revoke all on function public.rectification_telemetry_summary(integer) from public, anon, authenticated;

grant execute on function public.record_rectification_telemetry(
  uuid, uuid, integer, text, integer, integer, integer, integer, integer, integer,
  integer, integer, integer, integer, text, boolean, integer, text, text, text
) to service_role;
grant execute on function public.purge_expired_rectification_telemetry() to service_role;

do $$
begin
  if exists (select 1 from pg_roles where rolname = 'admin_runtime') then
    grant execute on function public.rectification_telemetry_summary(integer) to admin_runtime;
  end if;
end;
$$;

commit;