diff --git a/api/routes/anchors.py b/api/routes/anchors.py index 1cbc78d1..969baee0 100644 --- a/api/routes/anchors.py +++ b/api/routes/anchors.py @@ -1,3 +1,6 @@ +import logging +from datetime import datetime + from fastapi import APIRouter, Depends, Request from pydantic import BaseModel from typing import Literal @@ -7,10 +10,47 @@ from bot.crontab import sync_crontab from api.ws import manager from api.auth import auth_dependency +from shared import notify_due + +logger = logging.getLogger(__name__) router = APIRouter() +async def _refresh_anchor_due_cache(conn: asyncpg.Connection, user_id: str) -> None: + """Refresh the cached anchor schedule + the "anchor" Redis due-component + for *user_id* after any anchor create/update/delete. + + Anchor edits are rare and already happen inside an authenticated, + synchronous request — recomputing here (rather than waiting for the + next real notification check) keeps the gate accurate immediately + instead of lagging behind an edit by up to the safety TTL. See + shared/notify_due.py for the full gating scheme. + + Best-effort: this is a cache-warming side effect of a successful anchor + mutation, not part of the mutation itself. ANY failure here (a Postgres + blip re-reading anchors, a malformed row, Redis being unreachable) must + never turn an already-committed create/update/delete into a user-facing + error, and must never suppress the ``anchors_updated`` WS broadcast the + caller sends right after this. Errors are logged and swallowed — the + gate simply stays stale until the next real notification check + self-heals it (or the safety TTL expires it), which is a performance + detail, not a correctness one. + """ + try: + anchors = await get_anchors(conn) + await notify_due.set_cached_anchors(user_id, anchors) + next_boundary = notify_due.next_anchor_boundary(anchors, datetime.now()) + if next_boundary is not None: + await notify_due.set_component_due(user_id, "anchor", next_boundary.timestamp()) + except Exception: + logger.warning( + "anchor due-cache refresh failed for user_id=%s — mutation already " + "committed; gate will self-heal on the next real notification check", + user_id, exc_info=True, + ) + + class AnchorUpdate(BaseModel): name: str time: str @@ -38,6 +78,7 @@ async def create_anchor(body: AnchorUpdate, request: Request, anchor = {"id": anchor_id, **body.model_dump()} await upsert_anchor(conn, anchor) await sync_crontab(request.app.state.pool, request.state.user_id) + await _refresh_anchor_due_cache(conn, request.state.user_id) await manager.broadcast({"type": "anchors_updated"}, request.state.user_id) return anchor @@ -49,6 +90,7 @@ async def update_anchor(anchor_id: str, body: AnchorUpdate, request: Request, anchor = {"id": anchor_id, **body.model_dump()} await upsert_anchor(conn, anchor) await sync_crontab(request.app.state.pool, request.state.user_id) + await _refresh_anchor_due_cache(conn, request.state.user_id) await manager.broadcast({"type": "anchors_updated"}, request.state.user_id) return anchor @@ -59,5 +101,6 @@ async def delete_anchor_route(anchor_id: str, request: Request, conn: asyncpg.Connection = Depends(get_db_conn)): await delete_anchor(conn, anchor_id) await sync_crontab(request.app.state.pool, request.state.user_id) + await _refresh_anchor_due_cache(conn, request.state.user_id) await manager.broadcast({"type": "anchors_updated"}, request.state.user_id) return {"ok": True} diff --git a/api/routes/internal.py b/api/routes/internal.py index 3fa0fb34..6286d084 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -1,6 +1,16 @@ """Internal API routes — not mounted in OpenAPI schema. Called by Fly.io cron machines. All endpoints require X-Internal-Token header. + +Neon/idle-spin-down note +------------------------- +``_run_notification_check`` is gated by ``shared.notify_due`` — a Redis +next-due cache — so that regardless of how often this endpoint is actually +invoked (by whatever scheduler ends up calling it), it costs nothing but a +single Redis range query when no user has any anchor/followup/meeting-event +work due right now. See shared/notify_due.py for the full scheme. This makes +the endpoint safe to call as frequently as desired without keeping managed +Postgres (Neon) awake between real due events. """ from __future__ import annotations @@ -9,11 +19,14 @@ import logging import os import secrets +import time +from datetime import datetime from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from bot.message_handler import check_followups from bot import notify +from shared import notify_due import db.postgres as pg import db.pg_auth_queries as pg_auth_queries @@ -76,22 +89,49 @@ async def _get_all_linked_users(pool) -> list[dict]: return [{"id": str(r["id"]), "telegram_chat_id": r["telegram_chat_id"]} for r in rows] -async def _check_anchor_transitions(pool, user_id: str, dispatch_fn) -> None: - """Check and fire any due anchor transitions for the given user.""" - from datetime import datetime +async def _get_linked_users_by_id(pool, user_ids: list[str]) -> list[dict]: + """Return linked-user rows scoped to *user_ids* — used once the Redis + next-due gate has already told us which users are actually due, so we + don't pay for a full-table scan on every notification-check invocation.""" + if not user_ids: + return [] + async with pg.get_conn(pool) as conn: + rows = await conn.fetch( + """ + SELECT u.id, tc.telegram_chat_id + FROM users u + JOIN telegram_connections tc ON tc.user_id = u.id + WHERE tc.telegram_chat_id IS NOT NULL + AND tc.telegram_chat_id != '' + AND u.id = ANY($1::uuid[]) + """, + user_ids, + ) + return [{"id": str(r["id"]), "telegram_chat_id": r["telegram_chat_id"]} for r in rows] + + +async def _check_anchor_transitions(pool, user_id: str, dispatch_fn) -> datetime | None: + """Check and fire any due anchor transitions for the given user. + + Returns the next anchor-schedule boundary (start/end of a time block) + strictly after ``now``, or ``None`` if the user has no anchors at all. + Used by the caller to populate the Redis "anchor" due-component (see + shared/notify_due.py) — computed from data already fetched here, no + extra Postgres cost. + """ from bot.handler_utils import get_current_anchor, is_anchor_active from bot.message_handler import _get_anchors_and_plan, _init_followup_states - from db.pg_queries.anchors import get_anchors from bot.anchor_trigger import trigger_anchor + now = datetime.now() try: from datetime import date as _date today = str(_date.today()) anchors, plan = await _get_anchors_and_plan(pool, user_id, today) + await notify_due.set_cached_anchors(user_id, anchors) current_anchor = get_current_anchor(anchors) if current_anchor and is_anchor_active(current_anchor): if plan and current_anchor["id"] in plan.get("anchors", {}): - now = datetime.now() await _init_followup_states( pool, user_id, today, current_anchor["id"], [t["id"] for t in plan["anchors"][current_anchor["id"]]["tasks"] if t.get("id")], @@ -103,32 +143,116 @@ async def _check_anchor_transitions(pool, user_id: str, dispatch_fn) -> None: user_id=user_id, dispatch_fn=dispatch_fn, ) - except Exception as e: - logger.warning("Anchor transition check failed for user %s: %s", user_id, e) + return notify_due.next_anchor_boundary(anchors, now) + except Exception: + # Broad catch pre-dates this gating change (any downstream call — + # dispatch_fn, trigger_anchor — can raise many exception types this + # function can't enumerate). Logged at ERROR (not WARNING) with a + # traceback because a bug here is now indistinguishable, at the + # call site, from "this user simply has no anchors" — the caller + # branches on this returning None either way. This does not cause + # under-notification (the stale cached due-time just stays in the + # past, so the user is rechecked again next cycle), but a + # persistent failure here should be loud in logs rather than a + # quiet recurring WARNING. + logger.error( + "Anchor transition check failed for user_id=%s", user_id, exc_info=True + ) + return None + + +async def _recompute_and_cache_due( + user_id: str, anchor_next: datetime | None, followup_next: datetime | None +) -> None: + """Write the freshest anchor/followup next-due estimates back to Redis. + + Called once per user after a real check has run — self-perpetuating: + each real run recomputes its own future due time from data it already + fetched, so the cache never needs a separate background refresh job. + The "meeting" component is contributed independently by tether-premium's + drain_meeting_events via the same shared.notify_due.set_component_due — + this function only owns the anchor/followup components. + """ + now_ts = time.time() + if anchor_next is not None: + await notify_due.set_component_due(user_id, "anchor", anchor_next.timestamp()) + else: + # No anchors configured — nothing to re-check on a timer for this + # component; leave it unset (absent from the hash) rather than + # writing a synthetic "never" value, since a future anchor + # create/update will populate it and recompute the combined score. + pass + if followup_next is not None: + await notify_due.set_component_due(user_id, "followup", followup_next.timestamp()) + else: + # No active followup rows right now — the anchor component's next + # boundary (which triggers _init_followup_states) is what will next + # wake this user for the followup side too. + pass + if anchor_next is None and followup_next is None: + logger.debug( + "notify_due: no anchor or followup component for user_id=%s — " + "relying on existing cache entries / fail-open on next check", + user_id, + ) + _ = now_ts # reserved for future observability (e.g. staleness metrics) async def _run_notification_check(pool, ws_manager) -> None: - """For each linked user: check anchor transitions, follow-ups, meeting events.""" + """For each linked user: check anchor transitions, follow-ups, meeting events. + + Gated by shared.notify_due: a single Redis range query decides which + users (if any) have real work due right now. When nothing is due, this + function returns without touching Postgres at all — see module docstring. + """ if pool is None: logger.warning("_run_notification_check: no pool available, skipping") return - try: - users = await _get_all_linked_users(pool) - except Exception as e: - logger.error("_run_notification_check: failed to load users: %s", e) + now_ts = time.time() + due_user_ids = await notify_due.get_due_user_ids(now_ts) + + if due_user_ids is None: + # Redis gating unavailable — fail open to the original, unfiltered + # behaviour rather than skipping notifications. ERROR (not WARNING): + # a one-off blip here is harmless, but if this fires on EVERY + # invocation it means gating has silently stopped working and the + # endpoint is back to a full Postgres scan every call — exactly the + # Neon-idle-spin-down regression this feature exists to prevent. + # There's no metric/counter for "N consecutive fail-opens" yet + # (follow-up item), so ERROR-level log visibility is the current + # signal that something needs attention. + logger.error( + "notify.check_gating_unavailable — falling back to full linked-user scan" + ) + try: + users = await _get_all_linked_users(pool) + except Exception as e: + logger.error("_run_notification_check: failed to load users: %s", e) + return + elif not due_user_ids: + logger.debug("notify.check_skipped_empty") return + else: + logger.info("notify.check_processed n=%d", len(due_user_ids)) + try: + users = await _get_linked_users_by_id(pool, due_user_ids) + except Exception as e: + logger.error("_run_notification_check: failed to load due users: %s", e) + return for user in users: user_id = str(user["id"]) + anchor_next: datetime | None = None try: dispatch_fn = functools.partial( notify.dispatch, pool=pool, ws_manager=ws_manager ) - await _check_anchor_transitions(pool, user_id, dispatch_fn) + anchor_next = await _check_anchor_transitions(pool, user_id, dispatch_fn) except Exception as e: logger.warning("Anchor check failed for user %s: %s", user_id, e) + followup_next: datetime | None = None try: async def _notify_send(text, uid=user_id): await notify.dispatch( @@ -141,7 +265,7 @@ async def _notify_send(text, uid=user_id): ) send_fn = lambda text, uid=user_id: asyncio.ensure_future(_notify_send(text, uid)) - await check_followups(pool, user_id, send_fn) + followup_next = await check_followups(pool, user_id, send_fn) except Exception as e: logger.warning("Followup check failed for user %s: %s", user_id, e) @@ -167,6 +291,8 @@ async def _notify_meeting_send(text, uid=user_id): except Exception as e: logger.warning("Meeting event drain failed for user %s: %s", user_id, e) + await _recompute_and_cache_due(user_id, anchor_next, followup_next) + async def _renew_expiring_watches(pool) -> None: """Renew expiring Google Calendar watch channels (daily cron).""" diff --git a/bot/message_handler.py b/bot/message_handler.py index 9ba5f91c..091e20d1 100644 --- a/bot/message_handler.py +++ b/bot/message_handler.py @@ -55,6 +55,7 @@ create_node, ) import db.postgres as pg +from shared import notify_due import db.pg_auth_queries as pg_auth_queries from cryptography.fernet import Fernet @@ -325,11 +326,33 @@ async def apply_mutations(mutations: list[dict], pool, user_id: str, today: str) logger.error("Failed to apply mutation %s: %s", m, e) -async def check_followups(pool, user_id: str, send_fn) -> None: - """Called every polling cycle. Sends batched pre/post-ack messages for due tasks.""" - from datetime import datetime, date - now = datetime.now() - today = str(date.today()) +def _parse_ts(ts_str: str | None, *, default: "datetime") -> "datetime": + from datetime import datetime + if not ts_str: + return default + try: + return datetime.fromisoformat(ts_str) + except ValueError: + return default + + +def classify_followup_row(row: dict, config: dict, now: "datetime") -> tuple[str | None, "datetime | None"]: + """Pure classification of a single followup_state row against its resolved + config: is it due right now, and if not (or after it fires), when's the + next candidate check time? + + Returns ``(queue, next_candidate)``: + - ``queue`` is ``"pre"``/``"post"`` if the row is due right now for that + ack phase, else ``None``. + - ``next_candidate`` is the earliest ``datetime`` this row could next + become due — ``None`` if the row has exhausted its allotted pings for + its current phase (no further automatic check needed for it). + + No I/O — depends only on the row, its config, and the current time, so + the next-due estimate used for Redis gating (see shared/notify_due.py) + is independently testable without a database. + """ + from datetime import datetime, timedelta def minutes_since(ts_str: str | None) -> float: if not ts_str: @@ -340,12 +363,55 @@ def minutes_since(ts_str: str | None) -> float: return float('inf') return (now - ts).total_seconds() / 60 + if row["acknowledged_at"] is None: + ref_ts = row["last_ping_at"] or row["sequence_started_at"] + if row["pre_ack_pings_sent"] >= config["pre_ack_max_pings"]: + return None, None + if minutes_since(ref_ts) >= config["pre_ack_interval_min"]: + # Due now — will be pinged this pass. The next candidate (one + # interval after "now") is an approximation: the precise value + # depends on the ping actually being recorded by the caller. + return "pre", now + timedelta(minutes=config["pre_ack_interval_min"]) + return None, _parse_ts(ref_ts, default=now) + timedelta(minutes=config["pre_ack_interval_min"]) + else: + ref_ts = row["last_ping_at"] or row["acknowledged_at"] + if row["post_ack_pings_sent"] >= config["post_ack_pings"]: + return None, None + if minutes_since(ref_ts) >= config["post_ack_interval_min"]: + return "post", now + timedelta(minutes=config["post_ack_interval_min"]) + return None, _parse_ts(ref_ts, default=now) + timedelta(minutes=config["post_ack_interval_min"]) + + +async def check_followups(pool, user_id: str, send_fn): + """Called every polling cycle. Sends batched pre/post-ack messages for due tasks. + + Returns the earliest ``datetime`` at which a currently-active-but-not-yet-due + followup row could next become due (or ``None`` if there are no active rows + at all, i.e. the followup component is entirely governed by the next anchor + trigger). Callers use this to populate the Redis next-due gate + (``shared.notify_due.set_component_due(user_id, "followup", ...)``) so a + future invocation can skip this Postgres round-trip entirely when nothing + is due yet — see shared/notify_due.py for the full gating scheme. + """ + from datetime import datetime, date + now = datetime.now() + today = str(date.today()) + + next_candidate: datetime | None = None + + def _consider(candidate: datetime | None) -> None: + nonlocal next_candidate + if candidate is None: + return + if next_candidate is None or candidate < next_candidate: + next_candidate = candidate + # Phase 1: load all data (connection held only for reads + config lookups) async with pg.get_conn(pool, user_id) as conn: rows = await get_active_followup_states(conn, today) if not rows: logger.debug("check_followups: no active followup states for %s", today) - return + return None pre_ack_due = [] post_ack_due = [] @@ -365,16 +431,12 @@ def minutes_since(ts_str: str | None) -> float: logger.debug("check_followups: skipping task %s — no followup config", row["task_id"]) continue - if row["acknowledged_at"] is None: - ref_ts = row["last_ping_at"] or row["sequence_started_at"] - if (row["pre_ack_pings_sent"] < config["pre_ack_max_pings"] - and minutes_since(ref_ts) >= config["pre_ack_interval_min"]): - pre_ack_due.append(row) - else: - ref_ts = row["last_ping_at"] or row["acknowledged_at"] - if (row["post_ack_pings_sent"] < config["post_ack_pings"] - and minutes_since(ref_ts) >= config["post_ack_interval_min"]): - post_ack_due.append(row) + queue, candidate = classify_followup_row(row, config, now) + _consider(candidate) + if queue == "pre": + pre_ack_due.append(row) + elif queue == "post": + post_ack_due.append(row) plan = await get_plan(conn, today) if (pre_ack_due or post_ack_due) else {} @@ -428,6 +490,8 @@ def minutes_since(ts_str: str | None) -> float: for row in anchor_rows: await record_ping(conn, row["id"], "post", now) + return next_candidate + # --------------------------------------------------------------------------- # v2 Pipeline: helpers @@ -1423,11 +1487,41 @@ def run_polling(token: str, chat_id: str, poll_user_id: str | None = None) -> No _active_chat = last_chat_id if last_chat_id else chat_id _send = lambda m, cid=_active_chat: _send_telegram(token, cid, m) + # Meeting events are PUSH-based (arrive via the WS listener into + # an in-process queue at unpredictable times), not time-based — + # there's no "next_due timestamp" to precompute for them, so they + # don't fit the Redis due-time gate below at all. drain_meeting_events + # already self-gates for free (checks its queue is non-empty before + # ever touching Postgres — see tether_premium docstring), so it is + # deliberately EXEMPT from the is_due() gate and always called: if + # it were nested inside the gate, a meeting event could arrive for + # a user with no anchor/followup due and sit un-drained until + # something else made that user "due" again. if last_user_id: + try: + from tether_premium.bot.scheduling.events import drain_meeting_events + loop.run_until_complete(drain_meeting_events(pool=pool, user_id=last_user_id, send_fn=_send)) + except ImportError: + pass + except Exception as _de: + logger.warning("Meeting event drain failed: %s", _de) + + # Redis next-due gate — this loop ticks every ~30s (Telegram + # long-poll timeout) regardless of whether anything is actually + # due; without this check every tick would hit Postgres 2-3x via + # _get_anchors_and_plan/check_followups even when idle, defeating + # managed-Postgres (Neon) auto-suspend. See shared/notify_due.py + # for the full scheme. Gated ONCE per user (not per sub-check) — + # anchor-transition can create new followup rows that + # check_followups then needs to see in the SAME pass, so splitting + # the gate per sub-function risks skipping just-created rows via a + # stale followup-only cache entry. + if last_user_id and loop.run_until_complete(notify_due.is_due(last_user_id)): _anchors_and_plan = loop.run_until_complete( _get_anchors_and_plan(pool, last_user_id, _today) ) _anchors, _plan = _anchors_and_plan + loop.run_until_complete(notify_due.set_cached_anchors(last_user_id, _anchors)) _current_anchor = get_current_anchor(_anchors) _anchor_running = _current_anchor and is_anchor_active(_current_anchor) if not _anchor_running: @@ -1441,16 +1535,21 @@ def run_polling(token: str, chat_id: str, poll_user_id: str | None = None) -> No _dt.now() )) - # Run follow-up pings for the active user - if last_user_id: - loop.run_until_complete(check_followups(pool, last_user_id, _send)) - try: - from tether_premium.bot.scheduling.events import drain_meeting_events - loop.run_until_complete(drain_meeting_events(pool=pool, user_id=last_user_id, send_fn=_send)) - except ImportError: - pass - except Exception as _de: - logger.warning("Meeting event drain failed: %s", _de) + _followup_next = loop.run_until_complete(check_followups(pool, last_user_id, _send)) + + # Recompute-after-run: write fresh anchor/followup next-due + # estimates back to Redis from data just fetched above (zero + # extra Postgres cost). Self-perpetuating — no separate + # background refresh job needed. + _anchor_next = notify_due.next_anchor_boundary(_anchors, _dt.now()) + if _anchor_next is not None: + loop.run_until_complete( + notify_due.set_component_due(last_user_id, "anchor", _anchor_next.timestamp()) + ) + if _followup_next is not None: + loop.run_until_complete( + notify_due.set_component_due(last_user_id, "followup", _followup_next.timestamp()) + ) except Exception as e: logger.error("Polling error: %s", e) time.sleep(5) diff --git a/shared/notify_due.py b/shared/notify_due.py new file mode 100644 index 00000000..32ce56bb --- /dev/null +++ b/shared/notify_due.py @@ -0,0 +1,340 @@ +"""Redis next-due gating for notification checks (anchors/followups/meetings). + +Purpose +------- +`_run_notification_check` (api/routes/internal.py) and the polling-loop +inline block (bot/message_handler.py `run_polling`) both need to answer +"is there any real notification work due for this user right now?" without +paying a Postgres round-trip to find out — otherwise every invocation +(regardless of caller frequency) keeps managed Postgres (Neon) awake even +when the app is completely idle. + +This module caches, per user, the earliest time at which either of two +TIME-based components next becomes due: + + - ``anchor`` — next anchor-schedule boundary (start/end of a time block) + - ``followup`` — next pre/post-ack followup ping + +Each component is written independently (``set_component_due``) by whichever +code just computed a fresh estimate for it. This means a single Redis query +(``get_due_user_ids``) — done by the cron-style entry point — sees a user +as due if ANY one component says so, regardless of which side wrote it. + +Meeting events are NOT modeled as a component here. They're PUSH-based +(delivered via an in-process WS-listener queue owned by tether-premium's +``tether_premium.bot.scheduling.events``, at unpredictable times) rather +than time-based, so there's no meaningful "next_due timestamp" to +precompute for them — they don't fit this module's timestamp-gating model. +``drain_meeting_events`` already self-gates for free (it checks its queue +is non-empty before ever touching Postgres) and is deliberately called +UNCONDITIONALLY by ``run_polling``, exempt from the ``is_due()`` gate here +— see the call site in bot/message_handler.py for the reasoning. The +``"meeting"`` value below is reserved in the ``Component`` type for +possible future use (e.g. if the event queue is ever moved to a shared/ +cross-process backing store), but nothing currently writes it, and none +of the gates in this repo currently depend on it. + +Redis layout +------------ +- ``notify:due:{user_id}`` — HASH, fields are component names, values are + Unix timestamps (float, as strings). This is the durable per-component + record. +- ``notify:due_queue`` — ZSET, member=user_id, score=min() of that user's + known component values. This is the only thing the cheap "who's due" + query touches — an O(log N) range scan, no per-user Postgres reads. + +Fail-open philosophy +--------------------- +Redis is a cache, not a source of truth — the real due-ness is always +whatever Postgres/application state says. If Redis is unavailable (no +``REDIS_URL``, connection error) or a user has never been cached before +(cold start, new link, post-flush), every read here fails OPEN: treat the +user as due. This guarantees gating can only ever save a round-trip it +would otherwise have made — it can never *cause* a real notification to +be skipped. Writes fail silently (logged) for the same reason: a dropped +cache write just means "we'll re-check sooner than strictly necessary," +never "we silently stopped checking." + +``get_due_user_ids`` distinguishes "Redis unavailable" (returns ``None`` — +caller should fall back to the old unfiltered behaviour) from "Redis +available, confirmed nothing due" (returns ``[]``) — these must not be +conflated, or callers could never safely skip work. +""" +from __future__ import annotations + +import logging +import os +import time as _time +from datetime import datetime, timedelta +from typing import Any, Literal + +logger = logging.getLogger(__name__) + +_DUE_QUEUE_KEY = "notify:due_queue" +_DUE_HASH_PREFIX = "notify:due" +_ANCHORS_KEY_PREFIX = "notify:anchors" + +# Safety-net TTL on per-user cache entries. If a code path ever fails to +# recompute-after-run (a bug), the entry eventually expires and reads fall +# back to the "unknown user" fail-open path rather than staying wrong forever. +_SAFETY_TTL_SECONDS = 24 * 3600 + +Component = Literal["anchor", "followup", "meeting"] + + +def get_redis_url() -> str | None: + """Return REDIS_URL from environment, or None if unset.""" + return os.environ.get("REDIS_URL") or None + + +def _due_hash_key(user_id: str) -> str: + return f"{_DUE_HASH_PREFIX}:{user_id}" + + +def anchors_cache_key(user_id: str) -> str: + """Redis key for a user's cached anchor schedule (used by callers that + refresh the anchor cache on anchor CRUD).""" + return f"{_ANCHORS_KEY_PREFIX}:{user_id}" + + +async def _get_client( + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> Any: + """Build (or reuse) an async Redis client. Returns None if unavailable.""" + if redis_client is not None: + return redis_client + if server is not None: + import fakeredis.aioredis as faioredis + + return faioredis.FakeRedis(server=server) + url = redis_url or get_redis_url() + if url is None: + return None + import redis.asyncio as aioredis + + return aioredis.from_url(url) + + +async def set_component_due( + user_id: str, + component: Component, + next_ts: float, + *, + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> None: + """Record *component*'s next-due timestamp for *user_id*. + + Recomputes the combined ``due_queue`` score as the min of all known + components for this user, so the user is surfaced as due as soon as + ANY single component says so. + + Best-effort: any Redis error (including "not configured") is logged and + swallowed — a dropped cache write never blocks or breaks the caller. + """ + client = await _get_client(redis_client, redis_url, server) + if client is None: + logger.warning( + "notify_due.set_component_due: no Redis configured — " + "cache write skipped for user_id=%s component=%s", + user_id, component, + ) + return + try: + hkey = _due_hash_key(user_id) + await client.hset(hkey, component, repr(next_ts)) + await client.expire(hkey, _SAFETY_TTL_SECONDS) + raw = await client.hgetall(hkey) + values = [float(v) for v in raw.values()] + combined = min(values) if values else next_ts + await client.zadd(_DUE_QUEUE_KEY, {user_id: combined}) + except Exception: + logger.warning( + "notify_due.set_component_due: Redis error — failing open " + "(user_id=%s component=%s)", + user_id, component, exc_info=True, + ) + + +async def get_due_user_ids( + now: float | None = None, + *, + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> list[str] | None: + """Return user_ids whose combined next-due score is <= *now*. + + Returns ``None`` (the fail-open sentinel) if Redis is unavailable or + errors — callers MUST treat that distinctly from an empty list: ``None`` + means "gating unavailable, fall back to the real unfiltered check"; + ``[]`` means "gating confirms nothing is due right now." + """ + if now is None: + now = _time.time() + client = await _get_client(redis_client, redis_url, server) + if client is None: + logger.warning( + "notify_due.get_due_user_ids: no Redis configured — " + "gating unavailable, caller should fall back to unfiltered check" + ) + return None + try: + members = await client.zrangebyscore(_DUE_QUEUE_KEY, "-inf", now) + return [ + m.decode() if isinstance(m, (bytes, bytearray)) else m + for m in members + ] + except Exception: + logger.warning( + "notify_due.get_due_user_ids: Redis error — failing open", + exc_info=True, + ) + return None + + +async def is_due( + user_id: str, + now: float | None = None, + *, + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> bool: + """Cheap single-user due check — used by the polling-loop path. + + Fails open to True: an unknown user (cold start / new link), a Redis + error, or Redis being unconfigured all resolve to "due," never to + "skip." This mirrors ``get_due_user_ids``'s fail-open contract but + collapses "unavailable" and "unknown" to the same True result since a + single-user caller has no useful fallback distinction to make. + """ + if now is None: + now = _time.time() + client = await _get_client(redis_client, redis_url, server) + if client is None: + return True + try: + score = await client.zscore(_DUE_QUEUE_KEY, user_id) + if score is None: + return True # never cached — fail open + return float(score) <= now + except Exception: + logger.warning( + "notify_due.is_due: Redis error — failing open (user_id=%s)", + user_id, exc_info=True, + ) + return True + + +# --------------------------------------------------------------------------- +# Anchor schedule cache (JSON list) + pure next-boundary computation +# --------------------------------------------------------------------------- + +async def set_cached_anchors( + user_id: str, + anchors: list[dict], + *, + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> None: + """Refresh the cached anchor schedule for *user_id*. + + Called on anchor create/update/delete (rare, synchronous) and + opportunistically whenever a real anchor-transition check fetches + anchors from Postgres anyway (zero extra cost — self-healing cache). + """ + import json + + client = await _get_client(redis_client, redis_url, server) + if client is None: + logger.warning( + "notify_due.set_cached_anchors: no Redis configured — " + "cache write skipped for user_id=%s", user_id, + ) + return + try: + await client.set( + anchors_cache_key(user_id), json.dumps(anchors), ex=_SAFETY_TTL_SECONDS + ) + except Exception: + logger.warning( + "notify_due.set_cached_anchors: Redis error — failing open " + "(user_id=%s)", user_id, exc_info=True, + ) + + +async def get_cached_anchors( + user_id: str, + *, + redis_client: Any = None, + redis_url: str | None = None, + server: Any = None, +) -> list[dict] | None: + """Return the cached anchor schedule for *user_id*, or None on a cache + miss / Redis error (caller should fall back to a real Postgres fetch, + which should then call ``set_cached_anchors`` to repopulate).""" + import json + + client = await _get_client(redis_client, redis_url, server) + if client is None: + return None + try: + raw = await client.get(anchors_cache_key(user_id)) + if raw is None: + return None + return json.loads(raw) + except Exception: + logger.warning( + "notify_due.get_cached_anchors: Redis error — treating as cache miss " + "(user_id=%s)", user_id, exc_info=True, + ) + return None + + +def next_anchor_boundary(anchors: list[dict], now: datetime) -> datetime | None: + """Return the next time an anchor's active window starts or ends, strictly + after *now*. Pure function of (schedule, now) — no I/O. + + Returns None if there are no anchors at all (nothing to ever wake up + for on the anchor side — the followup component still governs). + + This recurs daily by construction: candidate boundaries are always + "the next occurrence of this time-of-day, today or tomorrow," so no + explicit midnight-rollover handling is needed. + """ + if not anchors: + return None + + candidates: list[datetime] = [] + for anchor in anchors: + try: + h, m = map(int, anchor["time"].split(":")) + duration = anchor.get("duration_minutes", 0) + except (KeyError, ValueError, AttributeError, TypeError): + # A malformed anchor row (bad/missing "time") must not crash the + # whole computation for every other anchor — skip it. This keeps + # `next_anchor_boundary` a safe pure function callers can rely on + # even against imperfect data, rather than requiring every call + # site to defend against it individually. + logger.warning( + "notify_due.next_anchor_boundary: skipping malformed anchor row: %r", + anchor, + ) + continue + for day_offset in (0, 1): + day = (now + timedelta(days=day_offset)).date() + start = datetime(day.year, day.month, day.day, h, m) + end = start + timedelta(minutes=duration) + if start > now: + candidates.append(start) + if end > now: + candidates.append(end) + + if not candidates: + return None + return min(candidates) diff --git a/tests/api/test_anchor_due_cache_refresh.py b/tests/api/test_anchor_due_cache_refresh.py new file mode 100644 index 00000000..92b15f1b --- /dev/null +++ b/tests/api/test_anchor_due_cache_refresh.py @@ -0,0 +1,80 @@ +"""Tests for api.routes.anchors._refresh_anchor_due_cache — the anchor-CRUD +side of the Redis next-due gating scheme (shared/notify_due.py). + +Pure unit tests: get_anchors and shared.notify_due are monkeypatched so no +real Postgres/Redis connection is needed. +""" +from __future__ import annotations + +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import api.routes.anchors as anchors_mod + +pytestmark = pytest.mark.timeout(30) + + +async def test_refresh_caches_current_anchor_list(monkeypatch): + anchors = [{"id": "a", "time": "09:00", "duration_minutes": 30}] + monkeypatch.setattr(anchors_mod, "get_anchors", AsyncMock(return_value=anchors)) + set_cached = AsyncMock() + monkeypatch.setattr(anchors_mod.notify_due, "set_cached_anchors", set_cached) + monkeypatch.setattr( + anchors_mod.notify_due, "next_anchor_boundary", MagicMock(return_value=None) + ) + + await anchors_mod._refresh_anchor_due_cache(conn=object(), user_id="u1") + + set_cached.assert_awaited_once_with("u1", anchors) + + +async def test_refresh_writes_anchor_component_when_boundary_exists(monkeypatch): + anchors = [{"id": "a", "time": "09:00", "duration_minutes": 30}] + boundary = datetime(2026, 6, 15, 9, 0) + monkeypatch.setattr(anchors_mod, "get_anchors", AsyncMock(return_value=anchors)) + monkeypatch.setattr(anchors_mod.notify_due, "set_cached_anchors", AsyncMock()) + monkeypatch.setattr( + anchors_mod.notify_due, "next_anchor_boundary", MagicMock(return_value=boundary) + ) + set_component_due = AsyncMock() + monkeypatch.setattr(anchors_mod.notify_due, "set_component_due", set_component_due) + + await anchors_mod._refresh_anchor_due_cache(conn=object(), user_id="u1") + + set_component_due.assert_awaited_once_with("u1", "anchor", boundary.timestamp()) + + +async def test_refresh_skips_component_write_when_no_anchors(monkeypatch): + """No anchors configured at all — nothing to write for the anchor + component (it simply stays absent from the combined score).""" + monkeypatch.setattr(anchors_mod, "get_anchors", AsyncMock(return_value=[])) + monkeypatch.setattr(anchors_mod.notify_due, "set_cached_anchors", AsyncMock()) + monkeypatch.setattr( + anchors_mod.notify_due, "next_anchor_boundary", MagicMock(return_value=None) + ) + set_component_due = AsyncMock() + monkeypatch.setattr(anchors_mod.notify_due, "set_component_due", set_component_due) + + await anchors_mod._refresh_anchor_due_cache(conn=object(), user_id="u1") + + set_component_due.assert_not_called() + + +async def test_refresh_swallows_get_anchors_failure_without_raising(monkeypatch): + """CRITICAL contract: this helper runs AFTER the real anchor mutation has + already committed (see call sites in create_anchor/update_anchor/ + delete_anchor_route). A failure re-reading anchors for the cache must + never propagate and turn an already-successful mutation into a 500, and + must never prevent the caller's subsequent `anchors_updated` WS + broadcast from running.""" + monkeypatch.setattr( + anchors_mod, "get_anchors", AsyncMock(side_effect=RuntimeError("db blip")) + ) + set_cached = AsyncMock() + monkeypatch.setattr(anchors_mod.notify_due, "set_cached_anchors", set_cached) + + await anchors_mod._refresh_anchor_due_cache(conn=object(), user_id="u1") # must not raise + + set_cached.assert_not_called() diff --git a/tests/api/test_internal_notification_gating.py b/tests/api/test_internal_notification_gating.py new file mode 100644 index 00000000..17145b08 --- /dev/null +++ b/tests/api/test_internal_notification_gating.py @@ -0,0 +1,157 @@ +"""Tests for the Redis next-due gating wired into api/routes/internal.py's +_run_notification_check — the core Neon-idle-spin-down fix. + +No real Postgres/Redis needed: pool-touching functions and shared.notify_due +are monkeypatched to assert call/no-call behaviour. +""" +from __future__ import annotations + +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import api.routes.internal as internal_mod + +pytestmark = pytest.mark.timeout(30) + + +# --------------------------------------------------------------------------- +# _run_notification_check — outer gate +# --------------------------------------------------------------------------- + +async def test_skips_all_postgres_when_nothing_due(monkeypatch): + """The whole point: when the Redis gate confirms nothing is due, neither + linked-user-fetch path may touch Postgres at all.""" + monkeypatch.setattr( + internal_mod.notify_due, "get_due_user_ids", AsyncMock(return_value=[]) + ) + get_all = AsyncMock() + get_by_id = AsyncMock() + monkeypatch.setattr(internal_mod, "_get_all_linked_users", get_all) + monkeypatch.setattr(internal_mod, "_get_linked_users_by_id", get_by_id) + + await internal_mod._run_notification_check(pool=object(), ws_manager=None) + + get_all.assert_not_called() + get_by_id.assert_not_called() + + +async def test_falls_back_to_full_scan_when_gating_unavailable(monkeypatch): + """None (Redis unavailable) must fail OPEN — fall back to the original + unfiltered behaviour rather than silently skipping everyone.""" + monkeypatch.setattr( + internal_mod.notify_due, "get_due_user_ids", AsyncMock(return_value=None) + ) + get_all = AsyncMock(return_value=[]) + get_by_id = AsyncMock() + monkeypatch.setattr(internal_mod, "_get_all_linked_users", get_all) + monkeypatch.setattr(internal_mod, "_get_linked_users_by_id", get_by_id) + + await internal_mod._run_notification_check(pool=object(), ws_manager=None) + + get_all.assert_awaited_once() + get_by_id.assert_not_called() + + +async def test_scopes_postgres_fetch_to_due_users_only(monkeypatch): + """When specific users are due, only THOSE users are fetched — never the + full linked-users table scan.""" + monkeypatch.setattr( + internal_mod.notify_due, "get_due_user_ids", AsyncMock(return_value=["u1", "u2"]) + ) + get_all = AsyncMock() + get_by_id = AsyncMock(return_value=[]) + monkeypatch.setattr(internal_mod, "_get_all_linked_users", get_all) + monkeypatch.setattr(internal_mod, "_get_linked_users_by_id", get_by_id) + + await internal_mod._run_notification_check(pool=object(), ws_manager=None) + + get_all.assert_not_called() + get_by_id.assert_awaited_once() + assert get_by_id.await_args.args[1] == ["u1", "u2"] + + +async def test_recomputes_due_cache_after_processing_each_user(monkeypatch): + """After the real checks run for a due user, the anchor/followup next-due + estimates must be written back to Redis (recompute-after-run pattern).""" + user = {"id": "u1", "telegram_chat_id": "123"} + anchor_next = datetime(2026, 6, 15, 10, 0) + followup_next = datetime(2026, 6, 15, 9, 30) + + monkeypatch.setattr( + internal_mod.notify_due, "get_due_user_ids", AsyncMock(return_value=["u1"]) + ) + monkeypatch.setattr(internal_mod, "_get_linked_users_by_id", AsyncMock(return_value=[user])) + monkeypatch.setattr( + internal_mod, "_check_anchor_transitions", AsyncMock(return_value=anchor_next) + ) + monkeypatch.setattr(internal_mod, "check_followups", AsyncMock(return_value=followup_next)) + + set_component_due = AsyncMock() + monkeypatch.setattr(internal_mod.notify_due, "set_component_due", set_component_due) + + await internal_mod._run_notification_check(pool=object(), ws_manager=None) + + calls = {c.args[1]: c.args[2] for c in set_component_due.await_args_list} + assert calls["anchor"] == anchor_next.timestamp() + assert calls["followup"] == followup_next.timestamp() + + +async def test_no_recompute_write_when_neither_component_present(monkeypatch): + """If a user has neither anchors nor active followups, nothing is written + — an absent component simply doesn't contribute to the combined score.""" + user = {"id": "u1", "telegram_chat_id": "123"} + monkeypatch.setattr( + internal_mod.notify_due, "get_due_user_ids", AsyncMock(return_value=["u1"]) + ) + monkeypatch.setattr(internal_mod, "_get_linked_users_by_id", AsyncMock(return_value=[user])) + monkeypatch.setattr(internal_mod, "_check_anchor_transitions", AsyncMock(return_value=None)) + monkeypatch.setattr(internal_mod, "check_followups", AsyncMock(return_value=None)) + + set_component_due = AsyncMock() + monkeypatch.setattr(internal_mod.notify_due, "set_component_due", set_component_due) + + await internal_mod._run_notification_check(pool=object(), ws_manager=None) + + set_component_due.assert_not_called() + + +# --------------------------------------------------------------------------- +# _check_anchor_transitions — returns anchor component + caches schedule +# --------------------------------------------------------------------------- + +async def test_check_anchor_transitions_caches_anchors_and_returns_boundary(monkeypatch): + anchors = [{"id": "a", "time": "09:00", "duration_minutes": 30}] + plan: dict = {} + sentinel_boundary = datetime(2026, 6, 15, 9, 0) + + monkeypatch.setattr( + "bot.message_handler._get_anchors_and_plan", + AsyncMock(return_value=(anchors, plan)), + ) + monkeypatch.setattr("bot.handler_utils.is_anchor_active", lambda a, now=None: False) + + set_cached = AsyncMock() + monkeypatch.setattr(internal_mod.notify_due, "set_cached_anchors", set_cached) + monkeypatch.setattr( + internal_mod.notify_due, + "next_anchor_boundary", + MagicMock(return_value=sentinel_boundary), + ) + + result = await internal_mod._check_anchor_transitions(object(), "u1", AsyncMock()) + + assert result is sentinel_boundary + set_cached.assert_awaited_once_with("u1", anchors) + + +async def test_check_anchor_transitions_returns_none_on_exception(monkeypatch): + monkeypatch.setattr( + "bot.message_handler._get_anchors_and_plan", + AsyncMock(side_effect=RuntimeError("boom")), + ) + + result = await internal_mod._check_anchor_transitions(object(), "u1", AsyncMock()) + + assert result is None diff --git a/tests/bot/test_followup_gating.py b/tests/bot/test_followup_gating.py new file mode 100644 index 00000000..67ee45fa --- /dev/null +++ b/tests/bot/test_followup_gating.py @@ -0,0 +1,120 @@ +"""Tests for the pure followup-due classification used by the Redis next-due +gating scheme (shared/notify_due.py). No database required — classify_followup_row +depends only on (row, config, now). +""" +from __future__ import annotations + +from datetime import datetime, timedelta + +CONFIG = { + "pre_ack_interval_min": 10, + "pre_ack_max_pings": 3, + "post_ack_interval_min": 15, + "post_ack_pings": 2, +} + +NOW = datetime(2026, 6, 15, 12, 0, 0) + + +def _row(**overrides) -> dict: + base = { + "acknowledged_at": None, + "sequence_started_at": (NOW - timedelta(minutes=30)).isoformat(), + "last_ping_at": None, + "pre_ack_pings_sent": 0, + "post_ack_pings_sent": 0, + } + base.update(overrides) + return base + + +def test_pre_ack_not_yet_due_returns_candidate_at_ref_plus_interval(): + from bot.message_handler import classify_followup_row + + row = _row(sequence_started_at=(NOW - timedelta(minutes=2)).isoformat()) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue is None + assert candidate == NOW - timedelta(minutes=2) + timedelta(minutes=10) + + +def test_pre_ack_due_returns_pre_queue_and_next_candidate_one_interval_out(): + from bot.message_handler import classify_followup_row + + row = _row(sequence_started_at=(NOW - timedelta(minutes=15)).isoformat()) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue == "pre" + assert candidate == NOW + timedelta(minutes=10) + + +def test_pre_ack_exhausted_pings_returns_no_candidate(): + """Once pre_ack_max_pings is hit, this row contributes no future due time + (it needs a real acknowledgement to progress, not a timer).""" + from bot.message_handler import classify_followup_row + + row = _row( + sequence_started_at=(NOW - timedelta(minutes=100)).isoformat(), + pre_ack_pings_sent=3, + ) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue is None + assert candidate is None + + +def test_post_ack_not_yet_due_returns_candidate_at_ref_plus_interval(): + from bot.message_handler import classify_followup_row + + row = _row( + acknowledged_at=(NOW - timedelta(minutes=5)).isoformat(), + ) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue is None + assert candidate == NOW - timedelta(minutes=5) + timedelta(minutes=15) + + +def test_post_ack_due_returns_post_queue_and_next_candidate(): + from bot.message_handler import classify_followup_row + + row = _row(acknowledged_at=(NOW - timedelta(minutes=20)).isoformat()) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue == "post" + assert candidate == NOW + timedelta(minutes=15) + + +def test_post_ack_exhausted_pings_returns_no_candidate(): + from bot.message_handler import classify_followup_row + + row = _row( + acknowledged_at=(NOW - timedelta(minutes=100)).isoformat(), + post_ack_pings_sent=2, + ) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue is None + assert candidate is None + + +def test_last_ping_at_takes_precedence_over_sequence_started_at(): + from bot.message_handler import classify_followup_row + + row = _row( + sequence_started_at=(NOW - timedelta(minutes=100)).isoformat(), + last_ping_at=(NOW - timedelta(minutes=1)).isoformat(), + ) + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue is None + assert candidate == NOW - timedelta(minutes=1) + timedelta(minutes=10) + + +def test_unparsable_timestamp_treated_as_never_pinged_and_due_immediately(): + from bot.message_handler import classify_followup_row + + row = _row(sequence_started_at="not-a-timestamp") + queue, candidate = classify_followup_row(row, CONFIG, NOW) + + assert queue == "pre" # minutes_since -> inf -> immediately due diff --git a/tests/shared/__init__.py b/tests/shared/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/shared/test_notify_due.py b/tests/shared/test_notify_due.py new file mode 100644 index 00000000..1c36bb86 --- /dev/null +++ b/tests/shared/test_notify_due.py @@ -0,0 +1,292 @@ +"""Tests for shared.notify_due — Redis next-due gating for notification checks. + +Covers: +- set_component_due / get_due_user_ids: combined ZSET scoring across components +- is_due: single-user cheap check +- Fail-open behaviour when Redis is unavailable (no server/redis_url/redis_client) +- next_anchor_boundary: pure function computing the next anchor start time +""" +from __future__ import annotations + +import time + +import fakeredis +import pytest + +pytestmark = pytest.mark.timeout(30) + + +# --------------------------------------------------------------------------- +# set_component_due / get_due_user_ids +# --------------------------------------------------------------------------- + +async def test_get_due_user_ids_returns_user_whose_due_time_has_passed(): + from shared.notify_due import get_due_user_ids, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + + await set_component_due("user-a", "anchor", now - 10, server=server) + + due = await get_due_user_ids(now, server=server) + assert due == ["user-a"] + + +async def test_get_due_user_ids_excludes_users_not_yet_due(): + from shared.notify_due import get_due_user_ids, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + + await set_component_due("user-future", "anchor", now + 3600, server=server) + + due = await get_due_user_ids(now, server=server) + assert due == [] + + +async def test_get_due_user_ids_returns_empty_list_when_queue_empty(): + from shared.notify_due import get_due_user_ids + + server = fakeredis.FakeServer() + due = await get_due_user_ids(time.time(), server=server) + assert due == [] + + +async def test_set_component_due_combines_min_across_components(): + """The combined due_queue score must be the MIN of all known components — + a user is due as soon as ANY one component (anchor/followup/meeting) is due, + not only when all of them are.""" + from shared.notify_due import get_due_user_ids, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + + await set_component_due("user-b", "anchor", now + 100, server=server) + await set_component_due("user-b", "followup", now + 10, server=server) + + # Not due yet (both components still in the future) + assert await get_due_user_ids(now, server=server) == [] + + # Due once we pass the earlier (followup) component, even though the + # anchor component is still far in the future. + assert await get_due_user_ids(now + 15, server=server) == ["user-b"] + + +async def test_set_component_due_updates_existing_component_independently(): + """Updating one component must not clobber a different component's value — + the combined score must reflect the min of the freshest values for BOTH.""" + from shared.notify_due import get_due_user_ids, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + + await set_component_due("user-c", "anchor", now + 50, server=server) + await set_component_due("user-c", "followup", now + 200, server=server) + # Push the followup component further out — anchor (50) should still govern. + await set_component_due("user-c", "followup", now + 300, server=server) + + assert await get_due_user_ids(now + 60, server=server) == ["user-c"] # anchor passed + assert await get_due_user_ids(now + 10, server=server) == [] # neither passed yet + + +async def test_set_component_due_meeting_component_from_premium_contributes(): + """Premium contributes a 'meeting' component to the SAME due_queue — the + cron's single range query must see meeting-only-due users too.""" + from shared.notify_due import get_due_user_ids, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + + await set_component_due("user-d", "meeting", now - 1, server=server) + + assert await get_due_user_ids(now, server=server) == ["user-d"] + + +# --------------------------------------------------------------------------- +# is_due +# --------------------------------------------------------------------------- + +async def test_is_due_true_when_combined_score_has_passed(): + from shared.notify_due import is_due, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + await set_component_due("user-e", "anchor", now - 5, server=server) + + assert await is_due("user-e", now, server=server) is True + + +async def test_is_due_false_when_combined_score_in_future(): + from shared.notify_due import is_due, set_component_due + + server = fakeredis.FakeServer() + now = time.time() + await set_component_due("user-f", "anchor", now + 3600, server=server) + + assert await is_due("user-f", now, server=server) is False + + +async def test_is_due_fails_open_true_for_unknown_user(): + """A user never seen before (no cache entry — cold start / new link) must + be treated as due so their first real check isn't silently skipped.""" + from shared.notify_due import is_due + + server = fakeredis.FakeServer() + assert await is_due("never-seen-user", time.time(), server=server) is True + + +# --------------------------------------------------------------------------- +# Fail-open when Redis is unavailable entirely +# --------------------------------------------------------------------------- + +async def test_get_due_user_ids_returns_none_when_redis_unavailable(monkeypatch): + """None is the fail-open sentinel — distinct from an empty list, so callers + can tell 'confirmed nothing due' apart from 'gating unavailable, do the + real (unfiltered) check instead.'""" + from shared import notify_due + + monkeypatch.delenv("REDIS_URL", raising=False) + result = await notify_due.get_due_user_ids(time.time()) + assert result is None + + +async def test_is_due_returns_true_when_redis_unavailable(monkeypatch): + from shared import notify_due + + monkeypatch.delenv("REDIS_URL", raising=False) + assert await notify_due.is_due("any-user", time.time()) is True + + +async def test_set_component_due_noop_when_redis_unavailable(monkeypatch): + """Must not raise — gating is best-effort, never fatal to the caller.""" + from shared import notify_due + + monkeypatch.delenv("REDIS_URL", raising=False) + await notify_due.set_component_due("any-user", "anchor", time.time()) # no raise + + +# --------------------------------------------------------------------------- +# Fail-open when Redis IS configured but the call itself raises (timeout, +# connection reset, auth failure) — the case that matters most in production, +# distinct from "REDIS_URL unset" above. +# --------------------------------------------------------------------------- + +class _RaisingRedisClient: + """Stand-in for a configured-but-broken Redis client — every method + raises, simulating a connection drop/timeout mid-call.""" + + async def _boom(self, *args, **kwargs): + raise ConnectionError("simulated Redis connection failure") + + zrangebyscore = _boom + zscore = _boom + hset = _boom + hgetall = _boom + expire = _boom + zadd = _boom + get = _boom + set = _boom + + +async def test_get_due_user_ids_fails_open_when_redis_call_raises(): + from shared.notify_due import get_due_user_ids + + result = await get_due_user_ids(time.time(), redis_client=_RaisingRedisClient()) + assert result is None + + +async def test_is_due_fails_open_when_redis_call_raises(): + from shared.notify_due import is_due + + result = await is_due("some-user", time.time(), redis_client=_RaisingRedisClient()) + assert result is True + + +async def test_set_component_due_does_not_raise_when_redis_call_raises(): + from shared.notify_due import set_component_due + + # Must not propagate — best-effort cache write, caller never sees this. + await set_component_due("some-user", "anchor", time.time(), redis_client=_RaisingRedisClient()) + + +async def test_get_cached_anchors_fails_open_when_redis_call_raises(): + from shared.notify_due import get_cached_anchors + + result = await get_cached_anchors("some-user", redis_client=_RaisingRedisClient()) + assert result is None + + +async def test_set_cached_anchors_does_not_raise_when_redis_call_raises(): + from shared.notify_due import set_cached_anchors + + await set_cached_anchors("some-user", [{"id": "a"}], redis_client=_RaisingRedisClient()) + + +# --------------------------------------------------------------------------- +# next_anchor_boundary — pure function, no Redis/Postgres involved +# --------------------------------------------------------------------------- + +from datetime import datetime # noqa: E402 + + +def _dt(h, m, *, day=15): + return datetime(2026, 6, day, h, m) + + +async def test_next_anchor_boundary_returns_next_start_today(): + from shared.notify_due import next_anchor_boundary + + anchors = [ + {"id": "grind_am", "time": "09:00", "duration_minutes": 60}, + {"id": "grind_pm", "time": "14:00", "duration_minutes": 60}, + ] + now = _dt(8, 0) + boundary = next_anchor_boundary(anchors, now) + assert boundary == _dt(9, 0) + + +async def test_next_anchor_boundary_returns_active_anchor_end_when_inside_window(): + """While inside an anchor's active window, the next boundary is that + anchor's END (when is_anchor_active flips false), not its start again.""" + from shared.notify_due import next_anchor_boundary + + anchors = [{"id": "grind_am", "time": "09:00", "duration_minutes": 60}] + now = _dt(9, 30) + boundary = next_anchor_boundary(anchors, now) + assert boundary == _dt(10, 0) + + +async def test_next_anchor_boundary_wraps_to_tomorrow_after_last_anchor(): + from shared.notify_due import next_anchor_boundary + + anchors = [{"id": "grind_am", "time": "09:00", "duration_minutes": 60}] + now = _dt(23, 0) + boundary = next_anchor_boundary(anchors, now) + assert boundary == _dt(9, 0, day=16) + + +async def test_next_anchor_boundary_returns_none_when_no_anchors(): + from shared.notify_due import next_anchor_boundary + + assert next_anchor_boundary([], _dt(9, 0)) is None + + +async def test_next_anchor_boundary_skips_malformed_row_without_crashing(): + """A bad row (missing/garbage 'time') must not take down the computation + for every other anchor — it's skipped, logged, and the rest still work.""" + from shared.notify_due import next_anchor_boundary + + anchors = [ + {"id": "bad", "time": "not-a-time", "duration_minutes": 30}, + {"id": "missing-time", "duration_minutes": 30}, + {"id": "good", "time": "09:00", "duration_minutes": 30}, + ] + boundary = next_anchor_boundary(anchors, _dt(8, 0)) + assert boundary == _dt(9, 0) + + +async def test_next_anchor_boundary_returns_none_when_all_rows_malformed(): + from shared.notify_due import next_anchor_boundary + + anchors = [{"id": "bad", "time": "garbage"}] + assert next_anchor_boundary(anchors, _dt(8, 0)) is None