HealthLog/src/lib/insights/feature-blocks.ts

/**
 * Briefing-scoped feature read blocks: full-history extremes, the
 * bucketed rollup series, and the v1.22 cross-signal integration blocks
 * (flagged labs, preventive-care due/overdue, workout aggregate). Each
 * is one bounded DB read projecting to its slice of the
 * `AggregatedFeatures` payload.
 *
 * Extracted verbatim from `features.ts`, which re-exports this module
 * so every existing call site keeps importing from there; the assembly
 * point (`extractFeatures`) stays in the hub.
 */
import { prisma } from "@/lib/db";
import { Prisma } from "@/generated/prisma/client";
import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql";
import { dayWeightedRowsSql, foldMeanSql } from "@/lib/measurements/day-mean";
import {
  loadUserSourcePriority,
  utcBucketEnd,
} from "@/lib/rollups/measurement-read";
import { classifyAgainstEffectiveRange } from "@/lib/labs/reference-range";
import { calendarDaysUntil } from "@/lib/measurement-reminders/due-day";
import { resolveLabFields } from "@/lib/labs/serialise";
import { isModuleEnabled } from "@/lib/modules/gate";
import {
  readRollupBuckets,
  ROLLUP_FOLD_WINDOW_MS,
} from "@/lib/rollups/measurement-rollups";
import {
  canonicalWorkoutsInRange,
  readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
import { deriveBucketedTypes } from "@/lib/signals/adapters/correlation";
import type {
  AggregatedFeatures,
  BucketedSeries,
} from "@/lib/insights/features";
import type {
  MeasurementType,
  RollupGranularity,
} from "@/generated/prisma/client";

/** All-time aggregate (full history) for one measurement type. */
interface AllTimeExtremes {
  mean: number | null;
  min: number | null;
  max: number | null;
}

/**
 * v1.18.11 — full-history min / max / mean per measurement type, folded in
 * SQL with NO row materialisation in JS. Fills the `allTime*` feature fields;
 * the windowed `summarize()` covers trends + recent windows, and this covers
 * the long-horizon figures the prompt labels "allTime".
 *
 * One source per (type, day): each source is folded per day first, each day
 * keeps its ladder-canonical source, then the days fold into the figures —
 * the same pick the dashboard's all-time figures make. Averaging every raw
 * row let a device that samples densely (a watch's pulse) outweigh the
 * readings the source ladder prefers on the same days.
 *
 * Where the days come from. A day is the same per-source fold either way —
 * count, sum, the extremes of the readings and, for an hourly-mean type, the
 * day value — and the DAY rollup tier already holds exactly that per (type,
 * source, day), recomputed on every write. Reading the raw rows instead meant
 * two passes over every reading of these types the account ever recorded on
 * every call, the largest read on the database. So the figures fold:
 *
 *   - the DAY tier, for every whole UTC day from `tierFrom` on, for each type
 *     whose tier rows in that range all carry the reading extremes
 *     (migration 0417; the boot re-fold fills older rows) and account for
 *     the live record day by day (same days, same reading counts);
 *   - the raw rows for the rest: the days before `tierFrom` (older than the
 *     fold window, or the partial day the `floor` falls in), and every day of
 *     a type the tier does not cover yet or no longer matches.
 *
 * The split is at a UTC midnight, so no day is folded from both, and the
 * canonical pick per day is the same on both sides. The tier's pulse day
 * value follows the stored day-mean method; the raw fold follows this
 * process's mirror of it, which trails a change by at most a minute.
 *
 * Only the four types that expose `allTime*` fields are aggregated (weight,
 * systolic, diastolic, pulse). Returns a map keyed by `MeasurementType`; a type
 * with no rows is simply absent.
 */
export async function readAllTimeExtremes(
  userId: string,
  types: readonly MeasurementType[],
  /**
   * The Coach's lookback floor. Given, "all time" means all the history the
   * Coach may read, and nothing older enters the figures.
   */
  floor: Date | null = null,
): Promise<Map<MeasurementType, AllTimeExtremes>> {
  const out = new Map<MeasurementType, AllTimeExtremes>();
  if (types.length === 0) return out;
  // The first whole UTC day the tier may answer: after the fold window's
  // edge (the fold writes nothing older, and the edge day may be partial)
  // and after the floor's own day (only part of it is inside the floor).
  const windowEdge = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
  const tierFrom = utcBucketEnd(
    floor !== null && floor > windowEdge ? floor : windowEdge,
    "DAY",
  );
  const typeList = (list: readonly MeasurementType[]) =>
    Prisma.join(list.map((t) => Prisma.sql`${t}::measurement_type`));

  const [priorityJson, tierRows] = await Promise.all([
    loadUserSourcePriority(userId),
    // A type is read from the tier only when the tier accounts for the live
    // record day by day: every day with a live reading has DAY rows whose
    // counts add up to that day's live count, and every DAY row has its day
    // of live readings. A bucket that outlived its rows (a hard delete with no
    // rollup recompute) or a day the tier never got fails the comparison and
    // the type is read raw. The figures from here are the anchor the Coach
    // snapshot checks its coarse history against, so they must never repeat
    // a stale bucket. The live side is a count per day off the partial
    // `(user_id, type, measured_at)` live index, not a read of the rows.
    prisma.$queryRaw<Array<{ type: string }>>`
      WITH live AS (
        SELECT m."type", date_trunc('day', m."measured_at") AS day,
               COUNT(*)::bigint AS n
        FROM measurements m
        WHERE m."user_id" = ${userId}
          AND m."deleted_at" IS NULL
          AND m."type" IN (${typeList(types)})
          AND m."measured_at" >= ${tierFrom}
        GROUP BY 1, 2
      ),
      tier AS (
        SELECT r."type", r."bucket_start" AT TIME ZONE 'UTC' AS day,
               SUM(r."count")::bigint AS n,
               bool_and(
                 r."reading_min_value" IS NOT NULL
                 AND r."reading_max_value" IS NOT NULL
               ) AS complete
        FROM measurement_rollups r
        WHERE r."user_id" = ${userId}
          AND r."granularity" = 'DAY'
          AND r."type" IN (${typeList(types)})
          AND r."bucket_start" >= ${tierFrom}
        GROUP BY 1, 2
      )
      SELECT COALESCE(l."type", t."type")::text AS type
      FROM live l
      FULL JOIN tier t ON t."type" = l."type" AND t.day = l.day
      GROUP BY 1
      HAVING bool_and(COALESCE(l.n = t.n AND t.complete, false))
    `,
  ]);
  const tierTypes = new Set(tierRows.map((r) => r.type));
  const fromTier = types.filter((t) => tierTypes.has(t));
  const rawOnly = types.filter((t) => !tierTypes.has(t));

  const since =
    floor === null ? Prisma.empty : Prisma.sql`AND m."measured_at" >= ${floor}`;
  const rawArms: Prisma.Sql[] = [];
  if (rawOnly.length > 0) {
    rawArms.push(Prisma.sql`
      SELECT m."type", m."source", m."measured_at", m."value"
      FROM measurements m
      WHERE m."user_id" = ${userId}
        AND m."deleted_at" IS NULL
        AND m."type" IN (${typeList(rawOnly)})
        ${since}`);
  }
  if (fromTier.length > 0) {
    rawArms.push(Prisma.sql`
      SELECT m."type", m."source", m."measured_at", m."value"
      FROM measurements m
      WHERE m."user_id" = ${userId}
        AND m."deleted_at" IS NULL
        AND m."type" IN (${typeList(fromTier)})
        ${since}
        AND m."measured_at" < ${tierFrom}`);
  }
  const rank = (alias: string) =>
    Prisma.raw(
      buildSourceRankCase(priorityJson, `${alias}."type"`, `${alias}."source"`),
    );
  // The tier's canonical days, in the shape of the raw fold's: the day's
  // reading-weighted sum and count, its reading extremes, and its day value
  // with a weight of one (a DAY row's `mean` is the day value for an
  // hourly-mean type).
  const tierCanon =
    fromTier.length === 0
      ? Prisma.empty
      : Prisma.sql`
    UNION ALL
    SELECT * FROM (
      SELECT DISTINCT ON (r."type", r."bucket_start")
        r."type",
        r."count"                                                AS cnt,
        COALESCE(r."sum_value", r."count" * r."mean")::double precision
                                                                 AS total,
        r."reading_min_value"                                    AS min_value,
        r."reading_max_value"                                    AS max_value,
        r."mean"                                                 AS wsum,
        1.0::double precision                                    AS wdays
      FROM measurement_rollups r
      WHERE r."user_id" = ${userId}
        AND r."granularity" = 'DAY'
        AND r."type" IN (${typeList(fromTier)})
        AND r."bucket_start" >= ${tierFrom}
      ORDER BY r."type", r."bucket_start", (${rank("r")}), r."source"::text
    ) tier`;
  // A day of pulse is the mean of its hours' means and the full history the
  // mean of its days (`day-mean.ts`). `per_source` carries, per source and
  // day, the day-weighted sum and the day's weight (one) beside the reading
  // count and sum; the mean below reads the weighted pair for those types
  // only, and every other type keeps the reading-weighted mean, unchanged.
  //
  // One fold of one scan, no join. The weighted pair used to be a second
  // grouping of the same rows, joined back to the per-source days before the
  // pick; on a measurements table without planner statistics (a new install,
  // an import under the autoanalyze threshold) that join ran as a nested
  // loop that re-ran the weighting for every source day, and ten years of
  // readings took seconds. `tests/integration/per-source-fold-parity.test.ts`
  // holds the joined statement as the reference.
  const mean = Prisma.raw(
    foldMeanSql({
      typeColumn: 'c."type"',
      total: "c.total",
      count: "c.cnt",
      weightedSum: "c.wsum",
      weightSum: "c.wdays",
    }),
  );
  const rows = await prisma.$queryRaw<
    Array<{ type: string; mean: number; min: number; max: number }>
  >`
    WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}),
    per_source AS (
      SELECT
        l."type",
        l."source",
        date_trunc('day', l."measured_at")              AS day,
        COUNT(*)::int                                   AS cnt,
        SUM(l."value")::double precision                AS total,
        MIN(l."value")::double precision                AS min_value,
        MAX(l."value")::double precision                AS max_value,
        SUM(l."value" * l.day_weight)::double precision AS wsum,
        SUM(l.day_weight)::double precision             AS wdays
      FROM ${dayWeightedRowsSql("live", null, { bySource: true })} l
      GROUP BY l."type", l."source", 3
    )
    SELECT
      c."type"::text                                  AS type,
      ${mean}::double precision                       AS mean,
      MIN(c.min_value)::double precision              AS min,
      MAX(c.max_value)::double precision              AS max
    FROM (
      SELECT * FROM (
        SELECT DISTINCT ON (p."type", p.day)
          p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays
        FROM per_source p
        ORDER BY p."type", p.day, (${rank("p")}), p."source"::text
      ) raw
      ${tierCanon}
    ) c
    GROUP BY c."type"
  `;
  for (const r of rows) {
    out.set(r.type as MeasurementType, {
      mean: r.mean === null ? null : Number(r.mean),
      min: r.min === null ? null : Number(r.min),
      max: r.max === null ? null : Number(r.max),
    });
  }
  return out;
}

/**
 * v1.4.36 — bucket-window definitions for the
 * `bucketedMeasurements` payload. Mirrors the rollup populator's
 * granularity ladder so the read-side picks up whatever the persistent
 * table holds without a recompute round-trip.
 *
 * The 90 / 365 / 1825-day windows are non-overlapping: each row of
 * `measurement_rollups` lives at exactly one granularity, and the
 * downstream model reads the union of the three series per type.
 * Total volume per type for a heavy power user (5 years of daily
 * data): 90 DAY + 39 WEEK + 50 MONTH = 179 buckets × ~40 bytes JSON
 * each ≈ 7 KB. Eight metric types lands at ~56 KB — well under the
 * 5 MB cap above, vs 25.9 MB for the v1.4.35 rawMeasurements shape.
 */
const BUCKET_WINDOWS: Array<{
  granularity: RollupGranularity;
  fromDays: number;
  toDays: number;
}> = [
  { granularity: "DAY", fromDays: 0, toDays: 90 },
  { granularity: "WEEK", fromDays: 90, toDays: 365 },
  { granularity: "MONTH", fromDays: 365, toDays: 1825 },
];

/**
 * Types the bucketed payload covers. Mirrors the aggregate branches
 * above so the model never sees a bucket for a metric whose aggregate
 * block was suppressed. New `MeasurementType` enum values flow in by
 * adding one row; the rollup populator already covers every type.
 */
// Derived from the signal registry: every signal flagged
// `surfaces.correlationEligible` projects to its DB `MeasurementType`. The list
// is a membership/iteration set (each type is read independently), so order is
// not significant; the registry-invariant test pins the set byte-for-byte.
const BUCKETED_TYPES: MeasurementType[] = deriveBucketedTypes();

/** Days a preventive-care item must be due within to surface as "due soon". */
const PREVENTIVE_DUE_HORIZON_DAYS = 21;

/** Cap on items surfaced per preventive-care bucket. */
const PREVENTIVE_MAX_PER_BUCKET = 5;

/** Cap on flagged biomarkers surfaced to the briefing. */
const LABS_MAX_FLAGGED = 8;

/** Only lab readings within this many months are considered "recent". */
const LABS_LOOKBACK_MONTHS = 12;

/** Trailing window (days) for the workout aggregate. */
const WORKOUT_WINDOW_DAYS = 90;

/** Trailing window (days) for the ECG recording descriptor. */
const ECG_WINDOW_DAYS = 90;

/** Defensive cap on ECG rows read for the descriptor (a bound, not a ceiling). */
const ECG_MAX_ROWS = 500;

const MS_PER_DAY = 24 * 60 * 60 * 1000;

/**
 * Strip control chars + collapse whitespace, then bound the length, before a
 * user-supplied label can reach the briefing prompt. Mirrors the labs /
 * illness snapshot label handling — a self-scoped prompt-injection surface.
 */
function sanitizeLabel(text: string, max = 80): string {
  let out = "";
  for (const ch of text) {
    const code = ch.codePointAt(0) ?? 0;
    out += code < 0x20 || (code >= 0x7f && code <= 0x9f) ? " " : ch;
  }
  return out.replace(/\s+/g, " ").trim().slice(0, max);
}

/**
 * v1.22 — recent FLAGGED biomarkers (abnormal or trending) for the briefing.
 * Most-recent reading per biomarker over the lookback window; hidden markers
 * excluded; qualitative rows neutral. Only abnormal (below/above) OR trending
 * markers surface, bounded. Returns `undefined` when nothing is flagged so the
 * block is omitted rather than emitting an empty shape, and when the `labs`
 * module is off, before any row is read: the briefing prompt never carries a
 * switched-off domain.
 */
export async function readLabsBriefingBlock(
  userId: string,
  now: number,
): Promise<AggregatedFeatures["labs"] | undefined> {
  if (!(await isModuleEnabled(userId, "labs"))) return undefined;
  const nowDate = new Date(now);
  const cutoff = new Date(nowDate);
  cutoff.setMonth(cutoff.getMonth() - LABS_LOOKBACK_MONTHS);

  const rows = await prisma.labResult.findMany({
    where: { userId, deletedAt: null, takenAt: { gte: cutoff, lte: nowDate } },
    orderBy: { takenAt: "desc" },
    take: LABS_MAX_FLAGGED * 16,
    select: {
      analyte: true,
      panel: true,
      unit: true,
      value: true,
      valueText: true,
      referenceLow: true,
      referenceHigh: true,
      sourceReferenceLow: true,
      sourceReferenceHigh: true,
      sourceReferenceText: true,
      sourceUnit: true,
      takenAt: true,
      biomarkerId: true,
      biomarker: {
        select: {
          id: true,
          name: true,
          unit: true,
          lowerBound: true,
          upperBound: true,
          panel: true,
          hidden: true,
        },
      },
    },
  });
  if (rows.length === 0) return undefined;

  // Group rows per biomarker identity (linked id, else lower-cased analyte),
  // newest-first, so we can read the latest reading + the immediately prior one
  // for a trend. Hidden markers are dropped entirely.
  const byMarker = new Map<string, typeof rows>();
  for (const row of rows) {
    if (row.biomarker?.hidden) continue;
    const resolved = resolveLabFields(row, row.biomarker);
    const key = row.biomarkerId ?? `analyte:${resolved.analyte.toLowerCase()}`;
    const list = byMarker.get(key) ?? [];
    list.push(row);
    byMarker.set(key, list);
  }

  const flagged: NonNullable<AggregatedFeatures["labs"]>["flagged"] = [];
  for (const list of byMarker.values()) {
    const latest = list[0];
    const resolved = resolveLabFields(latest, latest.biomarker);
    const rangeStatus = classifyAgainstEffectiveRange(
      latest.value,
      resolved.effectiveRange,
    );

    // Trend = latest numeric reading vs the immediately prior numeric reading.
    let trend: "rising" | "falling" | "flat" | null = null;
    if (latest.value !== null) {
      const prior = list.find((r, i) => i > 0 && r.value !== null);
      if (prior?.value != null) {
        const delta = latest.value - prior.value;
        const eps = Math.max(Math.abs(prior.value) * 0.02, 1e-9);
        trend = delta > eps ? "rising" : delta < -eps ? "falling" : "flat";
      }
    }

    const isAbnormal = rangeStatus === "below" || rangeStatus === "above";
    const isTrending = trend === "rising" || trend === "falling";
    if (!isAbnormal && !isTrending) continue;

    flagged.push({
      analyte: resolved.analyte,
      value: latest.value,
      valueText: latest.valueText ? sanitizeLabel(latest.valueText, 60) : null,
      unit: resolved.unit,
      rangeStatus,
      trend,
      takenAt: latest.takenAt.toISOString(),
      daysAgo: Math.round((now - latest.takenAt.getTime()) / MS_PER_DAY),
    });
  }

  if (flagged.length === 0) return undefined;
  // Abnormal markers lead, then most-recent first.
  flagged.sort((a, b) => {
    const abn = (s: typeof a.rangeStatus) =>
      s === "below" || s === "above" ? 0 : 1;
    const d = abn(a.rangeStatus) - abn(b.rangeStatus);
    return d !== 0 ? d : a.daysAgo - b.daysAgo;
  });
  return {
    flagged: flagged.slice(0, LABS_MAX_FLAGGED),
    flaggedCount: flagged.length,
  };
}

/**
 * v1.22 — preventive-care (Vorsorge) due + overdue read-side. Reads the
 * user's enabled, live reminders and buckets by the server-authoritative
 * `nextDueAt`. Returns `undefined` when nothing is due or overdue.
 *
 * `daysUntil` / `daysOverdue` are CALENDAR days in `timeZone`, counted by the
 * same `calendarDaysUntil` the checkup screens phrase their due line with.
 * They used to be an hour gap divided by 24, which is a different question:
 * a checkup at 09:00 read that evening came out "overdue by 0 days", and one
 * due tomorrow morning read late tonight came out due "in 0 days" — today's.
 * The model narrates these numbers, so it repeated whichever the arithmetic
 * had drifted to while the screen next to it said something else.
 *
 * `timeZone` is the user's profile zone; the horizon that bounds the read
 * stays an hours-based cutoff on purpose — it is a query bound, not a
 * statement to anyone about which day something falls on.
 */
export async function readPreventiveCareBlock(
  userId: string,
  now: number,
  timeZone: string,
): Promise<AggregatedFeatures["preventiveCare"] | undefined> {
  const horizon = new Date(now + PREVENTIVE_DUE_HORIZON_DAYS * MS_PER_DAY);
  const rows = await prisma.measurementReminder.findMany({
    // A booked visit's one-shot reminder rides this same engine with
    // `origin: ENCOUNTER`. It is not a checkup and must not appear on a
    // Vorsorge surface, or the list fills with appointments. Four read
    // sites carry this exclusion; `encounter-reminder-exclusion.test.ts`
    // proves every one of them, and the DTO mapper refuses such a row
    // outright so a site that lost its filter fails loudly.
    where: {
      userId,
      deletedAt: null,
      enabled: true,
      origin: { not: "ENCOUNTER" },
      nextDueAt: { not: null, lte: horizon },
    },
    orderBy: { nextDueAt: "asc" },
    take: (PREVENTIVE_MAX_PER_BUCKET + 1) * 4,
    select: { label: true, nextDueAt: true },
  });
  if (rows.length === 0) return undefined;

  const overdue: NonNullable<AggregatedFeatures["preventiveCare"]>["overdue"] =
    [];
  const due: NonNullable<AggregatedFeatures["preventiveCare"]>["due"] = [];
  const nowDate = new Date(now);
  for (const r of rows) {
    if (!r.nextDueAt) continue;
    const label = sanitizeLabel(r.label);
    if (!label) continue;
    const days = calendarDaysUntil(r.nextDueAt, nowDate, timeZone);
    // Today's checkup is due, not overdue — the hour it was booked for may
    // have passed, but the day it belongs to has not.
    if (days < 0) {
      overdue.push({ label, daysOverdue: -days });
    } else {
      due.push({ label, daysUntil: days });
    }
  }
  if (overdue.length === 0 && due.length === 0) return undefined;
  return {
    overdue: overdue.slice(0, PREVENTIVE_MAX_PER_BUCKET),
    due: due.slice(0, PREVENTIVE_MAX_PER_BUCKET),
  };
}

/**
 * v1.22 — workout aggregate over the trailing window. Provider-agnostic:
 * counts + summed duration + summed distance (km, when any source reported it)
 * over 7 / 30 days, plus the latest workout. Returns `undefined` when no
 * workouts fall in the window.
 */
export async function readWorkoutsBlock(
  userId: string,
  now: number,
): Promise<AggregatedFeatures["workouts"] | undefined> {
  const since = new Date(now - WORKOUT_WINDOW_DAYS * MS_PER_DAY);
  const [account, stored] = await Promise.all([
    prisma.user.findUnique({
      where: { id: userId },
      select: { sourcePriorityJson: true },
    }),
    readWorkoutsAroundRange({ from: since, to: null }, (bounds) =>
      prisma.workout.findMany({
        where: { userId, startedAt: bounds },
        orderBy: [{ startedAt: "desc" }, { id: "asc" }],
        take: 2000,
        select: {
          id: true,
          source: true,
          sportType: true,
          startedAt: true,
          durationSec: true,
          totalDistanceM: true,
        },
      }),
    ),
  ]);
  // One session synced by two services (or re-sent by HealthKit) is one
  // workout: the same collapse the workout list applies, so the count and the
  // minutes here match what the person sees there. The picker returns its rows
  // oldest first; the block reads the newest at index 0.
  const rows = canonicalWorkoutsInRange(
    stored,
    account?.sourcePriorityJson ?? null,
  ).reverse();
  if (rows.length === 0) return undefined;

  const tally = (windowDays: number) => {
    const cutoff = now - windowDays * MS_PER_DAY;
    let count = 0;
    let durationSec = 0;
    let distanceM = 0;
    let anyDistance = false;
    for (const w of rows) {
      if (w.startedAt.getTime() < cutoff) continue;
      count += 1;
      durationSec += w.durationSec;
      if (w.totalDistanceM != null) {
        distanceM += w.totalDistanceM;
        anyDistance = true;
      }
    }
    return {
      count,
      totalDurationMin: Math.round(durationSec / 60),
      totalDistanceKm: anyDistance
        ? Math.round((distanceM / 1000) * 10) / 10
        : null,
    };
  };

  const newest = rows[0];
  const latest = {
    sportType: sanitizeLabel(newest.sportType, 40),
    daysAgo: Math.round((now - newest.startedAt.getTime()) / MS_PER_DAY),
    durationMin: Math.round(newest.durationSec / 60),
    distanceKm:
      newest.totalDistanceM != null
        ? Math.round((newest.totalDistanceM / 1000) * 10) / 10
        : null,
  };

  return { last7: tally(7), last30: tally(30), latest };
}

/**
 * ECG recording descriptor for the briefing narrative (device-verdict
 * only). A bounded, non-diagnostic snapshot of the user's on-device ECG
 * recordings over the trailing window: how many exist, the distribution of the
 * RECORDING DEVICE's OWN verdicts (never HealthLog's), and the latest
 * recording's device verdict + average heart rate. The waveform is NEVER read
 * (no decrypt on this path) and NEVER enters the payload — the trace does not
 * cross into the prompt, so the model cannot interpret morphology. The prompt
 * (insight-generator rule) constrains any mention to the device's attribution;
 * the grounding gate constrains any restated count. Returns `undefined` when no
 * recordings fall in the window, so the block is omitted rather than emitting an
 * empty shape.
 */
export async function readEcgBriefingBlock(
  userId: string,
  now: number,
): Promise<AggregatedFeatures["ecg"] | undefined> {
  const since = new Date(now - ECG_WINDOW_DAYS * MS_PER_DAY);
  const rows = await prisma.ecgRecording.findMany({
    where: { userId, recordedAt: { gte: since } },
    orderBy: { recordedAt: "desc" },
    take: ECG_MAX_ROWS,
    // Descriptors only — the encrypted waveform blob is never selected, so no
    // decrypt happens and no trace can reach the narrative.
    select: {
      recordedAt: true,
      rhythmClassification: true,
      averageHeartRate: true,
    },
  });
  if (rows.length === 0) return undefined;

  const deviceVerdicts = { irregular: 0, notDetected: 0, inconclusive: 0 };
  for (const r of rows) {
    if (r.rhythmClassification === "IRREGULAR") deviceVerdicts.irregular += 1;
    else if (r.rhythmClassification === "NOT_DETECTED")
      deviceVerdicts.notDetected += 1;
    else if (r.rhythmClassification === "INCONCLUSIVE")
      deviceVerdicts.inconclusive += 1;
  }

  // An ECG row only ever carries an AFib-screening verdict; the shared enum's
  // other members (walking-steadiness / event codes) never apply here, so
  // anything else narrows to null.
  const latest = rows[0];
  const latestVerdict = latest.rhythmClassification;
  return {
    recordingCount: rows.length,
    deviceVerdicts,
    latestDeviceVerdict:
      latestVerdict === "IRREGULAR" ||
      latestVerdict === "NOT_DETECTED" ||
      latestVerdict === "INCONCLUSIVE"
        ? latestVerdict
        : null,
    latestRecordedDaysAgo: Math.round(
      (now - latest.recordedAt.getTime()) / MS_PER_DAY,
    ),
    latestAverageHeartRate: latest.averageHeartRate,
  };
}

/**
 * Read every BUCKETED_TYPES × BUCKET_WINDOWS combination from the
 * persistent rollup table and project to the wire shape. Empty
 * (type, granularity) combinations are dropped so the payload never
 * carries a labelled-but-empty series.
 */
export async function readBucketedSeries(
  userId: string,
  now: number,
): Promise<BucketedSeries[]> {
  const series: BucketedSeries[] = [];
  for (const type of BUCKETED_TYPES) {
    for (const window of BUCKET_WINDOWS) {
      const from = new Date(now - window.toDays * 24 * 60 * 60 * 1000);
      const to = new Date(now - window.fromDays * 24 * 60 * 60 * 1000);
      const rows = await readRollupBuckets(
        userId,
        type,
        window.granularity,
        from,
        to,
      );
      if (rows.length === 0) continue;
      series.push({
        type,
        granularity: window.granularity,
        buckets: rows.map((r) => ({
          bucketStart: r.bucketStart.toISOString(),
          mean: Math.round(r.mean * 100) / 100,
          count: r.count,
        })),
      });
    }
  }
  return series;
}