SourceBox-LLC
diff --git a/‎backend/app/api/notifications.py‎
Lines changed: 127 additions & 4 deletions b/‎backend/app/api/notifications.py‎
Lines changed: 127 additions & 4 deletions
diff --git a/‎backend/app/main.py‎
Lines changed: 169 additions & 0 deletions b/‎backend/app/main.py‎
Lines changed: 169 additions & 0 deletions
@@ -46,6 +46,11 @@
 
 _NOTIFICATION_KIND_TO_SETTING: dict[str, tuple[str, bool]] = {
     "motion": ("motion_notifications", True),
+    # motion_digest shares motion's inbox toggle — if you've muted
+    # motion in the bell-icon panel, you don't want a digest banner
+    # either.  Generated by _motion_digest_loop in app/main.py when
+    # an email cooldown window expires with extra events accumulated.
+    "motion_digest": ("motion_notifications", True),
     "camera_online": ("camera_transition_notifications", True),
     "camera_offline": ("camera_transition_notifications", True),
     "node_online": ("node_transition_notifications", True),
@@ -84,9 +89,11 @@
 #     "offline" events are — recovery is good news that doesn't need
 #     a midnight email.  Keeps the inbox feed and the email feed
 #     intentionally different shapes.
-#   - Motion is omitted entirely from this map for v1.  Adding it
-#     before the digest/cooldown work in v1.1 is the difference
-#     between "useful" and "unsubscribed within an hour."
+#   - Motion is in the map but defaults FALSE — see the motion +
+#     motion_digest entries below for the full rationale.  v1.1
+#     ships motion email with per-camera cooldown + digest summary
+#     so volume is bounded; we still default-off because per-org
+#     volume variance is too high to opt users in by default.
 #   - All other defaults are True so an org that hasn't visited the
 #     settings page still gets the operator-critical alerts.
 #
@@ -124,6 +131,21 @@
     "member_added":        ("email_member_audit", True),
     "member_role_changed": ("email_member_audit", True),
     "member_removed":      ("email_member_audit", True),
+    # Motion detection — high volume, deferred to v1.1 cooldown design.
+    # Both kinds (immediate first-event "motion" + cooldown summary
+    # "motion_digest") share email_motion.  DEFAULT FALSE on purpose:
+    # motion volume is wildly variable per customer (1 indoor doorbell
+    # vs. 10 outdoor cameras), and opting users in by default risks
+    # surprise day-one volume on outdoor cams → spam-marks → tanks
+    # Resend sender reputation for ALL email kinds across ALL customers.
+    # Customers explicitly opt in via the Settings UI toggle.
+    # Volume cap mechanism: app/api/notifications.py::_claim_motion_cooldown_or_silence
+    # silences subsequent events within the cooldown window per camera;
+    # app/main.py::_motion_digest_loop emits ONE summary at window
+    # close.  Worst case: 2 emails per cycle per camera regardless of
+    # event volume.
+    "motion":        ("email_motion", False),
+    "motion_digest": ("email_motion", False),
 }
 # Note: ``disk_critical`` is intentionally NOT in this map.  Disk-full
 # is platform infrastructure state (our Fly volume) — irrelevant to
@@ -181,6 +203,95 @@ def email_enabled_for_kind(db: Session, org_id: str, kind: str) -> bool:
     return val == "true"
 
 
+# ── Motion email cooldown ──────────────────────────────────────────
+# Per-camera cooldown anchor mechanism.  Used by the email branch of
+# create_notification (when kind="motion") to silence per-event emails
+# that arrive during an active window, AND by ``_motion_digest_loop``
+# in app/main.py to find expired anchors and emit the summary.
+#
+# Anchor data: a Setting row per (org_id, camera_id) where the key is
+# ``motion_email_cooldown_start:{camera_id}`` and the value is the ISO
+# timestamp of when the immediate email was sent.  The colon-suffix
+# pattern matches the ``cloudnode_disk_low_emit_at:{node_id}``
+# precedent in app/api/nodes.py:72-121.
+#
+# Lifecycle: written by _claim_motion_cooldown_or_silence on the first
+# motion event after expiry, read by both subsequent in-window events
+# (which silence) and the digest loop (which emits + deletes), deleted
+# by the digest loop after the window closes.  Persistent across
+# process restarts which is the whole reason for using Setting over an
+# in-memory dict — a deploy mid-window must NOT re-spam users.
+
+
+def _motion_cooldown_anchor_key(camera_id: str) -> str:
+    """Setting key for the per-camera motion email cooldown anchor.
+
+    Mirrors the ``cloudnode_disk_low_emit_at:{node_id}`` pattern in
+    app/api/nodes.py — colon-suffix so the digest loop can find all
+    active anchors with a single ``Setting.key.like("motion_email_cooldown_start:%")``
+    query, and parse the camera_id back out by splitting on ":".
+    """
+    return f"motion_email_cooldown_start:{camera_id}"
+
+
+def _motion_cooldown_minutes(db: Session, org_id: str) -> int:
+    """Per-org cooldown duration in minutes.  Default 15.
+
+    Hidden setting for v1.1 — operators set it via direct DB write or
+    a future env-bootstrap.  Min 1 (no point in zero — the immediate
+    fires anyway).  Falls back to 15 on any parse failure so a
+    corrupt value can't disable emails entirely.
+    """
+    raw = Setting.get(db, org_id, "email_motion_cooldown_minutes", "15")
+    try:
+        return max(1, int(raw))
+    except (ValueError, TypeError):
+        return 15
+
+
+def _claim_motion_cooldown_or_silence(
+    db: Session, org_id: str, camera_id: Optional[str]
+) -> bool:
+    """Returns True if this event should email immediately (and writes
+    the anchor as a side effect); False if a cooldown anchor is active
+    and the event should be silenced.
+
+    Concurrency note: read-then-write is non-atomic in SQLite without
+    an explicit transaction.  Two motion events arriving in the same
+    microsecond could both see "anchor missing" and both enqueue an
+    immediate email.  Worst case is one duplicate first-event email,
+    which we accept (matches the cloudnode_disk_low precedent and is
+    far better than silently losing alerts).  If/when we go multi-
+    replica, swap to a dedicated table with INSERT...ON CONFLICT.
+    """
+    if camera_id is None:
+        # Defensive — motion notifications today always carry a camera_id
+        # (see ws.py emit), but a future caller without one shouldn't
+        # silently break.  Default to "send immediate" rather than
+        # "silence" because a missing camera_id would also break the
+        # digest loop's per-camera grouping.
+        return True
+
+    key = _motion_cooldown_anchor_key(camera_id)
+    existing = Setting.get(db, org_id, key, "")
+    cooldown_min = _motion_cooldown_minutes(db, org_id)
+    now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+
+    if existing:
+        try:
+            anchor = datetime.fromisoformat(existing)
+            if (now - anchor).total_seconds() < cooldown_min * 60:
+                return False  # silenced — within active cooldown window
+        except ValueError:
+            # Malformed value (manual edit / corruption).  Treat as
+            # expired and overwrite below — better than refusing to
+            # email forever.
+            pass
+
+    Setting.set(db, org_id, key, now.isoformat())
+    return True
+
+
 def _build_email_content(notif: Notification) -> tuple[str, str, str]:
     """Render a notification into ``(subject, body_text, body_html)``
     via the Jinja2 templates in ``app/templates/emails/``.
@@ -436,7 +547,18 @@ def create_notification(
         # bell panel too.
         try:
             if email_enabled_for_kind(session, org_id, kind):
-                _enqueue_email_for_notification(session, notif, audience)
+                if kind == "motion":
+                    # v1.1 cooldown gate: only email the FIRST motion
+                    # event per camera per cooldown window.  Subsequent
+                    # in-window events still hit the inbox + SSE above
+                    # but skip the email outbox.  ``_motion_digest_loop``
+                    # in app/main.py picks up expired anchors and
+                    # emits a single ``motion_digest`` summary.
+                    if _claim_motion_cooldown_or_silence(session, org_id, camera_id):
+                        _enqueue_email_for_notification(session, notif, audience)
+                    # else: silenced — digest will summarise
+                else:
+                    _enqueue_email_for_notification(session, notif, audience)
         except Exception:
             logger.exception(
                 "[Notifications] email enqueue side-channel failed for kind=%s",
@@ -796,6 +918,7 @@ class EmailPreferences(BaseModel):
     email_mcp_key_audit: Optional[bool] = None
     email_cloudnode_disk_low: Optional[bool] = None
     email_member_audit: Optional[bool] = None
+    email_motion: Optional[bool] = None
 
 
 def _current_email_prefs(db: Session, org_id: str) -> dict:
 
@@ -123,6 +123,16 @@
     os.getenv("DISK_CRITICAL_REEMIT_INTERVAL_SECONDS", str(6 * 3600))
 )
 
+# Cadence for the motion-email digest sweep.  60s is fast enough that
+# a 15-minute cooldown's digest is delivered within ~1 minute of window
+# expiry (acceptable lag for "here's a summary of the last 15 min"),
+# but slow enough that the LIKE scan over Setting rows is a rounding
+# error.  See app/api/notifications.py::_claim_motion_cooldown_or_silence
+# for the per-camera anchor mechanism this loop drains.
+MOTION_DIGEST_INTERVAL_SECONDS = int(
+    os.getenv("MOTION_DIGEST_INTERVAL_SECONDS", "60")
+)
+
 
 @asynccontextmanager
 async def lifespan(app):
@@ -146,6 +156,13 @@ async def lifespan(app):
     # _check_and_emit_disk_critical for the rationale.  Pure in-
     # memory debounce, no DB persistence.
     disk_check_task = asyncio.create_task(_disk_check_loop())
+    # Motion-email digest sweep — drains expired per-camera cooldown
+    # anchors written by the email-immediate path in
+    # app/api/notifications.py::_claim_motion_cooldown_or_silence and
+    # emits a single ``motion_digest`` notification per camera that
+    # accumulated additional motion events during the cooldown window.
+    # See _motion_digest_loop below for the full lifecycle.
+    motion_digest_task = asyncio.create_task(_motion_digest_loop())
     print(
         f"[App] SourceBox Sentry Command Center started "
         f"(log retention: {LOG_RETENTION_DAYS}d, "
@@ -159,6 +176,7 @@ async def lifespan(app):
     release_refresh_task.cancel()
     email_worker_task.cancel()
     disk_check_task.cancel()
+    motion_digest_task.cancel()
     print("[System] Shutdown complete")
 
 
@@ -810,6 +828,157 @@ async def _disk_check_loop():
             logger.exception("[DiskCheck] tick failed")
 
 
+async def _motion_digest_loop():
+    """Per-camera motion-email digest sweep.
+
+    Runs every ``MOTION_DIGEST_INTERVAL_SECONDS`` (60s default).  Drains
+    cooldown anchors written by the immediate-email path (see
+    ``app/api/notifications.py::_claim_motion_cooldown_or_silence``).
+    For each anchor whose window has expired, counts MotionEvent rows
+    that landed in the window and — if there were extras AND the org
+    still has email_motion enabled — emits a single ``motion_digest``
+    notification (which itself enqueues an email via the standard
+    create_notification path).  The anchor is deleted regardless of
+    whether a digest was emitted, so the next motion event for that
+    camera triggers a fresh immediate email + new anchor.
+
+    Per-anchor try/except so one corrupt row can't poison the tick;
+    the outer try/except catches anything that escapes (DB connection
+    drops, session reset) so the loop survives transient failures and
+    picks up where it left off on the next tick.
+    """
+    from app.models.models import Camera, MotionEvent, Setting
+    from app.api.notifications import (
+        create_notification,
+        email_enabled_for_kind,
+        _motion_cooldown_minutes,
+    )
+
+    while True:
+        await asyncio.sleep(MOTION_DIGEST_INTERVAL_SECONDS)
+
+        try:
+            db = SessionLocal()
+            try:
+                now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+                anchors = (
+                    db.query(Setting)
+                    .filter(Setting.key.like("motion_email_cooldown_start:%"))
+                    .all()
+                )
+                for anchor_row in anchors:
+                    try:
+                        # Parse camera_id back from the colon-suffixed key.
+                        # ``split(":", 1)`` handles the (extremely unlikely)
+                        # case of a colon in the camera_id itself.
+                        if ":" not in anchor_row.key:
+                            db.delete(anchor_row)
+                            db.commit()
+                            continue
+                        camera_id = anchor_row.key.split(":", 1)[1]
+
+                        if not anchor_row.value:
+                            db.delete(anchor_row)
+                            db.commit()
+                            continue
+                        try:
+                            anchor_ts = datetime.fromisoformat(anchor_row.value)
+                        except ValueError:
+                            # Corrupt timestamp — drop the row so the next
+                            # motion event starts fresh rather than being
+                            # silenced forever by an unparseable anchor.
+                            db.delete(anchor_row)
+                            db.commit()
+                            continue
+
+                        cooldown_min = _motion_cooldown_minutes(db, anchor_row.org_id)
+                        if (now - anchor_ts).total_seconds() < cooldown_min * 60:
+                            # Window still open — leave the anchor alone.
+                            continue
+
+                        window_end = anchor_ts + timedelta(minutes=cooldown_min)
+                        # Count events strictly AFTER the anchor (the immediate
+                        # email already covered the anchor-time event itself)
+                        # and up to the window edge.  Events past window_end
+                        # belong to a later cycle — but no later cycle exists
+                        # yet because the anchor is still here, so this filter
+                        # is defensive against clock skew + late-arriving events.
+                        extra_count = (
+                            db.query(MotionEvent)
+                            .filter(
+                                MotionEvent.org_id == anchor_row.org_id,
+                                MotionEvent.camera_id == camera_id,
+                                MotionEvent.timestamp > anchor_ts,
+                                MotionEvent.timestamp <= window_end,
+                            )
+                            .count()
+                        )
+
+                        if extra_count > 0 and email_enabled_for_kind(
+                            db, anchor_row.org_id, "motion"
+                        ):
+                            # Re-resolve display name at digest-emit time.
+                            # The camera might have been renamed since the
+                            # immediate email; show the current name.
+                            cam = (
+                                db.query(Camera)
+                                .filter_by(
+                                    camera_id=camera_id,
+                                    org_id=anchor_row.org_id,
+                                )
+                                .first()
+                            )
+                            display = (
+                                cam.name if cam and cam.name else camera_id
+                            )
+                            create_notification(
+                                org_id=anchor_row.org_id,
+                                kind="motion_digest",
+                                title=(
+                                    f"{extra_count} more motion event"
+                                    f"{'s' if extra_count != 1 else ''} "
+                                    f"on {display}"
+                                ),
+                                body=(
+                                    f"{extra_count} additional motion event"
+                                    f"{'s were' if extra_count != 1 else ' was'} "
+                                    f'detected on "{display}" in the '
+                                    f"{cooldown_min}-minute window after the "
+                                    f"first alert."
+                                ),
+                                severity="info",
+                                audience="all",
+                                link=f"/dashboard?camera={camera_id}",
+                                camera_id=camera_id,
+                                meta={
+                                    "event_count": extra_count,
+                                    "window_start": anchor_ts.isoformat(),
+                                    "window_end": window_end.isoformat(),
+                                    "cooldown_minutes": cooldown_min,
+                                },
+                                db=db,
+                            )
+
+                        # Always delete the anchor — window has closed.  Next
+                        # motion on this camera starts a fresh cycle.
+                        db.delete(anchor_row)
+                        db.commit()
+                    except Exception:
+                        logger.exception(
+                            "[MotionDigest] anchor processing failed key=%s org=%s",
+                            anchor_row.key,
+                            anchor_row.org_id,
+                        )
+                        try:
+                            db.rollback()
+                        except Exception:
+                            pass
+            finally:
+                db.close()
+        except Exception:
+            logger.exception("[MotionDigest] tick failed")
+
+
 async def _offline_sweep_loop():
     """Background task — periodically flip stale 'online' rows to 'offline'.