"""Bounded read-only data API provider for Dashboard v5 Sprint 6B."""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import math
import os
import re
import secrets
import sqlite3
import time
from statistics import median
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs
from zoneinfo import ZoneInfo

from apple_health_analytics import daily_points
from dashboard_v5.contracts import nutrition_queue_key
from dashboard_v5.data_provider import (
    ADMINISTERED,
    SYMPTOM_LABELS,
    parse_day,
    parse_score,
    public_text,
)
from dashboard_v5.document_chunks import (
    decode_chunk_cursor,
    document_chunks,
    encode_chunk_cursor,
)
from dashboard_v5.document_originals import probe_original
from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, LAB_ALLOWLIST
from dashboard_v5.metric_catalog_v2 import (
    APPLE_ANALYTICS_UNITS,
    APPLE_SOURCE_SPECS,
    BY_ID_V2,
    LAB_CATALOG_SPECS,
    MetricV2,
    SourceInventoryLimitError,
    catalog_search,
    inventory_sources,
    public_catalog,
)

TZ_NAME = "Europe/Zurich"
LOCAL_TZ = ZoneInfo(TZ_NAME)
MAX_RANGE_DAYS = 3660
MAX_SERIES_ROWS = 3660
MAX_RAW_APPLE_ROWS = 100000
MAX_RAW_LAB_ROWS = 10000
MAX_EVENT_SOURCE_ROWS = 10000
MAX_DAILY_SOURCE_ROWS = 30000
MAX_EVENT_ROWS = 1000
MAX_LAB_ROWS = 100
MAX_SEARCH_ROWS_PER_GROUP = 20
MAX_CALENDAR_DAYS = 62
MAX_NUTRITION_DAYS = 180
MAX_NUTRITION_ITEMS = 200
MAX_NUTRITION_QUEUE = 50
MAX_FUTURE_DAYS = 366
MAX_QUERY_BYTES = 512
QUERY_TIMEOUT_SECONDS = 1.0
ALLOWED_RESOLUTIONS = frozenset({"day", "week"})
ALLOWED_EVENT_TYPES = frozenset(
    {"medication_administered", "symptom_day", "health_event", "health_period"}
)
NUTRIENT_ALLOWLIST = {
    "energy.energy": ("Energie", "kcal"),
    "nutrient.protein": ("Protein", "g"),
    "nutrient.carb": ("Kohlenhydrate", "g"),
    "nutrient.fat": ("Fett", "g"),
    "nutrient.fiber": ("Ballaststoffe", "g"),
    "nutrient.sugar": ("Zucker", "g"),
    "nutrient.saturated": ("Gesättigte Fettsäuren", "g"),
    "nutrient.salt": ("Salz", "g"),
    "nutrient.sodium": ("Natrium", "mg"),
}
MEAL_LABELS = {
    "breakfast": "Frühstück",
    "lunch": "Mittagessen",
    "dinner": "Abendessen",
    "snack": "Snack",
    "unassigned": "Nicht zugeordnet",
}
FORBIDDEN_METADATA_TEXT = re.compile(
    r"(?:/home/|/tmp/|/Users/|/(?:var|etc|usr|opt|srv|root|run|mnt|media|private)/|"
    r"~/|file:|https?://|[A-Za-z]:[\\/]|"
    r"(?:^|[\s:|])/(?:[^/\s]+/)+[^/\s]+|\.\.[\\/]|\\\\[^\\/\s]+[\\/]|"
    r"\.hermes(?:/|$))",
    re.IGNORECASE,
)
DRIVE_ID_TOKEN = re.compile(r"(?<![A-Za-z0-9_-])[A-Za-z0-9_-]{25,}(?![A-Za-z0-9_-])")
API_PATHS = frozenset(
    {
        "/api/v1/metric-catalog",
        "/api/v1/series",
        "/api/v1/events",
        "/api/v1/labs",
        "/api/v1/calendar",
        "/api/v1/search",
        "/api/v1/nutrition/days",
        "/api/v1/nutrition/mapping-queue",
    }
)
EXPECTED_ANALYTICS_UNITS = APPLE_ANALYTICS_UNITS


@dataclass
class APIError(Exception):
    status: int
    code: str

    def __str__(self) -> str:
        return self.code


def safe_metadata_text(
    value: Any, maximum: int, *, allow_empty: bool = False
) -> str | None:
    text = public_text(value, maximum, allow_empty=allow_empty)
    drive_id_like = bool(
        text
        and any(
            any(char.isalpha() for char in token)
            and any(char.isdigit() for char in token)
            for token in DRIVE_ID_TOKEN.findall(text)
        )
    )
    if text is None or FORBIDDEN_METADATA_TEXT.search(text) or drive_id_like:
        return None
    return text


def api_document_id(
    raw_id: Any, document_day: str | None, category: str, institution: str
) -> str:
    material = json.dumps(
        [int(raw_id), document_day, category, institution],
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return f"api-document-{hashlib.sha256(b'health-api-document-v1:' + material).hexdigest()[:24]}"


def api_observation_id(prefix: str, raw_id: Any, *parts: Any) -> str:
    material = json.dumps(
        [int(raw_id), *parts],
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return f"{prefix}-{hashlib.sha256(b'health-api-observation-v1:' + material).hexdigest()[:24]}"


def connect_read_only(database: Path) -> sqlite3.Connection:
    connection = sqlite3.connect(f"{database.resolve().as_uri()}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA query_only=ON")
    connection.execute("PRAGMA trusted_schema=OFF")
    deadline = time.monotonic() + QUERY_TIMEOUT_SECONDS
    connection.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000)
    return connection


def table_exists(connection: sqlite3.Connection, table: str) -> bool:
    return (
        connection.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
        ).fetchone()
        is not None
    )


def parse_query(query: str, allowed: set[str]) -> dict[str, str]:
    if len(query.encode("utf-8")) > MAX_QUERY_BYTES:
        raise APIError(414, "query_too_large")
    try:
        parsed = (
            parse_qs(
                query,
                keep_blank_values=True,
                strict_parsing=True,
                max_num_fields=max(1, len(allowed) + 1),
            )
            if query
            else {}
        )
    except ValueError as error:
        raise APIError(400, "invalid_query") from error
    if not set(parsed) <= allowed:
        raise APIError(400, "unknown_parameter")
    if any(len(values) != 1 for values in parsed.values()):
        raise APIError(400, "duplicate_parameter")
    return {key: values[0] for key, values in parsed.items()}


def parse_iso_day(raw: str, *, code: str = "invalid_date") -> date:
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", raw):
        raise APIError(400, code)
    try:
        return date.fromisoformat(raw)
    except ValueError as error:
        raise APIError(400, code) from error


def local_today() -> date:
    synthetic_day = os.environ.get("HEALTH_DASHBOARD_TEST_TODAY", "").strip()
    if synthetic_day and os.environ.get("HEALTH_DASHBOARD_TEST_INSTANCE_ID", "").strip():
        try:
            return date.fromisoformat(synthetic_day)
        except ValueError as error:
            raise APIError(503, "invalid_test_today") from error
    return datetime.now(LOCAL_TZ).date()


def parse_range(
    params: dict[str, str], *, required: bool = False
) -> tuple[date | None, date | None]:
    raw_start = params.get("from")
    raw_end = params.get("to")
    if bool(raw_start) != bool(raw_end):
        raise APIError(400, "incomplete_range")
    if required and not raw_start:
        raise APIError(400, "range_required")
    if not raw_start:
        return None, None
    start = parse_iso_day(raw_start)
    end = parse_iso_day(raw_end or "")
    if end < start:
        raise APIError(400, "range_reversed")
    if start > local_today() or end > local_today():
        raise APIError(422, "future_range_not_allowed")
    if (end - start).days + 1 > MAX_RANGE_DAYS:
        raise APIError(422, "range_too_large")
    return start, end


def _apple_rows(
    connection: sqlite3.Connection,
    identifier: str,
    start: date | None,
    end: date | None,
    *,
    include_raw_json: bool = False,
) -> list[sqlite3.Row]:
    if include_raw_json:
        sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,
                        file_name,file_hash,raw_json
                 FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL"""
    else:
        sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,file_name,file_hash
                 FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL"""
    parameters: list[Any] = [identifier]
    # Keep the (metric,start_date) index usable with a deliberately coarse raw
    # candidate interval.  It includes one whole calendar day on either side;
    # the upper edge is exclusive so all lexical ISO values on end + 1 are kept.
    # daily_points remains the sole authority for Europe/Zurich assignment and
    # exact start/end filtering—this SQL clause never assigns a health day.
    if start is not None and start > date.min:
        sql += " AND start_date>=?"
        parameters.append((start - timedelta(days=1)).isoformat())
    # date.max cannot represent the exclusive edge after end + 1.  Omitting
    # that impossible upper bound preserves the exact canonical filter while
    # avoiding an arithmetic overflow at the ISO boundary.
    if end is not None and end < date.max - timedelta(days=1):
        sql += " AND start_date<?"
        parameters.append((end + timedelta(days=2)).isoformat())
    sql += " ORDER BY start_date,end_date,source_name,id LIMIT ?"
    parameters.append(MAX_RAW_APPLE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_RAW_APPLE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    return rows


def _numeric(value: float, precision: int, value_type: str) -> int | float:
    rounded = round(float(value), precision)
    if value_type == "integer" or (precision == 0 and rounded.is_integer()):
        return int(rounded)
    return rounded


def _apple_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    rows = _apple_rows(connection, metric.source_identifier, start, end)
    points = daily_points(
        metric.source_identifier,
        mode=metric.aggregation,
        stable_only=False,
        start_date=start.isoformat() if start else None,
        end_date=end.isoformat() if end else None,
        _rows=rows,
    )
    result = []
    for day, point in sorted(points.items()):
        if point.unit != EXPECTED_ANALYTICS_UNITS.get(metric.source_identifier):
            continue
        result.append(
            {
                "date": day,
                "value": _numeric(point.value, metric.precision, metric.value_type),
                "quality": point.quality,
                "source_class": point.source_class,
            }
        )
    return result


def _blood_pressure_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    component = "systolic" if metric.id.endswith(".systolic") else "diastolic"
    by_day: dict[str, tuple[datetime, float]] = {}
    for row in _apple_rows(
        connection, "blood_pressure", start, end, include_raw_json=True
    ):
        try:
            timestamp = datetime.fromisoformat(
                str(row["start_date"]).replace("Z", "+00:00")
            )
            if timestamp.tzinfo is None:
                timestamp = timestamp.replace(tzinfo=LOCAL_TZ)
            day = timestamp.astimezone(LOCAL_TZ).date()
            payload = json.loads(str(row["raw_json"] or ""))
            unit = re.sub(r"[^a-z]", "", str(row["unit"] or "").casefold())
            systolic = float(payload["systolic"])
            diastolic = float(payload["diastolic"])
            if unit != "mmhg" or not all(
                math.isfinite(item) for item in (systolic, diastolic)
            ):
                continue
            value = systolic if component == "systolic" else diastolic
        except (ValueError, TypeError, KeyError, json.JSONDecodeError):
            continue
        if not math.isfinite(value):
            continue
        if start is not None and day < start:
            continue
        if end is not None and day > end:
            continue
        existing = by_day.get(day.isoformat())
        if existing is None or timestamp > existing[0]:
            by_day[day.isoformat()] = (timestamp, value)
    return [
        {
            "date": day,
            "value": _numeric(item[1], metric.precision, metric.value_type),
            "quality": "paired_observation",
            "source_class": "apple_health",
        }
        for day, item in sorted(by_day.items())
    ]


def _symptom_dimension_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    sql = "SELECT datum,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' AND symptom=?"
    parameters: list[Any] = [metric.source_identifier]
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum,id LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[str, list[int | None]] = defaultdict(list)
    for row in rows:
        day = parse_day(row["datum"])
        if day:
            grouped[day].append(parse_score(row["schwergrad"]))
    return [
        {"date": day, "value": scores[0], "quality": "observed"}
        for day, scores in sorted(grouped.items())
        if len(scores) == 1 and scores[0] is not None
    ]


def _symptom_total_points(
    connection: sqlite3.Connection,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    sql = "SELECT datum,symptom,schwergrad FROM symptom_log WHERE kontext='daily_quick_score'"
    parameters: list[Any] = []
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum,id LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[str, dict[str, list[int | None]]] = defaultdict(
        lambda: defaultdict(list)
    )
    for row in rows:
        day = parse_day(row["datum"])
        label = str(row["symptom"] or "")
        if day and label in SYMPTOM_LABELS:
            grouped[day][label].append(parse_score(row["schwergrad"]))
    result = []
    for day, dimensions in sorted(grouped.items()):
        if set(dimensions) != SYMPTOM_LABELS:
            continue
        if not all(
            len(scores) == 1 and scores[0] is not None for scores in dimensions.values()
        ):
            continue
        result.append(
            {
                "date": day,
                "value": sum(int(scores[0]) for scores in dimensions.values()),
                "quality": "complete",
            }
        )
    return result


def _nutrition_points(
    connection: sqlite3.Connection,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    sql = """SELECT datum,histamine_score,item_count,histamine_unknown_count,histamine_label
             FROM nutrition_daily_summary_v2 WHERE item_count>0
               AND histamine_unknown_count=0 AND histamine_score IS NOT NULL
               AND histamine_label IN ('classified','green','yellow','orange','red')"""
    parameters: list[Any] = []
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    result = []
    for row in rows:
        day = parse_day(row["datum"])
        try:
            value = float(row["histamine_score"])
        except (TypeError, ValueError):
            continue
        if day and math.isfinite(value):
            result.append({"date": day, "value": value, "quality": "complete"})
    return result


def _lab_key(parameter: Any, unit: Any) -> tuple[str, str]:
    def slug(value: Any) -> str:
        normalized = (
            str(value or "")
            .casefold()
            .replace("µ", "u")
            .replace("μ", "u")
            .replace("%", " percent ")
        )
        return re.sub(r"[^a-z0-9]+", "_", normalized).strip("_")

    return slug(parameter), slug(unit)


def _verified_lab_rows(
    connection: sqlite3.Connection,
    *,
    include_missing_reference: bool = False,
    start: date | None = None,
    end: date | None = None,
) -> list[dict[str, Any]]:
    if not table_exists(connection, "laborwerte"):
        return []
    date_expression = "COALESCE(NULLIF(l.abnahme_datum,''),l.befund_datum)"
    filters = [
        "lower(trim(COALESCE(l.validierungsstatus,'')))='validiert'",
        "l.verified_against_original=1",
        "l.reference_range_source='scanned_original'",
        "trim(COALESCE(l.einheit,''))<>''",
    ]
    values: list[Any] = []
    if start:
        filters.append(f"{date_expression}>=?")
        values.append(start.isoformat())
    if end:
        filters.append(f"{date_expression}<=?")
        values.append(end.isoformat())
    rows = list(
        connection.execute(
            """SELECT l.id,l.parameter_name,l.wert,l.einheit,l.reference_min,l.reference_max,
                  l.abnahme_datum,l.befund_datum,l.reference_range_source,l.source_type,
                  l.dokument_id,l.canonical_document_id,d.id AS linked_document_record_id,
                  d.document_date,d.kategorie,d.institution,d.review_status
               FROM laborwerte l LEFT JOIN dokumente d ON d.id=l.canonical_document_id
               WHERE """
            + " AND ".join(filters)
            + f" ORDER BY {date_expression},l.id LIMIT ?",
            (*values, MAX_RAW_LAB_ROWS + 1),
        )
    )
    if len(rows) > MAX_RAW_LAB_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[tuple[str, str, str], list[sqlite3.Row]] = defaultdict(list)
    for row in rows:
        canonical = LAB_ALLOWLIST.get(_lab_key(row["parameter_name"], row["einheit"]))
        day = parse_day(row["abnahme_datum"]) or parse_day(row["befund_datum"])
        if canonical and day and date.fromisoformat(day) <= local_today():
            grouped[(canonical[0], canonical[1], day)].append(row)
    result: list[dict[str, Any]] = []
    for (parameter, unit, day), same_day in sorted(grouped.items()):
        if len(same_day) != 1:
            continue
        row = same_day[0]
        raw_value = ("" if row["wert"] is None else str(row["wert"])).strip()
        if not EXACT_LAB_NUMBER.fullmatch(raw_value):
            continue
        numeric = float(raw_value.replace(",", "."))
        if not math.isfinite(numeric):
            continue
        value: int | float = int(numeric) if numeric.is_integer() else numeric
        reference_min = str(row["reference_min"] or "").strip()
        reference_max = str(row["reference_max"] or "").strip()
        reference_min = (
            reference_min
            if EXACT_LAB_NUMBER.fullmatch(reference_min)
            and math.isfinite(float(reference_min.replace(",", ".")))
            else None
        )
        reference_max = (
            reference_max
            if EXACT_LAB_NUMBER.fullmatch(reference_max)
            and math.isfinite(float(reference_max.replace(",", ".")))
            else None
        )
        item: dict[str, Any] = {
            "id": api_observation_id("lab-observation", row["id"], parameter, unit, day),
            "metric_id": LAB_CATALOG_SPECS[parameter].metric_id,
            "parameter": parameter,
            "value": value,
            "unit": unit,
            "date": day,
            "quality": "verified_original",
            "document_link_status": (
                "linked" if row["linked_document_record_id"] is not None else "unlinked"
            ),
            "document_review_status": (
                "reviewed"
                if row["linked_document_record_id"] is not None
                and row["review_status"] == "geprueft"
                else "review_pending"
                if row["linked_document_record_id"] is not None
                else "not_linked"
            ),
        }
        if reference_min is None and reference_max is None:
            if not include_missing_reference:
                continue
            item["reference_status"] = "missing_reference"
        else:
            item["reference"] = {
                "min": reference_min,
                "max": reference_max,
                "source": "scanned_original",
            }
            item["reference_status"] = "verified_observation_specific"
        if (
            row["canonical_document_id"] is not None
            and row["review_status"] == "geprueft"
        ):
            category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
            institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
            document_day = parse_day(row["document_date"])
            if category is not None and institution is not None:
                item["document"] = {
                    "id": api_document_id(
                        row["canonical_document_id"],
                        document_day,
                        category,
                        institution,
                    ),
                    "date": document_day,
                    "category": category,
                    "institution": institution,
                }
        result.append(item)
    return result


def _metric_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    if metric.value_type == "panel":
        raise APIError(400, "metric_not_series")
    if metric.source == "apple_health":
        if metric.source_identifier == "blood_pressure":
            return _blood_pressure_points(connection, metric, start, end)
        return _apple_points(connection, metric, start, end)
    if metric.id == "symptom.total":
        return _symptom_total_points(connection, start, end)
    if metric.id == "symptom.aphthae":
        return _symptom_dimension_points(connection, metric, start, end)
    if metric.id == "nutrition.histamine":
        return _nutrition_points(connection, start, end)
    if metric.source == "laborwerte":
        return [
            {
                key: row[key]
                for key in ("date", "value", "quality", "reference", "reference_status")
                if key in row
            }
            for row in _verified_lab_rows(connection, include_missing_reference=True)
            if row["metric_id"] == metric.id
            and (start is None or row["date"] >= start.isoformat())
            and (end is None or row["date"] <= end.isoformat())
        ]
    return []


def _weekly_rule(metric: MetricV2) -> dict[str, Any]:
    if metric.aggregation == "observation" or metric.value_type == "panel":
        return {
            "id": "unsupported_for_metric",
            "minimum_observations": None,
            "requires_complete_calendar_week": False,
        }
    if metric.expected_frequency == "daily":
        return {
            "id": "complete_calendar_week_daily_metric",
            "minimum_observations": 7,
            "requires_complete_calendar_week": True,
        }
    if metric.expected_frequency == "intermittent":
        return {
            "id": "observed_measurements_within_iso_week",
            "minimum_observations": 1,
            "requires_complete_calendar_week": False,
        }
    return {
        "id": "unsupported_for_metric",
        "minimum_observations": None,
        "requires_complete_calendar_week": False,
    }


def _add_past_only_baselines(points: list[dict[str, Any]], metric: MetricV2) -> None:
    rule = metric.baseline_rule
    if not rule:
        return
    window = int(rule["window_days"])
    minimum = int(rule["minimum_observations"])
    for point in points:
        day = date.fromisoformat(point["date"])
        prior = [
            float(candidate["value"])
            for candidate in points
            if date.fromisoformat(candidate["date"]) < day
            and (day - date.fromisoformat(candidate["date"])).days <= window
        ]
        point["baseline"] = (
            _numeric(float(median(prior)), metric.precision, metric.value_type)
            if len(prior) >= minimum
            else None
        )


def _weekly(
    points: list[dict[str, Any]], metric: MetricV2, rule: dict[str, Any]
) -> list[dict[str, Any]]:
    if rule["id"] == "unsupported_for_metric":
        return []
    grouped: dict[tuple[int, int], list[dict[str, Any]]] = defaultdict(list)
    for point in points:
        day = date.fromisoformat(point["date"])
        iso = day.isocalendar()
        grouped[(iso.year, iso.week)].append(point)
    result = []
    for (year, week), rows in sorted(grouped.items()):
        distinct = {row["date"] for row in rows}
        if rule["requires_complete_calendar_week"] and len(distinct) != 7:
            continue
        values = [float(row["value"]) for row in rows]
        if metric.aggregation == "sum":
            value = sum(values)
        elif metric.aggregation == "last":
            value = float(max(rows, key=lambda row: row["date"])["value"])
        else:
            value = sum(values) / len(values)
        week_start = date.fromisocalendar(year, week, 1)
        week_end = week_start + timedelta(days=6)
        item = {
            "week": f"{year}-W{week:02d}",
            "date": min(distinct),
            "week_start": week_start.isoformat(),
            "week_end": week_end.isoformat(),
            "drilldown_date": week_start.isoformat(),
            "value": _numeric(value, metric.precision, metric.value_type),
            "observation_count": len(rows),
            "observed_days": len(distinct),
            "quality": (
                "complete_calendar_week"
                if rule["requires_complete_calendar_week"]
                else "observed_measurements"
            ),
        }
        result.append(item)
    return result


def _coverage(
    points: list[dict[str, Any]], start: date | None, end: date | None
) -> dict[str, Any]:
    observed = sorted({date.fromisoformat(point["date"]) for point in points})
    effective_start = start or (observed[0] if observed else None)
    effective_end = end or (observed[-1] if observed else None)
    if effective_start is not None and effective_end is not None:
        span = (effective_end - effective_start).days + 1
        if span > MAX_RANGE_DAYS:
            raise APIError(422, "range_too_large")
    else:
        span = 0
    return {
        "from": effective_start.isoformat() if effective_start else None,
        "to": effective_end.isoformat() if effective_end else None,
        "expected_days": span,
        "observed_days": len(observed),
        "observation_count": len(points),
        "missing_days": max(0, span - len(observed)),
        "gaps": "omitted_not_interpolated",
    }


def _series(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    metric_id = params.get("metric", "")
    metric = BY_ID_V2.get(metric_id)
    if metric is None:
        raise APIError(404, "metric_not_allowed")
    resolution = params.get("resolution", "day")
    if resolution not in ALLOWED_RESOLUTIONS:
        raise APIError(400, "resolution_not_allowed")
    start, end = parse_range(params)
    points = _metric_points(connection, metric, start, end)
    today = local_today().isoformat()
    points = [point for point in points if point["date"] <= today]
    _add_past_only_baselines(points, metric)
    if start is None and points:
        observed_start = date.fromisoformat(points[0]["date"])
        observed_end = date.fromisoformat(points[-1]["date"])
        if (observed_end - observed_start).days + 1 > MAX_RANGE_DAYS:
            raise APIError(422, "range_too_large")
    coverage = _coverage(points, start, end)
    if end is None:
        coverage["to"] = today
        if coverage["from"] is not None:
            coverage["expected_days"] = (
                date.fromisoformat(today) - date.fromisoformat(coverage["from"])
            ).days + 1
            coverage["missing_days"] = max(
                0, coverage["expected_days"] - coverage["observed_days"]
            )
    aggregation_rule = {
        "id": "daily_metric_points",
        "minimum_observations": 1,
        "requires_complete_calendar_week": False,
    }
    if resolution == "week":
        aggregation_rule = _weekly_rule(metric)
        points = _weekly(points, metric, aggregation_rule)
    if len(points) > MAX_SERIES_ROWS:
        raise APIError(422, "row_limit_exceeded")
    return {
        "metric": metric.id,
        "label": metric.label,
        "unit": metric.unit,
        "aggregation": metric.aggregation,
        "aggregation_rule": aggregation_rule,
        "resolution": resolution,
        "source": {
            "type": metric.source,
            "identifier": metric.source_identifier,
            "parser": metric.parser_contract,
        },
        "coverage": coverage,
        "timezone": TZ_NAME,
        "today": today,
        "points": points,
    }


def _events(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    start, end = parse_range(params, required=True)
    assert start is not None and end is not None
    raw_types = params.get("types", "")
    if not raw_types:
        selected = set(ALLOWED_EVENT_TYPES)
    else:
        selected = set(raw_types.split(","))
        if "" in selected or not selected <= ALLOWED_EVENT_TYPES:
            raise APIError(400, "event_type_not_allowed")
    result: list[dict[str, Any]] = []
    if "medication_administered" in selected and table_exists(
        connection, "medication_administrations"
    ):
        administered_values = tuple(sorted(ADMINISTERED))
        if len(administered_values) != 6:
            raise APIError(503, "api_contract_invalid")
        rows = list(
            connection.execute(
                """SELECT datum,medication_name,event_type FROM medication_administrations
               WHERE datum>=? AND datum<=?
                 AND lower(trim(COALESCE(event_type,''))) IN (?,?,?,?,?,?)
               ORDER BY datum,id LIMIT ?""",
                (
                    start.isoformat(),
                    end.isoformat(),
                    *administered_values,
                    MAX_EVENT_SOURCE_ROWS + 1,
                ),
            )
        )
        if len(rows) > MAX_EVENT_SOURCE_ROWS:
            raise APIError(422, "source_row_limit_exceeded")
        for row in rows:
            label = safe_metadata_text(row["medication_name"], 120)
            day = parse_day(row["datum"])
            if day and label:
                result.append(
                    {"date": day, "type": "medication_administered", "label": label}
                )
    if "symptom_day" in selected and table_exists(connection, "symptom_log"):
        for point in _symptom_total_points(connection, start, end):
            result.append(
                {
                    "date": point["date"],
                    "type": "symptom_day",
                    "label": "Symptome vollständig dokumentiert",
                }
            )
    if "health_event" in selected and table_exists(connection, "health_events"):
        rows = list(
            connection.execute(
                "SELECT date,category,parameter FROM health_events WHERE date>=? AND date<=? ORDER BY date,id LIMIT ?",
                (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1),
            )
        )
        if len(rows) > MAX_EVENT_SOURCE_ROWS:
            raise APIError(422, "source_row_limit_exceeded")
        for row in rows:
            category = safe_metadata_text(row["category"], 80, allow_empty=True)
            parameter = safe_metadata_text(row["parameter"], 120, allow_empty=True)
            day = parse_day(row["date"])
            if day and category is not None and parameter is not None:
                result.append(
                    {
                        "date": day,
                        "type": "health_event",
                        "category": category,
                        "label": parameter,
                    }
                )
    if "health_period" in selected and table_exists(connection, "health_event_periods"):
        rows = list(
            connection.execute(
                """SELECT start_date,end_date,event_type,label FROM health_event_periods
               WHERE start_date<=? AND COALESCE(end_date,start_date)>=?
               ORDER BY start_date,id LIMIT ?""",
                (end.isoformat(), start.isoformat(), MAX_EVENT_SOURCE_ROWS + 1),
            )
        )
        if len(rows) > MAX_EVENT_SOURCE_ROWS:
            raise APIError(422, "source_row_limit_exceeded")
        for row in rows:
            event_type = safe_metadata_text(row["event_type"], 80)
            label = safe_metadata_text(row["label"], 120, allow_empty=True)
            start_day = parse_day(row["start_date"])
            end_day = parse_day(row["end_date"]) if row["end_date"] else None
            if (
                start_day is None
                or (end_day is not None and end_day < start_day)
                or (end_day is not None and date.fromisoformat(end_day) > local_today())
            ):
                continue
            if event_type and label is not None:
                result.append(
                    {
                        "date": start_day,
                        "end_date": end_day,
                        "type": "health_period",
                        "category": event_type,
                        "label": label,
                    }
                )
    result.sort(key=lambda item: (item["date"], item["type"], item.get("label", "")))
    if len(result) > MAX_EVENT_ROWS:
        raise APIError(422, "row_limit_exceeded")
    return {
        "from": start.isoformat(),
        "to": end.isoformat(),
        "types": sorted(selected),
        "timezone": TZ_NAME,
        "events": result,
    }


def _safe_documents(
    connection: sqlite3.Connection, needle: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "dokumente"):
        return []
    result = []
    for row in connection.execute(
        """SELECT id,document_date,kategorie,institution FROM dokumente
           WHERE review_status='geprueft' ORDER BY document_date DESC,id DESC LIMIT 200"""
    ):
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        document_day = parse_day(row["document_date"])
        if category is None or institution is None:
            continue
        haystack = f"{category} {institution} {document_day or ''}".casefold()
        if needle.casefold() not in haystack:
            continue
        result.append(
            {
                "id": api_document_id(row["id"], document_day, category, institution),
                "date": document_day,
                "category": category,
                "institution": institution,
                "drill_down_target": "document_metadata",
            }
        )
        if len(result) >= MAX_SEARCH_ROWS_PER_GROUP:
            break
    return result


def _fts_search_documents(
    connection: sqlite3.Connection, query: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "health_document_fts"):
        return []
    page = _record_document_rows(
        connection,
        {"q": query, "limit": str(MAX_SEARCH_ROWS_PER_GROUP)},
    )
    return [
        {
            "id": row["id"],
            "date": row["document_date"],
            "category": row["category"],
            "institution": row["institution"],
            "snippet": row["snippet"],
            "drill_down_target": "document",
        }
        for row in page["documents"]
    ]


def _search(connection: sqlite3.Connection, query: str) -> dict[str, Any]:
    clean = " ".join(query.split())
    if len(clean) < 2 or len(clean) > 80:
        raise APIError(400, "query_length_invalid")
    matches = catalog_search(connection, clean)
    groups: dict[str, list[dict[str, Any]]] = {
        "metrics": [],
        "laboratory": [],
        "symptoms_events": [],
        "documents": [],
        "days": [],
    }
    for item in matches:
        target = (
            "laboratory"
            if item["category"] == "laboratory"
            else ("symptoms_events" if item["category"] == "symptoms" else "metrics")
        )
        if len(groups[target]) < MAX_SEARCH_ROWS_PER_GROUP:
            groups[target].append(item)
    normalized = clean.casefold()
    if normalized in {"medikamente", "medikament", "medikation", "arznei"}:
        groups["symptoms_events"].append(
            {
                "id": "event.medication",
                "label": "Medikamente",
                "category": "events",
                "drill_down_target": "day",
                "availability": {"status": "supported"},
            }
        )
    groups["documents"] = _fts_search_documents(connection, clean)
    if not groups["documents"]:
        groups["documents"] = _safe_documents(connection, clean)
    try:
        parsed = parse_iso_day(clean)
    except APIError:
        parsed = None
    if parsed is not None and parsed <= local_today():
        groups["days"] = [{"date": parsed.isoformat(), "drill_down_target": "day"}]
    return {"query": clean, "timezone": TZ_NAME, "groups": groups}


def _safe_day_text(
    value: Any, maximum: int = 160, *, allow_empty: bool = True
) -> str | None:
    return safe_metadata_text(value, maximum, allow_empty=allow_empty)


def _day_symptoms(connection: sqlite3.Connection, day: str) -> dict[str, Any]:
    dimensions = {
        label: {"value": None, "note": None, "status": "not_documented"}
        for label in sorted(SYMPTOM_LABELS)
    }
    if table_exists(connection, "symptom_log"):
        rows = list(
            connection.execute(
                "SELECT symptom,schwergrad,notizen FROM symptom_log WHERE datum=? AND kontext='daily_quick_score' ORDER BY id LIMIT 100",
                (day,),
            )
        )
        for row in rows:
            label = str(row["symptom"] or "")
            if label not in dimensions:
                continue
            score = parse_score(row["schwergrad"])
            note = _safe_day_text(row["notizen"], 300)
            dimensions[label] = {
                "value": score,
                "note": note,
                "status": "observed" if score is not None else "not_documented",
            }
    values = [entry["value"] for entry in dimensions.values()]
    complete = all(value is not None for value in values)
    notes = sorted({entry["note"] for entry in dimensions.values() if entry["note"]})
    return {
        "dimensions": [{"name": name, **entry} for name, entry in dimensions.items()],
        "total": sum(values) if complete else None,
        "complete": complete,
        "documented_dimensions": sum(value is not None for value in values),
        "expected_dimensions": len(dimensions),
        "notes": notes,
    }


def _day_medications(
    connection: sqlite3.Connection, requested: str
) -> dict[str, list[dict[str, Any]]]:
    result = {"planned": [], "administered": [], "missed": [], "corrected": []}
    if not table_exists(connection, "medication_administrations"):
        return result
    rows = list(
        connection.execute(
            """SELECT datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source
           FROM medication_administrations
           WHERE datum=? OR scheduled_next_date=? ORDER BY id LIMIT 200""",
            (requested, requested),
        )
    )
    missed = {"missed", "verpasst", "ausgelassen"}
    corrected = {"corrected", "korrigiert", "correction"}
    for row in rows:
        event_type = str(row["event_type"] or "").strip().casefold()
        name = _safe_day_text(row["medication_name"], 120, allow_empty=False)
        dose = _safe_day_text(row["dose"], 40)
        route = _safe_day_text(row["route"], 40)
        note = _safe_day_text(row["notes"], 300)
        source = _safe_day_text(row["source"], 80)
        if not name or dose is None or route is None or note is None or source is None:
            continue
        item = {
            "name": name,
            "dose": dose,
            "route": route,
            "note": note,
            "source": source,
        }
        datum = parse_day(row["datum"])
        scheduled = parse_day(row["scheduled_next_date"])
        if datum == requested and event_type in ADMINISTERED:
            result["administered"].append(item)
        elif datum == requested and event_type in missed:
            result["missed"].append(item)
        elif datum == requested and event_type in corrected:
            result["corrected"].append(item)
        elif (
            datum == requested and event_type in {"planned", "scheduled", "geplant"}
        ) or scheduled == requested:
            result["planned"].append(item)
    return result


def _day_events(
    connection: sqlite3.Connection, requested: str
) -> dict[str, list[dict[str, Any]]]:
    events: list[dict[str, Any]] = []
    periods: list[dict[str, Any]] = []
    if table_exists(connection, "health_events"):
        for row in connection.execute(
            "SELECT category,parameter,value,unit,source,notes FROM health_events WHERE date=? ORDER BY id LIMIT 200",
            (requested,),
        ):
            fields = {
                "category": _safe_day_text(row["category"], 80),
                "label": _safe_day_text(row["parameter"], 120),
                "value": _safe_day_text(row["value"], 80),
                "unit": _safe_day_text(row["unit"], 40),
                "source": _safe_day_text(row["source"], 80),
                "note": _safe_day_text(row["notes"], 300),
            }
            if all(value is not None for value in fields.values()):
                events.append(fields)
    if table_exists(connection, "health_event_periods"):
        for row in connection.execute(
            """SELECT start_date,end_date,event_type,label,source,notes
               FROM health_event_periods WHERE start_date<=? AND COALESCE(end_date,start_date)>=?
               ORDER BY start_date,id LIMIT 200""",
            (requested, requested),
        ):
            start = parse_day(row["start_date"])
            end = parse_day(row["end_date"]) if row["end_date"] else start
            event_type = _safe_day_text(row["event_type"], 80, allow_empty=False)
            label = _safe_day_text(row["label"], 120)
            source = _safe_day_text(row["source"], 80)
            note = _safe_day_text(row["notes"], 300)
            if (
                start
                and end
                and start <= requested <= end
                and event_type
                and label is not None
                and source is not None
                and note is not None
            ):
                periods.append(
                    {
                        "start_date": start,
                        "end_date": end,
                        "type": event_type,
                        "label": label,
                        "source": source,
                        "note": note,
                    }
                )
    return {"events": events, "periods": periods}


def _day_documents(
    connection: sqlite3.Connection, requested: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "dokumente"):
        return []
    result = []
    rows = connection.execute(
        """SELECT id,document_date,kategorie,institution FROM dokumente
           WHERE document_date=? AND review_status='geprueft' ORDER BY id LIMIT 100""",
        (requested,),
    )
    for row in rows:
        category = _safe_day_text(row["kategorie"], 80)
        institution = _safe_day_text(row["institution"], 120)
        if category is None or institution is None:
            continue
        result.append(
            {
                "id": api_document_id(row["id"], requested, category, institution),
                "date": requested,
                "category": category,
                "institution": institution,
            }
        )
    return result


def _day_appointments(connection: sqlite3.Connection, requested: str) -> dict[str, Any]:
    if not table_exists(connection, "arztbesuche"):
        return {"status": "supported_no_data", "items": []}
    items = []
    for row in connection.execute(
        "SELECT arzt,klinik,grund,notizen FROM arztbesuche WHERE datum=? ORDER BY id LIMIT 100",
        (requested,),
    ):
        practitioner = _safe_day_text(row["arzt"], 120)
        institution = _safe_day_text(row["klinik"], 120)
        reason = _safe_day_text(row["grund"], 160)
        note = _safe_day_text(row["notizen"], 300)
        if all(
            value is not None for value in (practitioner, institution, reason, note)
        ):
            items.append(
                {
                    "date": requested,
                    "practitioner": practitioner,
                    "institution": institution,
                    "specialty_or_reason": reason,
                    "note": note,
                    # arztbesuche currently has no explicit status column.  Its date
                    # documents that an entry exists, but it never proves attendance,
                    # cancellation, or completion; retain that distinction explicitly.
                    "status": "documented_visit",
                }
            )
    return {"status": "supported" if items else "supported_no_data", "items": items}


def _day_nutrition(connection: sqlite3.Connection, requested: str) -> dict[str, Any]:
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {"status": "supported_no_data", "date": requested}
    row = connection.execute(
        """SELECT kcal,protein_g,carb_g,fat_g,item_count,histamine_score,
                  histamine_unknown_count,histamine_label
           FROM nutrition_daily_summary_v2 WHERE datum=? LIMIT 1""",
        (requested,),
    ).fetchone()
    if row is None:
        return {"status": "supported_no_data", "date": requested}
    complete = (
        int(row["item_count"] or 0) > 0
        and int(row["histamine_unknown_count"] or 0) == 0
        and row["histamine_score"] is not None
    )
    return {
        "status": "documented",
        "date": requested,
        "kcal": row["kcal"],
        "protein_g": row["protein_g"],
        "carb_g": row["carb_g"],
        "fat_g": row["fat_g"],
        "item_count": row["item_count"],
        "histamine_score": row["histamine_score"] if complete else None,
        "histamine_status": row["histamine_label"] if complete else "not_documented",
        "unknown_items": row["histamine_unknown_count"],
        "complete": complete,
    }


def _nutrition_complete(row: sqlite3.Row) -> bool:
    return (
        int(row["item_count"] or 0) > 0
        and int(row["histamine_unknown_count"] or 0) == 0
        and row["histamine_score"] is not None
        and row["histamine_label"] in {"classified", "green", "yellow", "orange", "red"}
    )


def _public_number(value: Any) -> int | float | None:
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        return None
    if not math.isfinite(numeric):
        return None
    rounded = round(numeric, 3)
    return int(rounded) if rounded.is_integer() else rounded


def _nutrition_days(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    start, end = parse_range(params)
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {"version": 1, "timezone": TZ_NAME, "definition": _histamine_definition(), "days": []}
    sql = """SELECT datum,kcal,protein_g,carb_g,fat_g,item_count,histamine_score,
                    histamine_max,histamine_unknown_count,histamine_label
             FROM nutrition_daily_summary_v2 WHERE datum<=?"""
    values: list[Any] = [local_today().isoformat()]
    if start:
        sql += " AND datum>=?"
        values.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        values.append(end.isoformat())
    sql += " ORDER BY datum DESC LIMIT ?"
    values.append(MAX_NUTRITION_DAYS + 1)
    rows = list(connection.execute(sql, values))
    if len(rows) > MAX_NUTRITION_DAYS:
        raise APIError(422, "row_limit_exceeded")
    days = []
    for row in reversed(rows):
        day = parse_day(row["datum"])
        if not day:
            continue
        complete = _nutrition_complete(row)
        days.append({
            "date": day,
            "macros": {
                "kcal": _public_number(row["kcal"]),
                "protein_g": _public_number(row["protein_g"]),
                "carb_g": _public_number(row["carb_g"]),
                "fat_g": _public_number(row["fat_g"]),
            },
            "item_count": int(row["item_count"] or 0),
            "histamine": {
                "score": _public_number(row["histamine_score"]) if complete else None,
                "max_score": int(row["histamine_max"] or 0) if complete else None,
                "mapping_coverage": round((int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)) / int(row["item_count"] or 1), 3) if int(row["item_count"] or 0) else 0,
                "unknown_items": int(row["histamine_unknown_count"] or 0),
                "status": "complete" if complete else "incomplete",
                "definition_version": _histamine_definition()["version"],
            },
        })
    return {"version": 1, "timezone": TZ_NAME, "definition": _histamine_definition(), "days": days}


def _histamine_definition() -> dict[str, Any]:
    return {
        "id": "sighi_mapping_load_v1",
        "version": "sighi_mapping_load_v2",
        "components": ["mapped_item_score_sum", "mapped_item_score_max", "mapping_coverage", "unknown_item_count"],
        "coverage_semantics": "unknown_or_unmapped_items_make_day_incomplete",
        "insufficient_inputs": "no_index",
        "score_scale": "SIGHi 0-3 or unknown per mapped item",
        "complete_rule": "complete only when every documented nutrition item has a mapping",
    }


def _nutrition_day_detail(connection: sqlite3.Connection, requested: date) -> dict[str, Any]:
    day = requested.isoformat()
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {"version": 1, "date": day, "timezone": TZ_NAME, "status": "supported_no_data"}
    row = connection.execute(
        """SELECT datum,kcal,protein_g,carb_g,fat_g,item_count,nutrient_json,
                  histamine_score,histamine_max,histamine_unknown_count,histamine_label
           FROM nutrition_daily_summary_v2 WHERE datum=? LIMIT 1""",
        (day,),
    ).fetchone()
    if row is None:
        return {"version": 1, "date": day, "timezone": TZ_NAME, "status": "supported_no_data"}
    complete = _nutrition_complete(row)
    nutrients: dict[str, dict[str, Any]] = {}
    nutrient_values: dict[str, int | float | None] = {}
    if table_exists(connection, "nutrition_item_nutrients"):
        for nutrient in connection.execute(
            """SELECT n.nutrient_key,n.value,n.unit
               FROM nutrition_item_nutrients n JOIN nutrition_items i ON i.id=n.item_id
               WHERE i.datum=?""",
            (day,),
        ):
            key = str(nutrient["nutrient_key"] or "")
            allowed = NUTRIENT_ALLOWLIST.get(key)
            if not allowed or str(nutrient["unit"] or "") != allowed[1]:
                continue
            value = _public_number(nutrient["value"])
            if value is None:
                continue
            nutrient_values[key] = _public_number(float(nutrient_values.get(key) or 0) + float(value))
    else:
        try:
            raw_nutrients = json.loads(row["nutrient_json"] or "{}")
        except json.JSONDecodeError:
            raw_nutrients = {}
        if isinstance(raw_nutrients, dict):
            for key in NUTRIENT_ALLOWLIST:
                value = _public_number(raw_nutrients.get(key))
                if value is not None:
                    nutrient_values[key] = value
    for key, (label, unit) in NUTRIENT_ALLOWLIST.items():
        if key in nutrient_values:
            nutrients[key] = {"label": label, "value": nutrient_values[key], "unit": unit}
    meals = []
    if table_exists(connection, "nutrition_meal_summary"):
        for meal in connection.execute(
            """SELECT meal,kcal,protein_g,carb_g,fat_g,item_count,histamine_score,histamine_label
               FROM nutrition_meal_summary WHERE datum=?
               ORDER BY CASE meal WHEN 'breakfast' THEN 1 WHEN 'lunch' THEN 2 WHEN 'dinner' THEN 3 WHEN 'snack' THEN 4 ELSE 9 END LIMIT 12""",
            (day,),
        ):
            meal_id = str(meal["meal"] or "unknown")
            meals.append({
                "meal": meal_id if meal_id in MEAL_LABELS else "unknown",
                "label": MEAL_LABELS.get(meal_id, "Unbekannt"),
                "item_count": int(meal["item_count"] or 0),
                "macros": {
                    "kcal": _public_number(meal["kcal"]),
                    "protein_g": _public_number(meal["protein_g"]),
                    "carb_g": _public_number(meal["carb_g"]),
                    "fat_g": _public_number(meal["fat_g"]),
                },
                "histamine": {
                    "score": _public_number(meal["histamine_score"]) if meal["histamine_label"] in {"classified", "green", "yellow", "orange", "red"} else None,
                    "status": "complete" if meal["histamine_label"] in {"classified", "green", "yellow", "orange", "red"} else "incomplete",
                },
            })
    existing_meals = {meal["meal"] for meal in meals}
    for meal_id, label in MEAL_LABELS.items():
        if meal_id not in existing_meals:
            meals.append({"meal": meal_id, "label": label, "item_count": 0, "macros": {"kcal": None, "protein_g": None, "carb_g": None, "fat_g": None}, "histamine": {"score": None, "status": "incomplete"}})
    meals.sort(key=lambda item: ["breakfast", "lunch", "dinner", "snack", "unassigned"].index(item["meal"]))
    items = []
    if table_exists(connection, "nutrition_items"):
        for item in connection.execute(
            """SELECT i.meal,i.name,i.brand,i.amount,i.amount_unit,i.kcal,i.protein_g,i.carb_g,i.fat_g,
                      h.canonical_food,h.sighi_score,h.traffic_light,h.confidence
               FROM nutrition_items i LEFT JOIN nutrition_histamine_scores h ON h.item_id=i.id
               WHERE i.datum=? ORDER BY i.meal,i.id LIMIT ?""",
            (day, MAX_NUTRITION_ITEMS + 1),
        ):
            if len(items) >= MAX_NUTRITION_ITEMS:
                raise APIError(422, "row_limit_exceeded")
            name = safe_metadata_text(item["name"], 120, allow_empty=False)
            brand = safe_metadata_text(item["brand"], 80, allow_empty=True)
            canonical = safe_metadata_text(item["canonical_food"], 120, allow_empty=True)
            if name is None or brand is None or canonical is None:
                continue
            items.append({
                "meal": item["meal"] if item["meal"] in MEAL_LABELS else "unassigned",
                "name": name,
                "brand": brand,
                "amount": _public_number(item["amount"]),
                "amount_unit": item["amount_unit"] if item["amount_unit"] in {"g", "ml", "piece", "portion"} else "unknown",
                "macros": {
                    "kcal": _public_number(item["kcal"]),
                    "protein_g": _public_number(item["protein_g"]),
                    "carb_g": _public_number(item["carb_g"]),
                    "fat_g": _public_number(item["fat_g"]),
                },
                "histamine": {
                    "canonical_food": canonical or None,
                    "score": int(item["sighi_score"]) if item["sighi_score"] in (0, 1, 2, 3) else None,
                    "status": "mapped" if item["sighi_score"] in (0, 1, 2, 3) else "unmapped",
                    "confidence": item["confidence"] if item["confidence"] in {"low", "medium", "high"} else "low",
                },
            })
    return {
        "version": 1,
        "date": day,
        "timezone": TZ_NAME,
        "status": "documented",
        "definition": _histamine_definition(),
        "coverage": {
            "status": "complete" if complete else "incomplete",
            "item_count": int(row["item_count"] or 0),
            "unknown_items": int(row["histamine_unknown_count"] or 0),
            "mapping_coverage": round((int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)) / int(row["item_count"] or 1), 3) if int(row["item_count"] or 0) else 0,
            "insufficient_inputs": None if complete else "no_index",
        },
        "macros": {
            "kcal": _public_number(row["kcal"]),
            "protein_g": _public_number(row["protein_g"]),
            "carb_g": _public_number(row["carb_g"]),
            "fat_g": _public_number(row["fat_g"]),
        },
        "nutrients": nutrients,
        "histamine": {
            "score": _public_number(row["histamine_score"]) if complete else None,
            "max_score": int(row["histamine_max"] or 0) if complete else None,
            "unknown_items": int(row["histamine_unknown_count"] or 0),
            "definition_version": _histamine_definition()["version"],
        },
        "meals": meals,
        "items": items,
    }


def _nutrition_mapping_queue(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    if not table_exists(connection, "nutrition_review_queue"):
        return {"version": 1, "timezone": TZ_NAME, "items": []}
    status = params.get("status", "open")
    if status not in {"open"}:
        raise APIError(400, "status_not_allowed")
    items = []
    for row in connection.execute(
        """SELECT normalized_name,example_name,occurrence_count,first_seen,last_seen,
                  suggested_canonical_food,suggested_score,reason,status
           FROM nutrition_review_queue WHERE status=?
           ORDER BY occurrence_count DESC,last_seen DESC LIMIT ?""",
        (status, MAX_NUTRITION_QUEUE + 1),
    ):
        if len(items) >= MAX_NUTRITION_QUEUE:
            raise APIError(422, "row_limit_exceeded")
        example = safe_metadata_text(row["example_name"], 160, allow_empty=False)
        suggested = safe_metadata_text(row["suggested_canonical_food"], 120, allow_empty=True)
        reason = safe_metadata_text(row["reason"], 160, allow_empty=True)
        if example is None or suggested is None or reason is None:
            continue
        items.append({
            "queue_key": nutrition_queue_key(str(row["normalized_name"] or "")),
            "example_name": example,
            "occurrence_count": int(row["occurrence_count"] or 0),
            "first_seen": parse_day(row["first_seen"]),
            "last_seen": parse_day(row["last_seen"]),
            "suggested_canonical_food": suggested,
            "suggested_score": int(row["suggested_score"]) if row["suggested_score"] in (0, 1, 2, 3) else "unknown",
            "reason": reason,
            "status": "open",
            "action_contract": "nutrition_mapping_action_v1",
            "allowed_decisions": ["assign", "composite", "ignore"],
            "allowed_confidence": ["low", "medium", "high"],
            "allowed_mapping_methods": ["sighi_reference", "ingredient_label", "manual_review", "local_alias"],
        })
    return {"version": 1, "timezone": TZ_NAME, "definition": _histamine_definition(), "items": items}


def _day(connection: sqlite3.Connection, day: date) -> dict[str, Any]:
    requested = day.isoformat()
    today = local_today().isoformat()
    measurements = []
    for metric in BY_ID_V2.values():
        if metric.value_type == "panel" or metric.source in {
            "laborwerte",
            "symptom_log",
            "nutrition_daily_summary_v2",
        }:
            continue
        points = _metric_points(connection, metric, day, day)
        if not points:
            continue
        point = points[-1]
        measurements.append(
            {
                "metric": metric.id,
                "label": metric.label,
                "value": point["value"],
                "unit": metric.unit,
                "source": metric.source,
                "aggregation": metric.aggregation,
                "quality": point["quality"],
            }
        )
    symptoms = _day_symptoms(connection, requested)
    medications = _day_medications(connection, requested)
    day_events = _day_events(connection, requested)
    labs = [
        row
        for row in _verified_lab_rows(connection, include_missing_reference=True)
        if row["date"] == requested
    ]
    nutrition = _day_nutrition(connection, requested)
    documents = _day_documents(connection, requested)
    appointments = _day_appointments(connection, requested)
    counts = {
        "measurements": len(measurements),
        "symptoms": symptoms["documented_dimensions"],
        "medications": sum(len(items) for items in medications.values()),
        "labs": len(labs),
        "events": len(day_events["events"]) + len(day_events["periods"]),
        "documents": len(documents),
        "nutrition": 1 if nutrition["status"] == "documented" else 0,
        "appointments": len(appointments["items"]),
    }
    categories = [name for name, count in counts.items() if count]
    metrics = {
        item["metric"]: {
            "value": item["value"],
            "unit": item["unit"],
            "aggregation": item["aggregation"],
            "quality": item["quality"],
        }
        for item in measurements
    }
    notices = []
    if not categories:
        notices.append("Für diesen Tag ist nichts dokumentiert.")
    if symptoms["documented_dimensions"] and not symptoms["complete"]:
        notices.append("Die Symptomdokumentation dieses Tages ist unvollständig.")
    if any(item.get("reference_status") == "missing_reference" for item in labs):
        notices.append(
            "Für mindestens eine Laborbeobachtung fehlt ein verifizierter Referenzbereich."
        )
    return {
        "version": 2,
        "date": requested,
        "today": today,
        "timezone": TZ_NAME,
        "is_today": requested == today,
        "is_future": requested > today,
        "coverage": {
            "status": "documented" if categories else "not_documented",
            "categories": categories,
            "counts": counts,
            "symptoms_complete": symptoms["complete"],
        },
        "measurements": measurements,
        "metrics": metrics,
        "symptoms": symptoms,
        "medications": medications,
        "events": day_events,
        "labs": labs,
        "nutrition": nutrition,
        "documents": documents,
        "appointments": appointments,
        "notices": notices[:3],
    }


def _calendar(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    raw_start = params.get("from")
    raw_end = params.get("to")
    if not raw_start or not raw_end:
        raise APIError(400, "range_required")
    start = parse_iso_day(raw_start)
    end = parse_iso_day(raw_end)
    if end < start:
        raise APIError(400, "range_reversed")
    days = (end - start).days + 1
    if days > MAX_CALENDAR_DAYS:
        raise APIError(422, "calendar_range_too_large")
    today = local_today()
    if end > today + timedelta(days=MAX_FUTURE_DAYS):
        raise APIError(422, "future_range_too_large")
    dates = [(start + timedelta(days=offset)).isoformat() for offset in range(days)]
    counts_by_day = {
        day: {
            "measurements": 0,
            "symptoms": 0,
            "medications": 0,
            "labs": 0,
            "events": 0,
            "documents": 0,
            "nutrition": 0,
            "appointments": 0,
        }
        for day in dates
    }

    def grouped_count(
        table: str, date_column: str, category: str, where: str = "1=1"
    ) -> None:
        if not table_exists(connection, table):
            return
        sql = (
            f"SELECT substr({date_column},1,10) AS day,COUNT(*) AS count FROM {table} "
            f"WHERE {where} AND substr({date_column},1,10) BETWEEN ? AND ? GROUP BY substr({date_column},1,10)"
        )
        for row in connection.execute(sql, (start.isoformat(), end.isoformat())):
            if row["day"] in counts_by_day:
                counts_by_day[row["day"]][category] += min(int(row["count"]), 999)

    # Apple Health uses the same allowlisted metric paths and canonical Zurich-day
    # normalization as /api/v1/day.  A row is visible only if its source identifier
    # is released in APPLE_SOURCE_SPECS; unrecognized Apple rows never influence
    # calendar categories or counts.
    apple_metrics: dict[str, MetricV2] = {}
    for metric in BY_ID_V2.values():
        if (
            metric.source == "apple_health"
            and metric.value_type != "panel"
            and metric.source_identifier in APPLE_SOURCE_SPECS
        ):
            apple_metrics.setdefault(metric.source_identifier, metric)
    for metric in apple_metrics.values():
        for point in _metric_points(connection, metric, start, end):
            observed_day = point["date"]
            if observed_day in counts_by_day:
                counts_by_day[observed_day]["measurements"] = min(
                    counts_by_day[observed_day]["measurements"] + 1, 999
                )

    grouped_count("vitalzeichen", "datum", "measurements")
    grouped_count("symptom_log", "datum", "symptoms", "kontext='daily_quick_score'")
    grouped_count(
        "laborwerte",
        "abnahme_datum",
        "labs",
        "validierungsstatus='validiert' AND verified_against_original=1",
    )
    grouped_count("health_events", "date", "events")
    grouped_count("dokumente", "document_date", "documents", "review_status='geprueft'")
    grouped_count("nutrition_daily_summary_v2", "datum", "nutrition")
    grouped_count("arztbesuche", "datum", "appointments")

    if table_exists(connection, "medication_administrations"):
        seen_medications: dict[str, set[int]] = {day: set() for day in dates}
        for row in connection.execute(
            """SELECT id,datum,scheduled_next_date FROM medication_administrations
               WHERE datum BETWEEN ? AND ? OR scheduled_next_date BETWEEN ? AND ? LIMIT 10000""",
            (start.isoformat(), end.isoformat(), start.isoformat(), end.isoformat()),
        ):
            for candidate in (
                parse_day(row["datum"]),
                parse_day(row["scheduled_next_date"]),
            ):
                if candidate in seen_medications:
                    seen_medications[candidate].add(int(row["id"]))
        for day, identifiers in seen_medications.items():
            counts_by_day[day]["medications"] = min(len(identifiers), 999)

    if table_exists(connection, "health_event_periods"):
        for row in connection.execute(
            """SELECT start_date,end_date FROM health_event_periods
               WHERE start_date<=? AND COALESCE(end_date,start_date)>=? LIMIT 10000""",
            (end.isoformat(), start.isoformat()),
        ):
            period_start = parse_day(row["start_date"])
            period_end = parse_day(row["end_date"]) if row["end_date"] else period_start
            if not period_start or not period_end:
                continue
            current = max(period_start, start.isoformat())
            bounded_end = min(period_end, end.isoformat())
            while current <= bounded_end:
                counts_by_day[current]["events"] = min(
                    counts_by_day[current]["events"] + 1, 999
                )
                current = (date.fromisoformat(current) + timedelta(days=1)).isoformat()

    result = []
    for current in dates:
        counts = counts_by_day[current]
        categories = [name for name, count in counts.items() if count]
        result.append(
            {
                "date": current,
                "is_today": current == today.isoformat(),
                "categories": categories,
                "counts": counts,
                "completeness": "documented" if categories else "not_documented",
                "codes": [f"has_{category}" for category in categories],
            }
        )
    return {
        "from": start.isoformat(),
        "to": end.isoformat(),
        "today": today.isoformat(),
        "timezone": TZ_NAME,
        "days": result,
    }


RECORD_TABS = frozenset(
    {"overview", "labs", "medications", "appointments", "documents", "report"}
)
RECORD_DOCUMENT_SORTS = {
    "document_date_desc": "document_date DESC,id DESC",
    "document_date_asc": "document_date ASC,id ASC",
    "import_date_desc": "upload_datum DESC,id DESC",
    "category": "kategorie COLLATE NOCASE,id DESC",
    "institution": "institution COLLATE NOCASE,id DESC",
    "type": "daten_typ,id DESC",
    "review_status": "review_status,id DESC",
}
RECORD_MAX_ROWS = 100
RECORD_PAGE_SIZE = 25
RECORD_MAX_PAGE_SIZE = 50
RECORD_MAX_MATCHES = 50
RECORD_MAX_SECTIONS = 20
RECORD_MAX_SNIPPET_CHARS = 320
_RECORD_CURSOR_KEY = secrets.token_bytes(32)
_RECORD_CURSOR_VERSION = 1
_RECORD_SORT_SPECS = {
    "document_date_desc": ("document_date", "desc", "desc"),
    "document_date_asc": ("document_date", "asc", "asc"),
    "import_date_desc": ("upload_datum", "desc", "desc"),
    "category": ("kategorie", "asc", "desc"),
    "institution": ("institution", "asc", "desc"),
    "type": ("daten_typ", "asc", "desc"),
    "review_status": ("review_status", "asc", "desc"),
}


def _cursor_binding(params: dict[str, str], *, search: bool) -> str:
    bound: dict[str, Any] = {
        key: params.get(key, "")
        for key in (
            "from",
            "to",
            "category",
            "institution",
            "type",
            "review_status",
            "q",
            "sort",
        )
    }
    bound["search"] = search
    return hashlib.sha256(
        json.dumps(bound, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def _encode_record_cursor(payload: dict[str, Any]) -> str:
    raw = json.dumps(
        {"v": _RECORD_CURSOR_VERSION, **payload},
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    signature = hmac.new(_RECORD_CURSOR_KEY, raw, hashlib.sha256).digest()[:16]
    return "record-cursor-" + base64.urlsafe_b64encode(raw + signature).decode(
        "ascii"
    ).rstrip("=")


def _decode_record_cursor(value: str, binding: str) -> dict[str, Any]:
    if not value.startswith("record-cursor-") or len(value) > 600:
        raise APIError(400, "invalid_cursor")
    encoded = value.removeprefix("record-cursor-")
    try:
        packed = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
        raw, supplied = packed[:-16], packed[-16:]
        expected = hmac.new(_RECORD_CURSOR_KEY, raw, hashlib.sha256).digest()[:16]
        payload = json.loads(raw)
    except (ValueError, UnicodeError, json.JSONDecodeError) as error:
        raise APIError(400, "invalid_cursor") from error
    if not hmac.compare_digest(supplied, expected):
        raise APIError(400, "invalid_cursor")
    if payload.get("v") != _RECORD_CURSOR_VERSION or payload.get("b") != binding:
        raise APIError(400, "cursor_mismatch")
    if not isinstance(payload.get("id"), int) or payload["id"] < 1:
        raise APIError(400, "invalid_cursor")
    return payload


def _record_page_size(params: dict[str, str]) -> int:
    raw = params.get("limit", str(RECORD_PAGE_SIZE))
    if not raw.isdigit():
        raise APIError(400, "invalid_limit")
    size = int(raw)
    if not 1 <= size <= RECORD_MAX_PAGE_SIZE:
        raise APIError(400, "invalid_limit")
    return size


def _record_document_rows(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    """Apply document filters and FTS/review join in SQLite before page limiting."""
    if not table_exists(connection, "dokumente"):
        return {
            "documents": [], "next_cursor": None, "truncated": False, "total": 0,
            "coverage": {"total": 0, "original_available": 0, "text_extracted": 0,
                         "content_reviewed": 0, "fulltext_searchable": 0},
        }
    fts_document_ids: set[int] = set()
    if table_exists(connection, "health_document_fts"):
        fts_document_ids = {
            int(row[0])
            for row in connection.execute(
                "SELECT DISTINCT document_id FROM health_document_fts"
            )
            if str(row[0]).isdigit()
        }
    coverage_rows = connection.execute(
        """SELECT id,review_status,extrahierte_inhalte,local_original_path,dateipfad
           FROM dokumente ORDER BY id LIMIT ?""",
        (RECORD_MAX_ROWS * 20 + 1,),
    ).fetchall()
    if len(coverage_rows) > RECORD_MAX_ROWS * 20:
        raise APIError(422, "source_row_limit_exceeded")
    original_available = 0
    for coverage_row in coverage_rows:
        original = probe_original(
            coverage_row["local_original_path"] or coverage_row["dateipfad"],
            reviewed=coverage_row["review_status"] == "geprueft",
        )
        try:
            original_available += original.status == "available"
        finally:
            original.close()
    document_coverage = {
        "total": len(coverage_rows),
        "original_available": original_available,
        "text_extracted": sum(
            bool(str(row["extrahierte_inhalte"] or "").strip())
            for row in coverage_rows
        ),
        "content_reviewed": sum(
            row["review_status"] == "geprueft" for row in coverage_rows
        ),
        "fulltext_searchable": sum(
            row["review_status"] == "geprueft"
            and int(row["id"]) in fts_document_ids
            for row in coverage_rows
        ),
    }
    start, end = parse_range(params)
    filters = ["1=1"]
    values: list[Any] = []
    if start:
        filters.append("d.document_date>=?")
        values.append(start.isoformat())
    if end:
        filters.append("d.document_date<=?")
        values.append(end.isoformat())
    for key, column, maximum in (
        ("category", "kategorie", 80),
        ("institution", "institution", 120),
        ("type", "daten_typ", 20),
        ("review_status", "review_status", 32),
    ):
        raw = params.get(key)
        if raw:
            clean = safe_metadata_text(raw, maximum, allow_empty=False)
            if clean is None:
                raise APIError(400, "filter_not_allowed")
            filters.append(f"d.{column}=?")
            values.append(clean)
    sort = params.get("sort", "document_date_desc")
    if sort not in RECORD_DOCUMENT_SORTS:
        raise APIError(400, "sort_not_allowed")
    page_size = _record_page_size(params)
    query_text = params.get("q")
    search = bool(query_text)
    binding = _cursor_binding(params, search=search)
    cursor_payload = (
        _decode_record_cursor(params["cursor"], binding)
        if params.get("cursor")
        else None
    )
    join = ""
    select_snippet = "NULL AS snippet"
    select_rank = "NULL AS search_rank"
    order = RECORD_DOCUMENT_SORTS[sort].replace("id", "d.id")
    query_values = list(values)
    search_cursor_clause = ""
    search_cursor_values: list[Any] = []
    if search:
        if not table_exists(connection, "health_document_fts"):
            raise APIError(503, "fts_unavailable")
        tokens = _fts_tokens(query_text or "")
        join = " JOIN health_document_fts f ON CAST(d.id AS TEXT)=f.document_id"
        filters.append("d.review_status='geprueft'")
        filters.append("health_document_fts MATCH ?")
        query_values.append(_fts_literal_query(tokens))
        select_snippet = "snippet(health_document_fts, 6, '', '', '…', ?) AS snippet"
        select_rank = "bm25(health_document_fts) AS search_rank"
        order = "search_rank ASC, d.id DESC"
        if cursor_payload:
            rank = cursor_payload.get("k")
            if not isinstance(rank, (int, float)):
                raise APIError(400, "invalid_cursor")
            search_cursor_clause = "WHERE (search_rank>? OR (search_rank=? AND id<?))"
            search_cursor_values.extend(
                (float(rank), float(rank), cursor_payload["id"])
            )
    else:
        column, primary_direction, id_direction = _RECORD_SORT_SPECS[sort]
        if cursor_payload:
            primary = cursor_payload.get("k")
            if not isinstance(primary, str):
                raise APIError(400, "invalid_cursor")
            primary_operator = ">" if primary_direction == "asc" else "<"
            id_operator = ">" if id_direction == "asc" else "<"
            collation = " COLLATE NOCASE" if sort in {"category", "institution"} else ""
            filters.append(
                f"(COALESCE(d.{column},''){collation}{primary_operator}? OR "
                f"(COALESCE(d.{column},''){collation}=?{collation} AND d.id{id_operator}?))"
            )
            query_values.extend((primary, primary, cursor_payload["id"]))
    where = " AND ".join(filters)
    if search:
        sql = f"""WITH raw_hits AS MATERIALIZED (
                        SELECT d.id,d.document_date,d.upload_datum,d.kategorie,
                               d.institution,d.daten_typ,d.review_status,
                               d.processing_quality,d.groessekbytes,d.extrahierte_inhalte,
                               d.local_original_path,d.dateipfad,f.chunk_no,
                               {select_snippet},{select_rank}
                        FROM dokumente d{join}
                        WHERE {where}
                    ), best_rank AS (
                        SELECT id,MIN(search_rank) AS search_rank
                        FROM raw_hits GROUP BY id
                    ), selected AS (
                        SELECT r.* FROM raw_hits r
                        JOIN best_rank b ON b.id=r.id AND b.search_rank=r.search_rank
                        WHERE r.chunk_no=(
                            SELECT MIN(r2.chunk_no) FROM raw_hits r2
                            WHERE r2.id=r.id AND r2.search_rank=r.search_rank
                        )
                    )
                    SELECT * FROM selected {search_cursor_clause}
                    ORDER BY search_rank ASC,id DESC LIMIT ?"""
        parameters = [
            RECORD_MAX_SNIPPET_CHARS,
            *query_values,
            *search_cursor_values,
            page_size + 1,
        ]
    else:
        sql = f"""SELECT d.id,d.document_date,d.upload_datum,d.kategorie,d.institution,
                         d.daten_typ,d.review_status,d.processing_quality,d.groessekbytes,
                         d.extrahierte_inhalte,d.local_original_path,d.dateipfad,
                         {select_snippet},{select_rank}
                  FROM dokumente d{join}
                  WHERE {where}
                  ORDER BY {order} LIMIT ?"""
        parameters = [*query_values, page_size + 1]
    try:
        rows = connection.execute(sql, parameters).fetchall()
    except sqlite3.OperationalError as error:
        if search:
            raise APIError(400, "query_not_allowed") from error
        raise
    has_more = len(rows) > page_size
    result: list[dict[str, Any]] = []
    for row in rows[:page_size]:
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        if category is None or institution is None:
            continue
        document_day = parse_day(row["document_date"])
        reviewed = row["review_status"] == "geprueft"
        content_status = (
            "extracted"
            if str(row["extrahierte_inhalte"] or "").strip()
            else "not_extracted"
        )
        original = probe_original(
            row["local_original_path"] or row["dateipfad"], reviewed=reviewed
        )
        original_status = (
            "not_checked" if original.status == "not_reviewed" else original.status
        )
        original.close()
        result.append(
            {
                "id": api_document_id(row["id"], document_day, category, institution),
                "title": f"Dokument · {category or 'Ohne Kategorie'}",
                "document_date": document_day,
                "document_date_status": "known" if document_day else "unknown",
                "import_date": parse_day(row["upload_datum"]),
                "category": category,
                "institution": institution,
                "type": row["daten_typ"]
                if row["daten_typ"] in {"pdf", "image"}
                else "unknown",
                "review_status": row["review_status"]
                if row["review_status"] in {"geprueft", "nicht_geprueft"}
                else "unknown",
                "content_status": content_status,
                "original_status": original_status,
                "search_status": (
                    "searchable"
                    if reviewed and int(row["id"]) in fts_document_ids
                    else "not_searchable"
                ),
                "processing_quality": safe_metadata_text(
                    row["processing_quality"], 40, allow_empty=True
                ),
                "size_kb": int(row["groessekbytes"])
                if isinstance(row["groessekbytes"], int)
                and 0 <= row["groessekbytes"] <= 2_000_000
                else None,
                "snippet": str(row["snippet"])[:RECORD_MAX_SNIPPET_CHARS]
                if row["snippet"]
                else None,
            }
        )
    next_cursor = None
    if has_more and result:
        last_row = rows[page_size - 1]
        if search:
            cursor_key: str | float = float(last_row["search_rank"])
        else:
            cursor_column = _RECORD_SORT_SPECS[sort][0]
            cursor_key = str(last_row[cursor_column] or "")
        next_cursor = _encode_record_cursor(
            {"b": binding, "k": cursor_key, "id": int(last_row["id"])}
        )
    return {
        "documents": result,
        "next_cursor": next_cursor,
        "truncated": has_more,
        "total": None,
        "coverage": document_coverage,
    }


def _fts_tokens(raw: str) -> list[str]:
    clean = " ".join(raw.split())
    if not 2 <= len(clean) <= 80:
        raise APIError(400, "query_length_invalid")
    tokens = re.findall(r"[A-Za-zÀ-ÿ0-9]{2,32}", clean)
    if not tokens or len(tokens) > 8 or " ".join(tokens).casefold() != clean.casefold():
        raise APIError(400, "query_not_allowed")
    return tokens


def _fts_literal_query(tokens: list[str]) -> str:
    """Construct only quoted FTS phrases; user input never supplies FTS syntax."""
    return " AND ".join(f'"{token.replace(chr(34), chr(34) * 2)}"' for token in tokens)


def _resolve_record_document(
    connection: sqlite3.Connection, opaque: str, *, reviewed: bool = False
) -> sqlite3.Row:
    if not re.fullmatch(r"api-document-[a-f0-9]{24}", opaque):
        raise APIError(404, "document_not_found")
    sql = "SELECT id,document_date,kategorie,institution,daten_typ,review_status,processing_quality,groessekbytes,datei_hash,upload_datum,extrahierte_inhalte,local_original_path,dateipfad FROM dokumente"
    if reviewed:
        sql += " WHERE review_status='geprueft'"
    for row in connection.execute(sql + " ORDER BY id LIMIT 10000"):
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        if (
            category is not None
            and institution is not None
            and api_document_id(
                row["id"], parse_day(row["document_date"]), category, institution
            )
            == opaque
        ):
            return row
    raise APIError(404, "document_not_found")


def _record_document_detail(
    connection: sqlite3.Connection, opaque: str, params: dict[str, str]
) -> dict[str, Any]:
    row = _resolve_record_document(connection, opaque, reviewed=True)
    category = safe_metadata_text(row["kategorie"], 80, allow_empty=True) or ""
    institution = safe_metadata_text(row["institution"], 120, allow_empty=True) or ""
    document_day = parse_day(row["document_date"])
    all_sections = document_chunks(row["extrahierte_inhalte"])
    content_status = "available" if all_sections else "no_extracted_content"
    original = probe_original(
        row["local_original_path"] or row["dateipfad"], reviewed=True
    )
    original_status = original.status
    original.close()
    try:
        after = decode_chunk_cursor(params["cursor"]) if params.get("cursor") else 0
    except ValueError as error:
        raise APIError(400, "invalid_cursor") from error
    page_size = _record_page_size(params)
    selected = [section for section in all_sections if section.number > after]
    has_more = len(selected) > page_size
    sections = [
        {"number": section.number, "text": section.text}
        for section in selected[:page_size]
    ]
    return {
        "id": opaque,
        "title": f"Dokument · {category or 'Ohne Kategorie'}",
        "document_date": document_day,
        "document_date_status": "known" if document_day else "unknown",
        "category": category,
        "institution": institution,
        "type": row["daten_typ"] if row["daten_typ"] in {"pdf", "image"} else "unknown",
        "review_status": "geprueft",
        "content_status": content_status,
        "original_status": original_status,
        "sections": sections,
        "section_total": len(all_sections),
        "next_cursor": encode_chunk_cursor(sections[-1]["number"])
        if has_more and sections
        else None,
        "truncated": has_more,
        "original_available": original_status == "available",
    }


def _record_document_matches(
    connection: sqlite3.Connection, opaque: str, params: dict[str, str]
) -> dict[str, Any]:
    row = _resolve_record_document(connection, opaque, reviewed=True)
    tokens = _fts_tokens(params.get("q", ""))
    if not table_exists(connection, "health_document_fts"):
        raise APIError(503, "fts_unavailable")
    fts_query = _fts_literal_query(tokens)
    matches: list[dict[str, Any]] = []
    rows = connection.execute(
        """SELECT f.chunk_no,f.content
           FROM health_document_fts f
           JOIN dokumente d ON CAST(d.id AS TEXT)=f.document_id
           WHERE d.id=? AND d.review_status='geprueft'
             AND health_document_fts MATCH ?
           ORDER BY f.chunk_no LIMIT ?""",
        (int(row["id"]), fts_query, RECORD_MAX_MATCHES + 1),
    )
    for match in list(rows)[:RECORD_MAX_MATCHES]:
        text = str(match["content"])
        position = min(
            (
                text.casefold().find(token.casefold())
                for token in tokens
                if text.casefold().find(token.casefold()) >= 0
            ),
            default=-1,
        )
        if position >= 0:
            matches.append(
                {
                    "section": int(match["chunk_no"]),
                    "start": position,
                    "end": position + len(tokens[0]),
                    "snippet": text[max(0, position - 100) : position + 220],
                }
            )
    return {
        "id": opaque,
        "query": " ".join(tokens),
        "matches": matches,
        "next_cursor": None,
        "truncated": len(matches) == RECORD_MAX_MATCHES,
        "total": None,
    }


def _record_medications(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    start, end = parse_range(params)
    result: dict[str, Any] = {
        "planned": [],
        "administered": [],
        "missed": [],
        "corrected": [],
    }
    if not table_exists(connection, "medication_administrations"):
        return result
    groups = {
        "planned": ("planned", "scheduled", "geplant"),
        "administered": tuple(sorted(ADMINISTERED)),
        "missed": ("missed", "verpasst", "ausgelassen"),
        "corrected": ("corrected", "korrigiert", "correction"),
    }
    truncated_sections: list[str] = []
    for bucket, events in groups.items():
        effective = (
            "COALESCE(NULLIF(scheduled_next_date,''),datum)"
            if bucket == "planned"
            else "datum"
        )
        placeholders = ",".join("?" for _ in events)
        filters = [f"lower(trim(COALESCE(event_type,''))) IN ({placeholders})"]
        values: list[Any] = list(events)
        if start:
            filters.append(f"{effective}>=?")
            values.append(start.isoformat())
        if end:
            filters.append(f"{effective}<=?")
            values.append(end.isoformat())
        direction = "ASC" if bucket == "planned" else "DESC"
        rows = connection.execute(
            "SELECT datum,medication_name,dose,route,event_type,"
            "scheduled_next_date,notes,source FROM medication_administrations WHERE "
            + " AND ".join(filters)
            + f" ORDER BY {effective} {direction},id {direction} LIMIT ?",
            (*values, RECORD_MAX_ROWS + 1),
        ).fetchall()
        if len(rows) > RECORD_MAX_ROWS:
            truncated_sections.append(bucket)
        for row in rows[:RECORD_MAX_ROWS]:
            recorded_day = parse_day(row["datum"])
            scheduled = parse_day(row["scheduled_next_date"])
            effective_day = (
                scheduled or recorded_day if bucket == "planned" else recorded_day
            )
            name = safe_metadata_text(row["medication_name"], 120)
            if not recorded_day or not effective_day or not name:
                continue
            result[bucket].append(
                {
                    "date": effective_day,
                    "recorded_date": recorded_day,
                    "scheduled_next_date": scheduled if bucket == "planned" else None,
                    "name": name,
                    "dose": safe_metadata_text(row["dose"], 40, allow_empty=True),
                    "route": safe_metadata_text(row["route"], 40, allow_empty=True),
                    "note": safe_metadata_text(row["notes"], 300, allow_empty=True),
                    "source": safe_metadata_text(row["source"], 80, allow_empty=True),
                }
            )
    result["truncated"] = bool(truncated_sections)
    result["truncated_sections"] = truncated_sections
    result["next_cursor"] = None
    return result


def _record_appointments(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    start, end = parse_range(params)
    order = params.get("order", "desc")
    if order not in {"asc", "desc"}:
        raise APIError(400, "sort_not_allowed")
    institution_filter = (
        safe_metadata_text(params["institution"], 120)
        if params.get("institution")
        else None
    )
    filters = ["1=1"]
    values: list[Any] = []
    if start:
        filters.append("datum>=?")
        values.append(start.isoformat())
    if end:
        filters.append("datum<=?")
        values.append(end.isoformat())
    if institution_filter:
        filters.append("klinik=?")
        values.append(institution_filter)
    rows = (
        connection.execute(
            "SELECT datum,arzt,klinik,grund,notizen FROM arztbesuche WHERE "
            + " AND ".join(filters)
            + f" ORDER BY datum {order.upper()},id {order.upper()} LIMIT ?",
            (*values, RECORD_MAX_ROWS + 1),
        ).fetchall()
        if table_exists(connection, "arztbesuche")
        else []
    )
    items = []
    for row in rows:
        day = parse_day(row["datum"])
        institution = safe_metadata_text(row["klinik"], 120, allow_empty=True)
        if (
            not day
            or (start and day < start.isoformat())
            or (end and day > end.isoformat())
            or (institution_filter and institution != institution_filter)
        ):
            continue
        practitioner = safe_metadata_text(row["arzt"], 120, allow_empty=True)
        reason = safe_metadata_text(row["grund"], 160, allow_empty=True)
        note = safe_metadata_text(row["notizen"], 300, allow_empty=True)
        if None not in (institution, practitioner, reason, note):
            items.append(
                {
                    "date": day,
                    "practitioner": practitioner,
                    "institution": institution,
                    "reason": reason,
                    "note": note,
                    "status": "documented_visit",
                }
            )
    return {
        "items": items[:RECORD_MAX_ROWS],
        "truncated": len(rows) > RECORD_MAX_ROWS,
        "next_cursor": None,
    }


def _lab_search_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "", value.casefold().replace("ß", "ss"))


LAB_SEARCH_ALIASES = {
    _lab_search_key("c-reaktives protein crp"): "CRP",
    _lab_search_key("crp"): "CRP",
    _lab_search_key("d-dimer"): "D-Dimer",
}


def _record_labs(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    query = params.get("q", "")
    if query and (len(query) > 80 or safe_metadata_text(query, 80) is None):
        raise APIError(400, "query_not_allowed")
    search_key = _lab_search_key(query) if query else ""
    canonical_query = LAB_SEARCH_ALIASES.get(search_key, query.casefold())
    start, end = parse_range(params)
    today = local_today()
    history_floor = today - timedelta(days=MAX_RANGE_DAYS - 1)
    all_rows = _verified_lab_rows(
        connection,
        include_missing_reference=True,
        start=history_floor,
        end=today,
    )
    grouped_rows: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in all_rows:
        grouped_rows[str(row["metric_id"])].append(row)
    histories: dict[str, dict[str, Any]] = {}
    previous_by_id: dict[str, dict[str, Any]] = {}
    for metric_id, observations in grouped_rows.items():
        observations.sort(key=lambda row: (str(row["date"]), str(row["id"])))
        histories[metric_id] = {
            "metric_id": metric_id,
            "earliest_date": observations[0]["date"],
            "latest_date": observations[-1]["date"],
            "observation_count": len(observations),
            "verified_observation_count": len(observations),
            "missing_reference_count": sum(
                item.get("reference_status") == "missing_reference"
                for item in observations
            ),
            "linked_document_count": sum(
                item.get("document_link_status") == "linked" for item in observations
            ),
            "unlinked_document_count": sum(
                item.get("document_link_status") != "linked" for item in observations
            ),
            "range_from": observations[0]["date"],
            "range_to": today.isoformat(),
        }
        for index, observation in enumerate(observations):
            if index:
                previous_by_id[str(observation["id"])] = observations[index - 1]
    result = []
    for source in all_rows:
        if canonical_query and canonical_query.casefold() not in str(source["parameter"]).casefold():
            continue
        if start and source["date"] < start.isoformat():
            continue
        if end and source["date"] > end.isoformat():
            continue
        previous = previous_by_id.get(str(source["id"]))
        value = source.get("value")
        prior = previous.get("value") if previous else None
        delta = (
            round(float(value) - float(prior), 6)
            if isinstance(value, (int, float)) and isinstance(prior, (int, float))
            else None
        )
        item = {
            **source,
            "history": histories[source["metric_id"]],
            "previous": prior,
            "absolute_change": delta,
            "percent_change": round(delta / float(prior) * 100, 4)
            if delta is not None and float(prior) != 0
            else None,
            "document_id": source.get("document", {}).get("id")
            if source.get("document")
            else None,
        }
        result.append(item)
    result.sort(key=lambda item: (item["date"], item["id"]), reverse=True)
    visible_histories = [
        history
        for history in histories.values()
        if not canonical_query
        or canonical_query.casefold()
        in str(next(item["parameter"] for item in grouped_rows[history["metric_id"]])).casefold()
    ]
    visible_histories.sort(key=lambda item: item["metric_id"])
    if len(result) > MAX_SERIES_ROWS:
        raise APIError(422, "row_limit_exceeded")
    return {
        "reference_policy": "observation_specific_verified_original",
        "today": today.isoformat(),
        "histories": visible_histories,
        "observations": result,
        "truncated": False,
        "completeness": "complete_page",
    }


def _next_planned_medications(
    connection: sqlite3.Connection, today: date, *, limit: int = 3
) -> list[dict[str, Any]]:
    if not table_exists(connection, "medication_administrations"):
        return []
    rows = connection.execute(
        """SELECT datum,medication_name,dose,route,scheduled_next_date,notes,source
           FROM medication_administrations
           WHERE lower(trim(COALESCE(event_type,''))) IN ('planned','scheduled','geplant')
             AND scheduled_next_date>?
           ORDER BY scheduled_next_date ASC,id ASC LIMIT ?""",
        (today.isoformat(), limit),
    )
    result = []
    for row in rows:
        recorded = parse_day(row["datum"])
        scheduled = parse_day(row["scheduled_next_date"])
        name = safe_metadata_text(row["medication_name"], 120)
        if recorded and scheduled and name:
            result.append(
                {
                    "date": scheduled,
                    "recorded_date": recorded,
                    "scheduled_next_date": scheduled,
                    "name": name,
                    "dose": safe_metadata_text(row["dose"], 40, allow_empty=True),
                    "route": safe_metadata_text(row["route"], 40, allow_empty=True),
                    "note": safe_metadata_text(row["notes"], 300, allow_empty=True),
                    "source": safe_metadata_text(row["source"], 80, allow_empty=True),
                }
            )
    return result


def _record_summary(connection: sqlite3.Connection) -> dict[str, Any]:
    document_page = _record_document_rows(
        connection, {"sort": "document_date_desc", "limit": "3"}
    )
    today = local_today()
    medications = _record_medications(connection, {})
    next_planned = _next_planned_medications(connection, today)
    appointments = _record_appointments(connection, {"order": "desc"})
    labs = _record_labs(connection, {})
    not_reviewed = 0
    if table_exists(connection, "dokumente"):
        not_reviewed = int(
            connection.execute(
                "SELECT COUNT(*) FROM dokumente WHERE review_status<>'geprueft'"
            ).fetchone()[0]
        )
    latest_labs: list[dict[str, Any]] = []
    seen_lab_parameters: set[tuple[str, str]] = set()
    for observation in labs["observations"]:
        key = (str(observation.get("parameter")), str(observation.get("unit")))
        if key not in seen_lab_parameters:
            seen_lab_parameters.add(key)
            latest_labs.append(observation)
    return {
        "labs": latest_labs[:3],
        "labs_truncated": labs["truncated"],
        "latest_administered": medications["administered"][:3],
        "next_planned": next_planned,
        "appointments": appointments["items"][:3],
        "documents": document_page["documents"],
        "documents_not_reviewed": not_reviewed,
    }


REPORT_SECTIONS = frozenset(
    {"overview", "labs", "medications", "symptoms", "appointments", "documents"}
)


def _doctor_report(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    start, end = parse_range(params, required=True)
    assert start and end
    selected = set(filter(None, params.get("sections", "").split(","))) or set(
        REPORT_SECTIONS
    )
    if not selected <= REPORT_SECTIONS:
        raise APIError(400, "section_not_allowed")
    lab_page = (
        _record_labs(
            connection,
            {"from": start.isoformat(), "to": end.isoformat()},
        )
        if "labs" in selected
        else {"observations": [], "truncated": False}
    )
    labs = lab_page["observations"]
    medications = (
        _record_medications(
            connection, {"from": start.isoformat(), "to": end.isoformat()}
        )
        if "medications" in selected
        else {"planned": [], "administered": [], "missed": [], "corrected": []}
    )
    appointments = (
        _record_appointments(
            connection,
            {"from": start.isoformat(), "to": end.isoformat(), "order": "asc"},
        )["items"]
        if "appointments" in selected
        else []
    )
    document_page = (
        _record_document_rows(
            connection,
            {
                "from": start.isoformat(),
                "to": end.isoformat(),
                "sort": "document_date_asc",
                "limit": str(RECORD_MAX_PAGE_SIZE),
            },
        )
        if "documents" in selected
        else {"documents": [], "truncated": False}
    )
    documents = document_page["documents"]
    symptoms: list[dict[str, Any]] = []
    if "symptoms" in selected:
        current = start
        while current <= end:
            day = current.isoformat()
            symptom = _day_symptoms(connection, day)
            events = _day_events(connection, day)
            if (
                symptom["documented_dimensions"]
                or events["events"]
                or events["periods"]
            ):
                symptoms.append({"date": day, "symptoms": symptom, "events": events})
            current += timedelta(days=1)
    overview = _record_summary(connection) if "overview" in selected else None
    payload = {
        "from": start.isoformat(),
        "to": end.isoformat(),
        "timezone": TZ_NAME,
        "selected_sections": sorted(selected),
        "overview": overview,
        "labs": labs,
        "medications": medications,
        "symptoms": symptoms,
        "appointments": appointments,
        "documents": documents,
        "provenance": "canonical_read_only_health_database",
        "completeness": {
            name: {
                "status": "documented" if section_payload else "unknown",
                "truncated": {
                    "overview": False,
                    "labs": bool(lab_page.get("truncated")),
                    "medications": bool(medications.get("truncated")),
                    "symptoms": False,
                    "appointments": False,
                    "documents": bool(document_page.get("truncated")),
                }[name],
            }
            for name, section_payload in {
                "overview": overview,
                "labs": labs,
                "medications": sum(
                    len(medications.get(kind, []))
                    for kind in ("planned", "administered", "missed", "corrected")
                ),
                "symptoms": symptoms,
                "appointments": appointments,
                "documents": documents,
            }.items()
            if name in selected
        },
        "medical_statement": "Dokumentierte Daten ohne Diagnose-, Kausalitäts- oder Therapieaussage.",
    }
    return payload


def dispatch_api(database: Path, path: str, query: str) -> dict[str, Any]:
    document_route = re.fullmatch(
        r"/api/v1/documents/(api-document-[a-f0-9]{24})(?:/(matches))?", path
    )
    allowed_record = {
        "/api/v1/record-summary",
        "/api/v1/record-labs",
        "/api/v1/medications",
        "/api/v1/appointments",
        "/api/v1/documents",
        "/api/v1/doctor-report",
    }
    if (
        path not in API_PATHS
        and path not in allowed_record
        and not document_route
        and not re.fullmatch(r"/api/v1/day/\d{4}-\d{2}-\d{2}", path)
        and not re.fullmatch(r"/api/v1/nutrition/day/\d{4}-\d{2}-\d{2}", path)
    ):
        raise APIError(404, "endpoint_not_found")
    connection = connect_read_only(database)
    try:
        if path == "/api/v1/record-summary":
            parse_query(query, set())
            return _record_summary(connection)
        if path == "/api/v1/record-labs":
            return _record_labs(connection, parse_query(query, {"q", "from", "to"}))
        if path == "/api/v1/medications":
            return _record_medications(connection, parse_query(query, {"from", "to"}))
        if path == "/api/v1/appointments":
            return _record_appointments(
                connection, parse_query(query, {"from", "to", "order", "institution"})
            )
        if path == "/api/v1/documents":
            return _record_document_rows(
                connection,
                parse_query(
                    query,
                    {
                        "from",
                        "to",
                        "category",
                        "institution",
                        "type",
                        "review_status",
                        "sort",
                        "q",
                        "cursor",
                        "limit",
                    },
                ),
            )
        if document_route:
            opaque, suffix = document_route.groups()
            if suffix == "matches":
                return _record_document_matches(
                    connection, opaque, parse_query(query, {"q"})
                )
            return _record_document_detail(
                connection,
                opaque,
                parse_query(query, {"cursor", "limit"}),
            )
        if path == "/api/v1/doctor-report":
            return _doctor_report(
                connection, parse_query(query, {"from", "to", "sections"})
            )
        if path == "/api/v1/metric-catalog":
            params = parse_query(query, {"q"})
            query_text = params.get("q", "")
            metrics = (
                catalog_search(connection, query_text)
                if query_text
                else public_catalog(connection)
            )
            inventory = inventory_sources(connection)
            return {
                "version": 2,
                "timezone": TZ_NAME,
                "today": local_today().isoformat(),
                "groups": {"metrics": metrics},
                "inventory": {
                    "apple_observed_identifiers": len(inventory["apple_health"]),
                    "apple_released_identifiers": sum(
                        item["status"] == "released"
                        for item in inventory["apple_health"]
                    ),
                    "canonical_verified_laboratory_parameters": len(
                        inventory["laboratory"]
                    ),
                    "observed_vital_sign_parameters": len(inventory["vital_signs"]),
                },
            }
        if path == "/api/v1/series":
            params = parse_query(query, {"metric", "from", "to", "resolution"})
            if not params.get("metric"):
                raise APIError(400, "metric_required")
            return _series(connection, params)
        if path.startswith("/api/v1/day/"):
            if query:
                raise APIError(400, "unknown_parameter")
            requested_day = parse_iso_day(path.rsplit("/", 1)[-1])
            if requested_day > local_today() + timedelta(days=MAX_FUTURE_DAYS):
                raise APIError(422, "future_date_not_allowed")
            return _day(connection, requested_day)
        if path == "/api/v1/nutrition/days":
            return _nutrition_days(connection, parse_query(query, {"from", "to"}))
        if path.startswith("/api/v1/nutrition/day/"):
            if query:
                raise APIError(400, "unknown_parameter")
            requested_day = parse_iso_day(path.rsplit("/", 1)[-1])
            if requested_day > local_today():
                raise APIError(422, "future_date_not_allowed")
            return _nutrition_day_detail(connection, requested_day)
        if path == "/api/v1/nutrition/mapping-queue":
            return _nutrition_mapping_queue(connection, parse_query(query, {"status"}))
        if path == "/api/v1/calendar":
            params = parse_query(query, {"from", "to"})
            return _calendar(connection, params)
        if path == "/api/v1/events":
            params = parse_query(query, {"from", "to", "types"})
            return _events(connection, params)
        if path == "/api/v1/labs":
            parse_query(query, set())
            rows = _verified_lab_rows(connection)
            if len(rows) > MAX_LAB_ROWS:
                raise APIError(422, "row_limit_exceeded")
            return {
                "timezone": TZ_NAME,
                "reference_policy": "observation_specific_verified_original",
                "observations": rows,
            }
        if path == "/api/v1/search":
            params = parse_query(query, {"q"})
            return _search(connection, params.get("q", ""))
        raise APIError(404, "endpoint_not_found")
    except SourceInventoryLimitError as error:
        raise APIError(422, "source_inventory_limit_exceeded") from error
    except sqlite3.Error as error:
        raise APIError(503, "data_unavailable") from error
    finally:
        connection.close()