thermograph/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 datetime
import os
import threading
import time

import polars as pl
from sqlalchemy import delete, select

import climate
import discord
import grading
import grid
import homepage
import metrics
import push
from db import sync_session_maker
from models import AccessToken, Notification, PushSubscription, Subscription, User
from views import OBS_COLS

_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"))

# 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 _dispatch_push(session, sub, title, body, event_date) -> None:
    """Best-effort Web Push of a just-created notification to all of the user's
    devices. Prunes endpoints the push service reports gone. Runs in the notifier
    daemon thread (no event loop), so the blocking send is fine here. Never raises
    — the in-app notification is already committed regardless."""
    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}"}
    for row in rows:
        info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
        if push.send(info, payload) == "gone":
            session.delete(row)   # committed with the rest of the pass


def _dispatch_discord(session, sub, title, body, event_date) -> None:
    """Best-effort Discord DM of a just-created notification, when the subscriber
    has linked Discord and opted in. Isolated like the push side-channel — the
    in-app row is already committed, and push/email remain the fallback for anyone
    not reachable on Discord. Runs in the notifier thread (no event loop)."""
    if not discord.dm_enabled():
        return
    user = session.get(User, sub.user_id)
    if not user or not user.discord_id or not user.discord_dm:
        return
    discord.send_dm(user.discord_id, title, body, _deep_link(sub, event_date))


# --- 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
    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
        # Additionally deliver over Web Push (the in-app row above is the record of
        # truth for the bell; push is a side-channel that must never break it).
        try:
            _dispatch_push(session, sub, title, body, event_date)
        except Exception:  # noqa: BLE001 - push failures stay isolated from the DB write
            pass
        # And, independently, a Discord DM for linked opted-in users.
        try:
            _dispatch_discord(session, sub, title, body, event_date)
        except Exception:  # noqa: BLE001 - Discord failures stay isolated too
            pass
    session.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."""
    now = time.time()
    today = datetime.date.today()
    created = 0
    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)
        for cell_id, cell_subs in by_cell.items():
            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()
    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


# 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)
    # Wait first, then run — gives the app a moment to finish booting.
    while not _STOP.wait(INTERVAL):
        metrics.record_heartbeat("subscription-notifier", INTERVAL)
        try:
            run_pass()
        except Exception:  # noqa: BLE001 - a bad pass must never kill the loop
            pass
        _maybe_refresh_homepage()
        _maybe_post_discord()


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()