feat: schedule mobile notification quiet hours (Closes #1392)

src/main.py

@@ -522,6 +522,30 @@ class FollowingNotificationPayload(BaseModel):
     enabled: bool
 
 
+class QuietHoursPayload(BaseModel):
+    enabled: bool
+    start: str = Field(pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$")
+    end: str = Field(pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$")
+    timezone: str = Field(min_length=1, max_length=64)
+
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str) -> str:
+        from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+        try:
+            ZoneInfo(value)
+        except ZoneInfoNotFoundError as error:
+            raise ValueError("Valid IANA timezone required") from error
+        return value
+
+    @model_validator(mode="after")
+    def distinct_boundaries(self):
+        if self.start == self.end:
+            raise ValueError("quiet hours require distinct start and end times")
+        return self
+
+
 StepUpAction = Literal[
     "merge_pull",
     "delete_source_branch",
@@ -2529,6 +2553,9 @@ async def push_status(request: Request):
     following_preferences = await asyncio.to_thread(
         _push_subscription_store.following_preferences, device_id
     )
+    quiet_hours = await asyncio.to_thread(
+        _push_subscription_store.quiet_hours, device_id
+    )
     delivery_health = await asyncio.to_thread(
         _push_subscription_store.delivery_health, device_id
     )
@@ -2545,6 +2572,10 @@ async def push_status(request: Request):
         "start_day_timezone": start_day_preferences["timezone"],
         "start_day_reminder_hour": start_day_preferences["reminder_hour"],
         "following_enabled": following_preferences["enabled"],
+        "quiet_hours_enabled": quiet_hours["enabled"],
+        "quiet_hours_start": quiet_hours["start"],
+        "quiet_hours_end": quiet_hours["end"],
+        "quiet_hours_timezone": quiet_hours["timezone"],
         "delivery_health": delivery_health,
     }
 
@@ -2712,6 +2743,31 @@ async def update_following_notifications(
     return {"following_enabled": payload.enabled}
 
 
+@app.put("/api/v1/push-subscription/quiet-hours")
+async def update_quiet_hours(payload: QuietHoursPayload, request: Request):
+    device_id = await dashboard_auth.session_management_id(
+        request.state.dashboard_session
+    )
+    if payload.enabled and not await asyncio.to_thread(
+        _push_subscription_store.is_subscribed, device_id
+    ):
+        raise HTTPException(status_code=409, detail="Enable device notifications first")
+    await asyncio.to_thread(
+        _push_subscription_store.set_quiet_hours,
+        device_id,
+        enabled=payload.enabled,
+        start=payload.start,
+        end=payload.end,
+        timezone=payload.timezone,
+    )
+    return {
+        "quiet_hours_enabled": payload.enabled,
+        "quiet_hours_start": payload.start,
+        "quiet_hours_end": payload.end,
+        "quiet_hours_timezone": payload.timezone,
+    }
+
+
 @app.patch("/api/v1/push-subscription/deadlines/snooze")
 async def snooze_deadline_reminder(request: Request):
     device_id = await dashboard_auth.session_management_id(

src/push_notifications.py

@@ -110,6 +110,7 @@ async def dispatch_following_changes(
     lease_seconds: float = 60.0,
     send_timeout_seconds: float = 10.0,
     max_concurrency: int = 8,
+    now: float | None = None,
 ) -> int:
     """Notify opted-in devices once for each privacy-safe Following change set."""
     if not configuration.enabled:
@@ -119,13 +120,15 @@ async def dispatch_following_changes(
         store.acquire_dispatch_lease,
         owner,
         channel="following",
-        now=time.time(),
+        now=time.time() if now is None else now,
         lease_seconds=max(15.0, lease_seconds, send_timeout_seconds + 5.0),
     )
     if not acquired:
         return 0
     try:
-        devices = await asyncio.to_thread(store.following_notification_devices)
+        devices = await asyncio.to_thread(
+            store.following_notification_devices, now=now
+        )
         if not devices:
             return 0
         snapshot = await following()
@@ -174,17 +177,25 @@ async def dispatch_following_changes(
                 device for device in pending if statuses.get(device.session_id) == "active"
             ]
         count = min(len(changed), 50)
-        payload = json.dumps({
-            "title": f"{count} watched item{'s' if count != 1 else ''} changed",
-            "body": "Open Following to review the latest activity.",
-            "route": "#/my-work/following",
-            "tag": f"stackchain-following-{fingerprint[:16]}",
-            "following_count": count,
-        }, separators=(",", ":"))
         semaphore = asyncio.Semaphore(max(1, max_concurrency))
 
         async def dispatch_device(device) -> int:
             async with semaphore:
+                payload = json.dumps({
+                    "title": (
+                        f"{count} watched update{'s' if count != 1 else ''} while alerts were paused"
+                        if device.catch_up
+                        else f"{count} watched item{'s' if count != 1 else ''} changed"
+                    ),
+                    "body": "Open Following to review the latest activity.",
+                    "route": "#/my-work/following",
+                    "tag": (
+                        "stackchain-following-catch-up"
+                        if device.catch_up
+                        else f"stackchain-following-{fingerprint[:16]}"
+                    ),
+                    "following_count": count,
+                }, separators=(",", ":"))
                 still_owner = await asyncio.to_thread(
                     store.acquire_dispatch_lease,
                     owner,
@@ -241,6 +252,7 @@ async def dispatch_unread_updates(
     max_concurrency: int = 8,
     max_individual_notifications: int = 3,
     endpoint_validator: Callable[[str], Awaitable[str]] | None = None,
+    now: float | None = None,
 ) -> int:
     if not configuration.enabled:
         return 0
@@ -249,7 +261,7 @@ async def dispatch_unread_updates(
         store.acquire_dispatch_lease,
         owner,
         channel="unread",
-        now=time.time(),
+        now=time.time() if now is None else now,
         lease_seconds=lease_seconds,
     )
     if not acquired:
@@ -265,7 +277,9 @@ async def dispatch_unread_updates(
         }
         unread_count = min(len(thread_revisions), 9999)
         await asyncio.to_thread(store.reconcile_unread, thread_revisions)
-        deliveries = await asyncio.to_thread(store.claim_unseen, thread_revisions)
+        deliveries = await asyncio.to_thread(
+            store.claim_unseen, thread_revisions, now=now
+        )
         if session_statuses is not None:
             try:
                 statuses = await session_statuses(
@@ -306,10 +320,10 @@ async def dispatch_unread_updates(
                     for thread_revision in delivery.thread_revisions
                     if thread_revision not in digest_pending
                 )
-                individual_revisions = new_revisions[
+                individual_revisions = () if delivery.catch_up else new_revisions[
                     :max(0, max_individual_notifications)
                 ]
-                overflow_revisions = (
+                overflow_revisions = delivery.thread_revisions if delivery.catch_up else (
                     delivery.digest_revisions
                     + new_revisions[len(individual_revisions):]
                 )
@@ -385,10 +399,22 @@ async def dispatch_unread_updates(
                         return count
                     payload = json.dumps(
                         {
-                            "title": f"{len(overflow_revisions)} new work updates",
-                            "body": "Tap to review them in Stackchain.",
+                            "title": (
+                                f"{len(overflow_revisions)} updates while alerts were paused"
+                                if delivery.catch_up
+                                else f"{len(overflow_revisions)} new work updates"
+                            ),
+                            "body": (
+                                "Open Updates to catch up in Stackchain."
+                                if delivery.catch_up
+                                else "Tap to review them in Stackchain."
+                            ),
                             "route": "#/my-work/updates",
-                            "tag": "stackchain-update-digest",
+                            "tag": (
+                                "stackchain-update-catch-up"
+                                if delivery.catch_up
+                                else "stackchain-update-digest"
+                            ),
                             "update_count": len(overflow_revisions),
                             "unread_count": unread_count,
                         },

src/push_subscription_store.py

@@ -7,7 +7,9 @@ import sqlite3
 import time
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
+from datetime import datetime, timezone as datetime_timezone
 from pathlib import Path
+from zoneinfo import ZoneInfo
 
 from src.private_state import connect_private_sqlite
 from src.state_encryption import (
@@ -22,6 +24,7 @@ class PushDelivery:
     subscription: dict
     thread_revisions: tuple[tuple[int, str], ...]
     digest_revisions: tuple[tuple[int, str], ...] = ()
+    catch_up: bool = False
 
     @property
     def thread_ids(self) -> tuple[int, ...]:
@@ -57,6 +60,7 @@ class FollowingNotificationDevice:
     session_id: str
     subscription: dict
     delivered_fingerprint: str | None
+    catch_up: bool = False
 
 
 class DisabledPushSubscriptionStore:
@@ -91,10 +95,13 @@ class DisabledPushSubscriptionStore:
     def following_preferences(self, session_id: str) -> dict:
         return {"enabled": False}
 
-    def following_notification_devices(self) -> list[FollowingNotificationDevice]:
+    def quiet_hours(self, session_id: str) -> dict:
+        return {"enabled": False, "start": "22:00", "end": "07:00", "timezone": "UTC"}
+
+    def following_notification_devices(self, *, now: float | None = None) -> list[FollowingNotificationDevice]:
         return []
 
-    def claim_unseen(self, thread_revisions) -> list[PushDelivery]:
+    def claim_unseen(self, thread_revisions, *, now: float | None = None) -> list[PushDelivery]:
         return []
 
     def acquire_dispatch_lease(self, *args, **kwargs) -> bool:
@@ -133,6 +140,9 @@ class DisabledPushSubscriptionStore:
     def set_following_preferences(self, *args, **kwargs) -> None:
         return None
 
+    def set_quiet_hours(self, *args, **kwargs) -> None:
+        return None
+
     def mark_following_delivered(self, *args, **kwargs) -> None:
         return None
 
@@ -183,6 +193,16 @@ def _revisions(
     return tuple(sorted(normalized.items()))
 
 
+def _inside_quiet_hours(*, now: float, start: str, end: str, timezone: str) -> bool:
+    local = datetime.fromtimestamp(now, tz=datetime_timezone.utc).astimezone(ZoneInfo(timezone))
+    minute = local.hour * 60 + local.minute
+    start_minute = int(start[:2]) * 60 + int(start[3:])
+    end_minute = int(end[:2]) * 60 + int(end[3:])
+    if start_minute < end_minute:
+        return start_minute <= minute < end_minute
+    return minute >= start_minute or minute < end_minute
+
+
 class PushSubscriptionStore:
     """Durable, device-bound Web Push subscriptions and delivery deduplication."""
 
@@ -262,6 +282,16 @@ class PushSubscriptionStore:
                     FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                         ON DELETE CASCADE
                 );
+                CREATE TABLE IF NOT EXISTS push_quiet_hours (
+                    session_id TEXT PRIMARY KEY,
+                    enabled INTEGER NOT NULL DEFAULT 0,
+                    start_time TEXT NOT NULL DEFAULT '22:00',
+                    end_time TEXT NOT NULL DEFAULT '07:00',
+                    timezone TEXT NOT NULL DEFAULT 'UTC',
+                    suppressed INTEGER NOT NULL DEFAULT 0,
+                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
+                        ON DELETE CASCADE
+                );
                 """
             )
             delivery_columns = {
@@ -612,20 +642,64 @@ class PushSubscriptionStore:
             ).fetchone()
         return {"enabled": bool(row[0]) if row else False}
 
-    def following_notification_devices(self) -> list[FollowingNotificationDevice]:
+    def set_quiet_hours(
+        self, session_id: str, *, enabled: bool, start: str, end: str, timezone: str
+    ) -> None:
+        with self._connect() as connection:
+            connection.execute(
+                """INSERT INTO push_quiet_hours(
+                       session_id, enabled, start_time, end_time, timezone
+                   ) VALUES (?, ?, ?, ?, ?)
+                   ON CONFLICT(session_id) DO UPDATE SET
+                       enabled = excluded.enabled,
+                       start_time = excluded.start_time,
+                       end_time = excluded.end_time,
+                       timezone = excluded.timezone,
+                       suppressed = CASE WHEN excluded.enabled = 1 THEN suppressed ELSE 0 END""",
+                (session_id, int(enabled), start, end, timezone),
+            )
+
+    def quiet_hours(self, session_id: str) -> dict:
+        with self._connect() as connection:
+            row = connection.execute(
+                """SELECT enabled, start_time, end_time, timezone
+                   FROM push_quiet_hours WHERE session_id = ?""",
+                (session_id,),
+            ).fetchone()
+        return {
+            "enabled": bool(row[0]) if row else False,
+            "start": row[1] if row else "22:00",
+            "end": row[2] if row else "07:00",
+            "timezone": row[3] if row else "UTC",
+        }
+
+    def following_notification_devices(
+        self, *, now: float | None = None
+    ) -> list[FollowingNotificationDevice]:
+        checked_at = time.time() if now is None else now
         with self._connect() as connection:
             rows = connection.execute(
-                """SELECT s.session_id, s.subscription_json, p.delivered_fingerprint
+                """SELECT s.session_id, s.subscription_json, p.delivered_fingerprint,
+                          q.enabled, q.start_time, q.end_time, q.timezone, q.suppressed
                    FROM push_subscriptions s
                    JOIN push_following_preferences p ON p.session_id = s.session_id
+                   LEFT JOIN push_quiet_hours q ON q.session_id = s.session_id
                    WHERE p.enabled = 1 ORDER BY s.session_id"""
             ).fetchall()
-        return [
-            FollowingNotificationDevice(
-                row[0], self._open_subscription(row[0], row[1]), row[2]
-            )
-            for row in rows
-        ]
+            devices = []
+            for row in rows:
+                if row[3] and _inside_quiet_hours(
+                    now=checked_at, start=row[4], end=row[5], timezone=row[6]
+                ):
+                    connection.execute(
+                        "UPDATE push_quiet_hours SET suppressed = 1 WHERE session_id = ?",
+                        (row[0],),
+                    )
+                    continue
+                devices.append(FollowingNotificationDevice(
+                    row[0], self._open_subscription(row[0], row[1]), row[2], bool(row[7])
+                ))
+        return devices
 
     def mark_following_delivered(self, session_id: str, fingerprint: str) -> None:
         with self._connect() as connection:
@@ -634,9 +708,14 @@ class PushSubscriptionStore:
                    WHERE session_id = ? AND enabled = 1""",
                 (fingerprint, session_id),
             )
+            connection.execute(
+                "UPDATE push_quiet_hours SET suppressed = 0 WHERE session_id = ?",
+                (session_id,),
+            )
 
     def claim_unseen(
-        self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]]
+        self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
+        *, now: float | None = None,
     ) -> list[PushDelivery]:
         candidates = _revisions(thread_revisions)
         if not candidates:
@@ -647,6 +726,21 @@ class PushSubscriptionStore:
             ).fetchall()
             deliveries = []
             for session_id, encoded in rows:
+                quiet = connection.execute(
+                    """SELECT enabled, start_time, end_time, timezone, suppressed
+                       FROM push_quiet_hours WHERE session_id = ?""",
+                    (session_id,),
+                ).fetchone()
+                catch_up = bool(quiet and quiet[4])
+                if quiet and quiet[0] and _inside_quiet_hours(
+                    now=time.time() if now is None else now,
+                    start=quiet[1], end=quiet[2], timezone=quiet[3],
+                ):
+                    connection.execute(
+                        "UPDATE push_quiet_hours SET suppressed = 1 WHERE session_id = ?",
+                        (session_id,),
+                    )
+                    continue
                 delivered = {
                     row[0]: row[1]
                     for row in connection.execute(
@@ -700,6 +794,7 @@ class PushSubscriptionStore:
                             self._open_subscription(session_id, encoded),
                             unseen,
                             tuple(digest_revisions),
+                            catch_up,
                         )
                     )
             return deliveries
@@ -770,3 +865,7 @@ class PushSubscriptionStore:
                 "DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?",
                 ((session_id, thread_id) for thread_id, _revision in values),
             )
+            connection.execute(
+                "UPDATE push_quiet_hours SET suppressed = 0 WHERE session_id = ?",
+                (session_id,),
+            )