"""Each metric's normal, by time slot (#5739, milestone 513).

A fixed cutoff suits some resources and not others: a router that serves
error pages answers 4xx all day, and a host that backs up at 02:00 runs hot
for that hour every night. So Steward learns, for each metric that opts in,
what it usually does in each hour, and a reading is judged against the
normal for its own hour (System 182 "Metric Norms and Anomalies").

What a slot's normal is:
- usual_hold: the level the slot's hours usually held. Each hour's p95
  (plugin_metrics_hourly.value_p95, #5738), at its 90th percentile across
  past days. A two-minute spike doesn't raise it; an hour spent high does.
- usual_peak: the same for each hour's max, so a page can say how high the
  slot usually goes. Most hours of a metric that bursts now and then see no
  burst, so this can read 0 for a slot such a metric bursts in as much as any.
- usual_anywhere: how high the metric has gone in most hours of the day, the
  median across the 24 hours of each one's highest peak. Bursts that land at
  any hour raise it. A job at a time of its own raises a few hours and leaves
  the median alone, so the same level at another time still stands out.

The slots:
- An hour of the day (24 slots) once a key has DAY_MIN_DAYS of history.
- An hour of the week (168 slots) once its history reaches back WEEK_MIN_DAYS,
  so a weekly job is learned too.
- Slots are UTC hours. Each takes in its neighbours (±NEIGHBOURS), so a job
  that starts a few minutes late, or an hour off across a daylight-saving
  change, stays inside its normal.

Only metrics registered with learn_normal() are learned: a host agent alone
reports thousands of per-core, per-disk and per-interface series, and most
have no use for a normal.

An hour an anomaly covers is learned only once someone marks the anomaly
normal (steward.core.anomalies): a level that jumps and stays is never
absorbed on its own, and a lapsed anomaly is never learned.
"""
from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import ARRAY, Boolean, DateTime, String, bindparam, delete, select, text, tuple_

from steward.models.norms import MetricNorm

DAY_MIN_DAYS = 5        # days of history before an hour-of-day normal is trusted
WEEK_MIN_DAYS = 21      # history reaching back this far learns hour-of-week instead
DAY_WINDOW = timedelta(days=28)    # what an hour-of-day normal learns from
WEEK_WINDOW = timedelta(days=56)   # what an hour-of-week normal learns from: 8 of each slot
NEIGHBOURS = 1          # hours either side a slot takes in
USUAL = 0.9             # "usual" is this percentile across past days: one odd day in ten is ignored


@dataclass(frozen=True)
class Learned:
    """A metric that learns a normal. `floor` is the smallest departure,
    in the metric's own unit, worth calling an anomaly (#5740), so that
    0.1% → 0.3% of a share or 2 ms → 4 ms of latency is never news. It may
    be a function, read at each pass, for a floor that is a setting (#5742).

    `gate` is (metric, least): the resource is judged only while that metric
    of its own averages at least `least` over the window, as a router's
    error share means nothing over a handful of requests (#5741). `least`
    may be a function, read at each pass, for a minimum that is a setting.
    `fewest`, for a share (%) gated on a per-second rate: the fewest events
    the share stands for over the window (rate × window × share / 100), as
    three 404s are 7% of a quiet router and no news (#6105); a number or a
    function, like `least`. `since`:
    history before it is not learned from, for a metric whose meaning
    changed then. `parts=False` learns only whole resources, not their parts
    ("<host>:core3"): a host's CPU has a normal, each of its cores doesn't."""
    source_module: str
    metric_name: str
    floor: float | Callable[[], float]
    gate: tuple[str, float | Callable[[], float]] | None = None
    since: datetime | None = None
    parts: bool = True
    fewest: float | Callable[[], float] | None = None

    @property
    def least_departure(self) -> float:
        return float(self.floor() if callable(self.floor) else self.floor)

    @property
    def floor_name(self) -> str | None:
        """The NORM_FLOORS setting the floor is read from, if it is one."""
        return getattr(self.floor, "floor_name", None)


LEARNED: dict[tuple[str, str], Learned] = {}


def learn_normal(source_module: str, metric_name: str, floor: float | Callable[[], float], *,
                 gate: tuple[str, float | Callable[[], float]] | None = None,
                 since: datetime | None = None, parts: bool = True,
                 fewest: float | Callable[[], float] | None = None) -> None:
    """Have Steward learn this metric's normal for every resource reporting
    it. A plugin calls this from setup(), as it calls prune_by_age."""
    LEARNED[(source_module, metric_name)] = Learned(
        source_module, metric_name, floor if callable(floor) else float(floor), gate, since, parts,
        fewest)


def floor_setting(app, name: str) -> Callable[[], float]:
    """A floor read from Settings → Thresholds (norms.floor.<name>) at each
    pass, so a saved change applies at once. The function carries the
    floor's name (`floor_name`), so a reader can say which setting it is."""
    from steward.core.settings import floor_of

    def read() -> float:
        return floor_of(app.config.get("THRESHOLDS", {}), name)
    read.floor_name = name
    return read


@dataclass(frozen=True)
class Norm:
    kind: str           # "day" or "week"
    slot: int
    usual_hold: float
    usual_peak: float
    days_seen: int
    usual_anywhere: float = 0.0

    @property
    def usual_height(self) -> float:
        """The highest this metric usually goes at this time: a level past it
        is news however briefly it lasted."""
        return max(self.usual_peak, self.usual_anywhere)


def slot_of(at: datetime, kind: str) -> int:
    """The slot `at` falls in: its UTC hour of the day (0–23), or its hour of
    the week counted from Monday 00:00 UTC (0–167)."""
    at = at.astimezone(timezone.utc)
    if kind == "week":
        return at.weekday() * 24 + at.hour
    return at.hour


# One statement learns every normal: the hours of each learned key (rolled up,
# plus the raw window not rolled up yet) less those an anomaly covers that
# nobody marked normal (#5740), which kind of slot its history supports, then
# each slot's usual hold and peak across past days, every hour counted in its
# own slot and its neighbours'.
_REBUILD = text("""
WITH learned AS (
    SELECT key, since, parts
    FROM unnest(CAST(:keys AS text[]), CAST(:starts AS timestamptz[]), CAST(:parts AS boolean[]))
         AS k(key, since, parts)
),
all_hours AS (
    SELECT h.source_module AS s, h.resource_name AS r, h.metric_name AS m, h.bucket AS b,
           COALESCE(h.value_p95, h.value_avg) AS hold, h.value_max AS peak
    FROM plugin_metrics_hourly h
    JOIN learned k ON k.key = h.source_module || '/' || h.metric_name
    WHERE h.bucket >= k.since AND (k.parts OR strpos(h.resource_name, ':') = 0)
    UNION ALL
    SELECT p.source_module, p.resource_name, p.metric_name, date_trunc('hour', p.recorded_at),
           percentile_cont(0.95) WITHIN GROUP (ORDER BY p.value), max(p.value)
    FROM plugin_metrics p
    JOIN learned k ON k.key = p.source_module || '/' || p.metric_name
    WHERE p.recorded_at >= k.since AND p.recorded_at < :this_hour
      AND (k.parts OR strpos(p.resource_name, ':') = 0)
    GROUP BY 1, 2, 3, 4
),
hours AS (
    SELECT * FROM all_hours h
    WHERE NOT EXISTS (
        SELECT 1 FROM metric_anomaly_spans sp
        JOIN metric_anomalies a ON a.id = sp.anomaly_id
        WHERE a.review <> 'normal'
          AND a.source_module = h.s AND a.resource_name = h.r AND a.metric_name = h.m
          AND sp.started_at < h.b + interval '1 hour'
          AND COALESCE(sp.ended_at, :now) > h.b)
),
keyed AS (
    SELECT s, r, m, count(DISTINCT date_trunc('day', b)) AS days, min(b) AS first
    FROM hours GROUP BY s, r, m
),
kinds AS (
    SELECT s, r, m, days, CASE WHEN first <= :week_ready THEN 'week' ELSE 'day' END AS kind
    FROM keyed WHERE days >= CAST(:day_min_days AS int)
),
anywhere AS (
    SELECT s, r, m, percentile_cont(0.5) WITHIN GROUP (ORDER BY highest) AS peak
    FROM (SELECT s, r, m, max(peak) AS highest FROM hours
          GROUP BY s, r, m, EXTRACT(HOUR FROM b AT TIME ZONE 'UTC')) by_hour
    GROUP BY s, r, m
),
slotted AS (
    SELECT h.s, h.r, h.m, k.kind, k.days, h.hold, h.peak,
           CASE k.kind
               WHEN 'week' THEN (EXTRACT(ISODOW FROM h.b AT TIME ZONE 'UTC')::int - 1) * 24
                                + EXTRACT(HOUR FROM h.b AT TIME ZONE 'UTC')::int
               ELSE EXTRACT(HOUR FROM h.b AT TIME ZONE 'UTC')::int
           END AS slot,
           CASE k.kind WHEN 'week' THEN 168 ELSE 24 END AS slots
    FROM hours h JOIN kinds k ON k.s = h.s AND k.r = h.r AND k.m = h.m
    WHERE k.kind = 'week' OR h.b >= :day_since
)
INSERT INTO metric_norms (source_module, resource_name, metric_name, slot, kind,
                          usual_hold, usual_peak, usual_anywhere, days_seen, updated_at)
SELECT l.s, l.r, l.m, (l.slot + n.off + l.slots) % l.slots, l.kind,
       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY l.hold),
       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY l.peak),
       a.peak, l.days, :now
FROM slotted l
JOIN anywhere a ON a.s = l.s AND a.r = l.r AND a.m = l.m
CROSS JOIN generate_series(-CAST(:neighbours AS int), CAST(:neighbours AS int)) AS n(off)
GROUP BY l.s, l.r, l.m, (l.slot + n.off + l.slots) % l.slots, l.kind, l.days, a.peak
""").bindparams(bindparam("keys", type_=ARRAY(String)),
               bindparam("starts", type_=ARRAY(DateTime(timezone=True))),
               bindparam("parts", type_=ARRAY(Boolean)))


async def rebuild_norms(session, now: datetime | None = None) -> int:
    """Relearn every normal from history, replacing the last pass's. Runs
    inside the caller's transaction. Returns how many slot rows it wrote."""
    now = now or datetime.now(timezone.utc)
    await session.execute(delete(MetricNorm))
    if not LEARNED:
        return 0
    week_since = now - WEEK_WINDOW
    result = await session.execute(_REBUILD, {
        "keys": [f"{s}/{m}" for s, m in LEARNED],
        # Each key learns from WEEK_WINDOW back, or from its own `since`.
        "starts": [max(week_since, k.since) if k.since else week_since for k in LEARNED.values()],
        "parts": [k.parts for k in LEARNED.values()],
        "day_since": now - DAY_WINDOW,
        "this_hour": now.replace(minute=0, second=0, microsecond=0),
        "week_ready": now - timedelta(days=WEEK_MIN_DAYS),
        "day_min_days": DAY_MIN_DAYS,
        "usual": USUAL,
        "neighbours": NEIGHBOURS,
        "now": now,
    })
    return result.rowcount or 0


async def norms_at(session, keys: list[tuple[str, str, str]],
                   at: datetime) -> dict[tuple[str, str, str], Norm]:
    """{(source, resource, metric): its normal for the slot `at` falls in}.
    A key with no normal yet (too little history, or not learned) is absent,
    and its caller falls back to the fixed cutoffs."""
    if not keys:
        return {}
    day, week = slot_of(at, "day"), slot_of(at, "week")
    rows = (await session.execute(
        select(MetricNorm)
        .where(tuple_(MetricNorm.source_module, MetricNorm.resource_name,
                      MetricNorm.metric_name).in_(keys))
        .where(((MetricNorm.kind == "day") & (MetricNorm.slot == day))
               | ((MetricNorm.kind == "week") & (MetricNorm.slot == week)))
    )).scalars().all()
    return {(n.source_module, n.resource_name, n.metric_name):
            Norm(n.kind, n.slot, n.usual_hold, n.usual_peak, n.days_seen, n.usual_anywhere) for n in rows}


async def run_norms(app) -> None:
    """The hourly pass: relearn every normal in one transaction."""
    async with app.db_sessionmaker() as session:
        async with session.begin():
            await rebuild_norms(session)