"""Read/write + send-time resolution for per-person notification preferences. Two responsibilities: * CRUD the ``notification_preferences`` row (get / upsert / serialize). * ``resolve_gating`` — the send-time decision used by ``NotificationService._deliver``: given a recipient + template + the push channels about to be attempted, return which push channels are muted and (if the recipient is inside a quiet-hours window) the UTC instant to park the push until. Both the in-app inbox row and critical templates are out of scope here: * the inbox row is always written (muting only suppresses *push*); * ``is_critical_template`` short-circuits the whole gate in the caller, so an ops-critical alert never mutes and never parks. """ from __future__ import annotations from datetime import UTC, datetime, time, timedelta from typing import Any from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy.ext.asyncio import AsyncSession from app.core.working_hours import resolve_timezone from app.identity.person_models import IdentityPerson from app.notifications.categories import classify, employee_categories from app.notifications.preference_models import NotificationPreference # Channels an employee can govern. The in-app inbox is intentionally absent — # it is the durable record and is never suppressed. GOVERNABLE_CHANNELS: tuple[str, ...] = ("feishu", "email") def _parse_hhmm(value: str | None) -> time | None: if not value: return None try: hh, mm = value.split(":", 1) h, m = int(hh), int(mm) except (ValueError, AttributeError): return None if not (0 <= h <= 23 and 0 <= m <= 59): return None return time(h, m) def _fmt_hhmm(t: time) -> str: return f"{t.hour:02d}:{t.minute:02d}" def quiet_window_park_until( now_utc: datetime, tz: ZoneInfo, start_str: str | None, end_str: str | None, ) -> datetime | None: """If ``now_utc`` falls inside the daily [start, end) local window, return the UTC instant the window next ends (when parked push should fire); otherwise ``None``. Windows where start > end wrap past midnight. A zero-length window (start == end) is treated as no window. """ start = _parse_hhmm(start_str) end = _parse_hhmm(end_str) if start is None or end is None or start == end: return None local = now_utc.astimezone(tz) cur = local.time() crosses_midnight = start > end if crosses_midnight: inside = cur >= start or cur < end else: inside = start <= cur < end if not inside: return None # Resolve the local datetime of the window's END. end_local = local.replace( hour=end.hour, minute=end.minute, second=0, microsecond=0, ) if crosses_midnight and cur >= start: # Evening portion of a midnight-wrapping window → end is tomorrow. end_local = end_local + timedelta(days=1) elif not crosses_midnight and end_local <= local: # Defensive: same-day window whose end already passed this minute. end_local = end_local + timedelta(days=1) return end_local.astimezone(UTC) class NotificationPreferenceService: def __init__(self, db: AsyncSession) -> None: self.db = db async def get(self, person_id: UUID) -> NotificationPreference | None: return await self.db.get(NotificationPreference, person_id) async def upsert( self, *, person_id: UUID, muted_channels: dict[str, list[str]], quiet_hours_enabled: bool, quiet_hours_start: str | None, quiet_hours_end: str | None, ) -> NotificationPreference: """Create or update the row. Input is sanitised: only governable channels and known category keys survive; quiet-hours times are normalised to ``HH:MM`` and dropped (with the flag forced off) if unparseable, so a malformed payload can never wedge the gate. """ valid_cats = {c.key for c in employee_categories() if c.mutable} clean: dict[str, list[str]] = {} for cat_key, channels in (muted_channels or {}).items(): if cat_key not in valid_cats: continue kept = [c for c in (channels or []) if c in GOVERNABLE_CHANNELS] if kept: clean[cat_key] = sorted(set(kept)) start_t = _parse_hhmm(quiet_hours_start) end_t = _parse_hhmm(quiet_hours_end) enabled = bool(quiet_hours_enabled and start_t and end_t and start_t != end_t) start_norm = _fmt_hhmm(start_t) if (enabled and start_t) else None end_norm = _fmt_hhmm(end_t) if (enabled and end_t) else None row = await self.get(person_id) if row is None: row = NotificationPreference(person_id=person_id) self.db.add(row) row.muted_channels = clean row.quiet_hours_enabled = enabled row.quiet_hours_start = start_norm row.quiet_hours_end = end_norm await self.db.flush() return row def serialize(self, row: NotificationPreference | None) -> dict[str, Any]: """Wire shape for the preference centre. Always includes the full category catalog so the frontend can render every toggle even when the person has no stored row yet.""" muted = dict(row.muted_channels) if row and row.muted_channels else {} return { "quiet_hours_enabled": bool(row.quiet_hours_enabled) if row else False, "quiet_hours_start": (row.quiet_hours_start if row else None), "quiet_hours_end": (row.quiet_hours_end if row else None), "muted_channels": muted, "governable_channels": list(GOVERNABLE_CHANNELS), "categories": [ { "key": c.key, "bucket": c.bucket, "mutable": c.mutable, } for c in employee_categories() ], } async def resolve_gating( self, *, recipient: IdentityPerson, template_key: str | None, requested_push_channels: list[str], now_utc: datetime | None = None, ) -> tuple[list[str], datetime | None]: """Return ``(muted_push_channels, park_until_utc)``. * ``muted_push_channels`` — channels the recipient muted for this template's category (subset of ``requested_push_channels``). * ``park_until_utc`` — when set, the recipient is inside a quiet-hours window and non-muted push should be deferred until this instant. Non-mutable categories (security / IT-ops) are never muted or parked. Callers must already have excluded critical templates. """ prefs = await self.get(recipient.id) if prefs is None: return [], None cat = classify(template_key) if not cat.mutable: return [], None muted_for_cat = set((prefs.muted_channels or {}).get(cat.key, [])) muted = [c for c in requested_push_channels if c in muted_for_cat] park_until: datetime | None = None if prefs.quiet_hours_enabled: park_until = quiet_window_park_until( now_utc or datetime.now(UTC), resolve_timezone(recipient), prefs.quiet_hours_start, prefs.quiet_hours_end, ) return muted, park_until __all__ = [ "GOVERNABLE_CHANNELS", "NotificationPreferenceService", "quiet_window_park_until", ]