thermograph/backend/notifications/notify.py

"""Subscription evaluation engine — the background worker that turns unusual
weather into notifications.

A daemon thread (modeled on app.py's neighbor warmer) wakes on an interval and, for
every active subscription, checks whether the subscribed cell's recent/forecast
weather crossed the user's percentile threshold on any watched metric. Crossings
become rows in the notifications table.

Two guards keep it quiet and cheap:

* **Per-event dedup** — the notifications table has a UNIQUE(subscription_id,
  event_date, metric, direction, kind); an INSERT OR IGNORE means re-running the
  pass never re-notifies the same event.
* **Weekly cap** — a subscription that has notified within the last 7 days is
  skipped entirely (Subscription.last_notified_at), so one alert = at most one
  notification per week, as specified.

Quota safety: a subscribed cell's ~45-year archive is read from the parquet cache
(climate.load_cached_history) and never re-fetched once present. A cell that has no
cached archive yet is fetched ONCE (budget-capped per pass) so a freshly-subscribed
city starts working; the recent/forecast bundle refreshes on its own hourly cadence.
A forecast-fetch failure just skips the cell, so a pass never dies on a rate limit.
"""
import concurrent.futures
import datetime
import logging
import os
import threading
import time

import polars as pl
from sqlalchemy import delete, select

from api import homepage
from api.payloads import OBS_COLS
from data import climate
from notifications import discord
from data import grading
from data import grid
from core import audit
from core import metrics
from notifications import push
from accounts.db import sync_session_maker
from accounts.models import AccessToken, Notification, PushSubscription, Subscription, User

log = logging.getLogger("thermograph.notify")

_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""

WEEK_SECONDS = 7 * 86400
INTERVAL = int(os.environ.get("THERMOGRAPH_NOTIFY_INTERVAL", "900"))  # 15 min default
SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400
LOOKBACK_DAYS = 3        # re-check the last few observed days (dedup makes it safe)
FORECAST_HORIZON_DAYS = 7
# The ~45-year archive is otherwise never fetched from the background loop. When a
# subscribed cell has no cached archive yet, fetch it ONCE (it then stays cached and
# is only ever read afterwards). This caps how many missing archives a single pass
# will fetch, so a burst of new subscriptions can't hammer the strict archive quota
# — the rest are picked up on later passes.
MAX_ARCHIVE_FETCHES_PER_PASS = int(os.environ.get("THERMOGRAPH_NOTIFY_ARCHIVE_FETCHES", "4"))

# Soft wall-clock budget for one whole pass, same discipline as the archive-fetch
# cap above: checked between cells (never mid-cell) so one pathological pass — a
# slow upstream, a DB hiccup, hundreds of cells each doing real work — can't run
# unbounded and push the next pass's wake past its own interval. 2x INTERVAL by
# default: generous enough that an ordinarily-slow pass still finishes, but a
# genuinely runaway one still yields before it would overlap the next wake.
PASS_DEADLINE_SECONDS = int(os.environ.get("THERMOGRAPH_NOTIFY_PASS_DEADLINE", str(INTERVAL * 2)))

# Metrics whose low tail also counts as unusual when a subscription is two-sided
# (cold snaps, unusually calm/dry). Precipitation is one-directional (high only).
TWO_SIDED_METRICS = set(grading.TEMP_METRICS)

METRIC_NOUN = {
    "tmax": "daytime high",
    "tmin": "overnight low",
    "feels": "feels-like temperature",
    "humid": "humidity",
    "wind": "wind speed",
    "gust": "wind gusts",
    "precip": "rainfall",
}
METRIC_UNIT = {
    "tmax": "°F", "tmin": "°F", "feels": "°F",
    "humid": " g/m³", "wind": " mph", "gust": " mph", "precip": " in",
}
# A readable descriptor per (metric, direction) so titles don't read like
# "unusually high high temperature". Precipitation is high-only.
METRIC_PHRASE = {
    ("tmax", "high"): "unusually hot day", ("tmax", "low"): "unusually cool day",
    ("tmin", "high"): "unusually warm night", ("tmin", "low"): "unusually cold night",
    ("feels", "high"): "unusually hot conditions", ("feels", "low"): "unusually cold conditions",
    ("humid", "high"): "unusually humid air", ("humid", "low"): "unusually dry air",
    ("wind", "high"): "unusually strong wind", ("wind", "low"): "unusually calm wind",
    ("gust", "high"): "unusually strong gusts", ("gust", "low"): "unusually light gusts",
    ("precip", "high"): "unusually heavy rain",
}

_STOP = threading.Event()
_thread = None


# --- message building --------------------------------------------------------
def _place(sub: Subscription) -> str:
    return sub.label or f"{sub.lat:.2f}, {sub.lon:.2f}"


def _compose(sub, event_date, metric, direction, graded):
    noun = METRIC_NOUN.get(metric, metric)
    pct = graded["percentile"]
    grade = graded.get("grade")
    value = graded.get("value")
    unit = METRIC_UNIT.get(metric, "")
    phrase = METRIC_PHRASE.get((metric, direction), f"unusual {noun}")
    title = f"{_place(sub)}: {phrase}"
    val_txt = f" ({value}{unit})" if value is not None else ""
    if sub.kind == "forecast":
        body = (f"Forecast for {event_date}: the {noun} is projected at the "
                f"{pct:g}th percentile{val_txt}, {grade}.")
    else:
        body = (f"On {event_date}, the {noun} hit the {pct:g}th "
                f"percentile{val_txt}, {grade}.")
    return title, body


# --- push delivery -----------------------------------------------------------
def _deep_link(sub: Subscription, event_date: str) -> str:
    """Where a tapped notification lands — the single-day breakdown for the
    subscribed cell on the event date."""
    return f"{BASE}/day#lat={sub.lat:.5f}&lon={sub.lon:.5f}&date={event_date}"


def _push_jobs(session, sub, title, body, event_date) -> list[tuple[int, dict, dict]]:
    """Every (subscription-row id, subscription_info, payload) to push to this
    user's devices, or [] if none. DB read only — see _send_push_job for the
    actual (network) send, which the caller runs off this session entirely."""
    rows = session.execute(
        select(PushSubscription).where(PushSubscription.user_id == sub.user_id)
    ).scalars().all()
    if not rows:
        return []
    payload = {"title": title, "body": body, "url": _deep_link(sub, event_date),
               "tag": f"sub-{sub.id}"}
    return [
        (row.id, {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}, payload)
        for row in rows
    ]


def _send_push_job(job: tuple[int, dict, dict]) -> "int | None":
    """Best-effort Web Push of one gathered job. Returns the subscription-row id
    to prune if the push service reports the endpoint gone, else None. Never
    raises. Runs off the notifier thread's send pool — no DB session here."""
    row_id, info, payload = job
    return row_id if push.send(info, payload) == "gone" else None


def _discord_job(session, sub, title, body, event_date) -> "tuple[str, str, str, str] | None":
    """The (discord_id, title, body, link) to DM for this subscription, or None
    when the subscriber hasn't linked Discord or opted in. DB read only — see
    _send_discord_job for the actual send."""
    if not discord.dm_enabled():
        return None
    user = session.get(User, sub.user_id)
    if not user or not user.discord_id or not user.discord_dm:
        return None
    return (user.discord_id, title, body, _deep_link(sub, event_date))


def _send_discord_job(job: tuple[str, str, str, str]) -> None:
    """Best-effort Discord DM of one gathered job. Never raises (send_dm itself
    doesn't). Runs off the notifier thread's send pool — no DB session here."""
    discord_id, title, body, link = job
    ok = discord.send_dm(discord_id, title, body, link)
    audit.log_activity("notif.discord", {"ok": bool(ok)})


# How many push/Discord sends the notifier fans out at once. These are pure
# network I/O with no DB session involved (see _push_jobs/_discord_job above),
# so a small thread pool is all that's needed — one bad weather event touching
# hundreds of subscribers no longer stretches a pass out send-by-send.
SEND_WORKERS = int(os.environ.get("THERMOGRAPH_NOTIFY_SEND_WORKERS", "8"))

# Belt-and-suspenders on top of push.py's own send timeout: f.result() below has
# no timeout by default and would otherwise wait forever for a future that never
# resolves, which — same as an untimed webpush() call — wedges the notifier for
# the process's life under the singleton flock. Generous on purpose; this only
# needs to fire if a per-send timeout was somehow bypassed, not to race it.
SEND_RESULT_TIMEOUT = int(os.environ.get("THERMOGRAPH_NOTIFY_SEND_RESULT_TIMEOUT", "60"))


def _flush_sends(push_jobs, discord_jobs) -> list[int]:
    """Deliver every gathered push/Discord job concurrently. Returns the
    push-subscription row ids the push service reported gone, for the caller
    to prune. Blocks until every job has finished or errored, and gives up
    waiting on (but does not lose track of) any single job past
    SEND_RESULT_TIMEOUT — a pass still waits for delivery to complete, it just
    no longer does so serially, and no misbehaving future can hang the result
    loop past a bounded wait."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=SEND_WORKERS) as pool:
        futures = [pool.submit(_send_push_job, job) for job in push_jobs]
        futures += [pool.submit(_send_discord_job, job) for job in discord_jobs]
        gone = []
        for f in futures:
            try:
                result = f.result(timeout=SEND_RESULT_TIMEOUT)
            except concurrent.futures.TimeoutError:
                # f.result() gave up waiting, but the worker thread itself can't be
                # killed — ThreadPoolExecutor.__exit__ still calls shutdown(wait=True),
                # so the pool as a whole won't return until that thread actually
                # finishes. The real backstop is the per-call socket timeouts this
                # commit adds (push.py's _SEND_TIMEOUT, discord.py's _TIMEOUT): they
                # bound how long a "stuck" send can truly run. This catch just stops
                # THIS result from being reported/counted as gone, and keeps checking
                # the remaining futures instead of getting stuck on one.
                log.warning("push/Discord send exceeded %ss timeout; treating as failed", SEND_RESULT_TIMEOUT)
                continue
            except Exception:  # noqa: BLE001 - one bad send must not lose the rest
                continue
            if result is not None:
                gone.append(result)
    return gone


# --- per-cell evaluation -----------------------------------------------------
def _obs_from_row(row) -> dict:
    return {k: row[k] for k in OBS_COLS if k in row}


def _candidate_rows(recent, kind: str, today: datetime.date):
    """Rows to grade for a subscription of the given kind, most-relevant first.
    `today` is a datetime.date; `recent["date"]` is a pl.Date column."""
    if recent is None or recent.is_empty() or "date" not in recent.columns:
        return []
    if kind == "forecast":
        window = (pl.col("date") > today) & (
            pl.col("date") <= today + datetime.timedelta(days=FORECAST_HORIZON_DAYS))
        ordered = recent.filter(window).sort("date")                     # soonest first
    else:
        window = (pl.col("date") <= today) & (
            pl.col("date") >= today - datetime.timedelta(days=LOOKBACK_DAYS))
        ordered = recent.filter(window).sort("date", descending=True)    # newest first
    return list(ordered.iter_rows(named=True))


def _first_trigger(sub: Subscription, rows, history, grade_cache):
    """The first (event_date, metric, direction, graded) that crosses, or None."""
    for row in rows:
        date_key = row["date"].isoformat()
        graded = grade_cache.get(date_key)
        if graded is None:
            graded = grading.grade_day(history, row["date"], _obs_from_row(row))
            grade_cache[date_key] = graded
        for metric in sub.metrics:
            g = graded.get(metric)
            if not g or g.get("percentile") is None:
                continue
            pct = g["percentile"]
            if pct >= sub.threshold:
                return (date_key, metric, "high", g)
            if sub.two_sided and metric in TWO_SIDED_METRICS and pct <= 100 - sub.threshold:
                return (date_key, metric, "low", g)
    return None


def _process_cell(session, cell_id, subs, today, now, archive_budget):
    """Evaluate every subscription on one cell, inserting any new notifications.

    `archive_budget` is a one-element list holding the remaining number of missing
    archives this pass may fetch; it's decremented when this cell fetches one.
    """
    try:
        cell = grid.from_id(cell_id)
    except Exception:  # noqa: BLE001 - a malformed cell_id shouldn't kill the pass
        return 0
    history = climate.load_cached_history(cell)
    if history is None or history.is_empty():
        # No cached archive for this subscribed cell. Grab it ONCE (get_history
        # fetches + caches the full archive); every later pass reads it from cache
        # above and never re-fetches. Budget-gated so a burst of new subscriptions
        # can't exhaust the archive quota in a single pass.
        if archive_budget[0] <= 0:
            return 0
        archive_budget[0] -= 1
        try:
            history, _ = climate.get_history(cell)
        except climate.WeatherUnavailable:
            return 0   # archive rate-limited — retry on a later pass
        except Exception:  # noqa: BLE001
            return 0
        if history is None or history.is_empty():
            return 0
    try:
        recent = climate.get_recent_forecast(cell)
    except climate.WeatherUnavailable:
        return 0   # upstream rate-limited — skip this cell this pass
    except Exception:  # noqa: BLE001
        return 0

    grade_cache = {}
    created = 0
    push_jobs = []
    discord_jobs = []
    for sub in subs:
        # Weekly cap: one notification per subscription per 7 days.
        if sub.last_notified_at and now - sub.last_notified_at < WEEK_SECONDS:
            continue
        rows = _candidate_rows(recent, sub.kind, today)
        hit = _first_trigger(sub, rows, history, grade_cache)
        if hit is None:
            continue
        event_date, metric, direction, graded = hit
        title, body = _compose(sub, event_date, metric, direction, graded)
        notif = Notification(
            user_id=sub.user_id,
            subscription_id=sub.id,
            event_date=event_date,
            metric=metric,
            direction=direction,
            kind=sub.kind,
            percentile=graded["percentile"],
            value=graded.get("value"),
            grade=graded.get("grade"),
            title=title,
            body=body,
            channel="inapp",
            created_at=now,
        )
        session.add(notif)
        try:
            session.flush()   # trips the UNIQUE dedup constraint if already seen
        except Exception:  # noqa: BLE001 - already notified for this exact event
            session.rollback()
            continue
        sub.last_notified_at = now
        created += 1
        audit.log_activity("notif.emit", {"user_id": str(sub.user_id), "sub_id": sub.id,
                                          "metric": metric, "direction": direction,
                                          "kind": sub.kind, "percentile": graded["percentile"]})
        # Gather who to reach over Web Push and Discord DM (the in-app row above
        # is the record of truth for the bell; these are side-channels that must
        # never break it) — the actual sends happen after commit, see below.
        try:
            push_jobs.extend(_push_jobs(session, sub, title, body, event_date))
        except Exception:  # noqa: BLE001 - gathering must not break the DB write
            pass
        try:
            job = _discord_job(session, sub, title, body, event_date)
            if job is not None:
                discord_jobs.append(job)
        except Exception:  # noqa: BLE001 - gathering must not break the DB write
            pass
        # Mirror into this deployment's subscription channel (dev/uat/prod), if one
        # is configured — a shared live feed of the notifier, independent of any
        # single user's DM opt-in. Best-effort, isolated like the channels above.
        try:
            discord.post_subscription_alert(title, body, _deep_link(sub, event_date))
        except Exception:  # noqa: BLE001 - channel post stays isolated too
            pass
    session.commit()

    # Deliver every push/Discord job gathered above concurrently, only now that
    # the notifications they're for are actually committed. A single cell can
    # carry hundreds of subscribers during a broad weather event, so this also
    # avoids sending one at a time and stretching the pass past its interval.
    if push_jobs or discord_jobs:
        gone = _flush_sends(push_jobs, discord_jobs)
        if gone:
            with sync_session_maker() as cleanup:
                cleanup.execute(delete(PushSubscription).where(PushSubscription.id.in_(gone)))
                cleanup.commit()

    return created


# --- housekeeping ------------------------------------------------------------
def _cleanup_sessions(session) -> None:
    """Drop expired login sessions (access tokens past their lifetime)."""
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
        seconds=SESSION_TTL_SECONDS
    )
    session.execute(delete(AccessToken).where(AccessToken.created_at < cutoff))
    session.commit()


# --- pass + loop -------------------------------------------------------------
def run_pass() -> int:
    """One full evaluation sweep over all active subscriptions. Returns the number
    of notifications created."""
    t0 = time.perf_counter()
    now = time.time()
    today = datetime.date.today()
    created = 0
    subs: list = []
    by_cell: dict[str, list] = {}
    archive_budget = [MAX_ARCHIVE_FETCHES_PER_PASS]   # missing archives to fetch this pass
    with sync_session_maker() as session:
        subs = session.execute(
            select(Subscription).where(Subscription.active.is_(True))
        ).scalars().all()
        by_cell: dict[str, list] = {}
        for sub in subs:
            by_cell.setdefault(sub.cell_id, []).append(sub)
        cells_skipped = 0
        for i, (cell_id, cell_subs) in enumerate(by_cell.items()):
            # Checked BETWEEN cells, not inside _process_cell — a soft budget, not
            # a hard per-item timeout. Once tripped, every remaining cell this pass
            # is skipped outright (not attempted-and-abandoned mid-flight) so a
            # skipped cell's subscriptions are simply picked up whole on the next
            # pass, same as a rate-limited one already is.
            if time.perf_counter() - t0 > PASS_DEADLINE_SECONDS:
                cells_skipped = len(by_cell) - i
                log.warning("notifier pass exceeded %ss deadline; skipping %d/%d remaining cell(s)",
                           PASS_DEADLINE_SECONDS, cells_skipped, len(by_cell))
                break
            try:
                created += _process_cell(session, cell_id, cell_subs, today, now, archive_budget)
            except Exception:  # noqa: BLE001 - one bad cell must not abort the pass
                session.rollback()
        try:
            _cleanup_sessions(session)   # opportunistic expired-token sweep
        except Exception:  # noqa: BLE001
            session.rollback()
    audit.log_activity("notify.pass", {
        "subs_evaluated": len(subs), "cells": len(by_cell), "created": created,
        "archives_fetched": MAX_ARCHIVE_FETCHES_PER_PASS - archive_budget[0],
        "cells_skipped_deadline": cells_skipped,
        "duration_ms": round((time.perf_counter() - t0) * 1000.0, 1)})
    return created


# The homepage's "unusual right now" feed rides along on this loop rather than
# running its own thread: this one already wakes on a timer and already carries
# the single-leader invariant a second daemon would have to re-establish. The
# sweep is cache-only, so it costs no upstream quota.
HOMEPAGE_REFRESH_INTERVAL = 3600.0
_last_homepage_refresh = 0.0


def _maybe_refresh_homepage() -> None:
    global _last_homepage_refresh
    now = time.time()
    if now - _last_homepage_refresh < HOMEPAGE_REFRESH_INTERVAL:
        return
    _last_homepage_refresh = now
    try:
        homepage.refresh()
    except Exception:  # noqa: BLE001 - the feed is optional; the homepage renders without it
        pass


# Pre-warming the SEO content derived-store rides this loop too, for the same
# reason the homepage sweep does: it's leader-only (the loop runs on the single
# elected leader) and cache-only, so it spends no upstream quota. The content
# token turns over when a cell's archive gains a day (~1x/day), so after each
# advance the first request to a /climate page would otherwise recompute a
# 45-year payload cold; warming rebuilds those rows off-request. It's paced and
# capped per tick (CONTENT_WARM_MAX_CITIES) so one tick can't stall the notifier,
# and the idempotent skip means a tick after everything is fresh is a cheap
# no-op — the ~1000-city set refreshes across a handful of ticks, well inside the
# <=1-day staleness the content token already tolerates.
CONTENT_WARM_INTERVAL = float(os.environ.get("THERMOGRAPH_CONTENT_WARM_INTERVAL", "1800"))  # 30 min
CONTENT_WARM_MAX_CITIES = int(os.environ.get("THERMOGRAPH_CONTENT_WARM_MAX_CITIES", "50"))
_last_content_warm = 0.0


def _maybe_warm_content() -> None:
    global _last_content_warm
    now = time.time()
    if now - _last_content_warm < CONTENT_WARM_INTERVAL:
        return
    _last_content_warm = now
    try:
        # Local import: warm_cities is a script-style top-level module (mirrors
        # api/internal_routes.py's warm-cities job), so only the leader pays its
        # import cost, and it's kept off notify.py's import graph.
        import warm_cities
        warm_cities.warm_content(limit=CONTENT_WARM_MAX_CITIES)
    except Exception:  # noqa: BLE001 - warming is best-effort; pages self-heal on request
        pass


# The daily "most unusual right now" post to Discord rides this loop too: once per
# day, after the feed is refreshed, leader-only like everything in run_loop. No
# webhook configured => no-op.
_last_discord_post_day: str | None = None


def _maybe_post_discord() -> None:
    global _last_discord_post_day
    if not discord.enabled():
        return
    today = datetime.date.today().isoformat()
    if _last_discord_post_day == today:
        return
    # Only record the day as done once a post actually lands, so a transient
    # failure retries on the next wake instead of silently skipping the day.
    if discord.post_daily_feed():
        _last_discord_post_day = today


def run_loop():
    # Beat once at startup so the dashboard shows the notifier alive immediately after a
    # (re)start, without waiting a full INTERVAL for the first pass. Then beat every wake:
    # healthy => a fresh beat at most INTERVAL apart, which is how the dashboard tells this
    # single-leader daemon apart from a genuinely dead one (per-worker thread checks can't).
    metrics.record_heartbeat("subscription-notifier", INTERVAL)
    audit.log_heartbeat("subscription-notifier", INTERVAL)
    # Wait first, then run — gives the app a moment to finish booting. The whole
    # body is guarded: this is a daemon thread, so ANY escaping exception (a bad
    # pass, a homepage refresh, a Discord post) would silently kill it for the
    # process's lifetime — no notifier, no heartbeat, no restart. Swallow per
    # iteration instead; the next wake retries and keeps the heartbeat fresh.
    while not _STOP.wait(INTERVAL):
        metrics.record_heartbeat("subscription-notifier", INTERVAL)
        # Log-shipped twin of the in-process metric above, so the observability
        # stack (Loki/Grafana) can show notifier liveness straight from the logs.
        audit.log_heartbeat("subscription-notifier", INTERVAL)
        try:
            run_pass()
            _maybe_refresh_homepage()
            _maybe_warm_content()
            _maybe_post_discord()
        except Exception:  # noqa: BLE001 - a bad iteration must never kill the loop
            pass


def start():
    """Start the notifier daemon thread (unless disabled via env)."""
    global _thread
    if os.environ.get("THERMOGRAPH_ENABLE_NOTIFIER", "1") == "0":
        return
    if _thread is not None:
        return
    _STOP.clear()
    _thread = threading.Thread(target=run_loop, name="subscription-notifier", daemon=True)
    _thread.start()


def stop():
    _STOP.set()