ops(motion): per-org server-side kill switch for motion-event ingestion

Closes the motion-runaway scenario from yesterday's SaaS-readiness
review: there was no way to stop a misbehaving / mis-configured
sensor from flooding events into MotionEvent without reaching the
node itself.  The per-camera recording policy is the granularity
operators usually want, but if a sensor is producing hundreds of
events per second it's faster to flip a server-side switch than to
SSH to the node and reconfigure FFmpeg's scene-change threshold.

Changes:

  - hls.py POST /motion: short-circuit before recording when
    Setting `motion_ingestion_enabled` is "false" for the camera's
    org.  Returns 200 + ingested:false rather than 4xx so the
    CloudNode treats this as a by-design rejection (same shape as
    the plan-cap suspension path) and doesn't burn its retry
    budget.  Default "true" so orgs that never touch the toggle
    keep the original always-ingest behaviour.

  - cameras.py: new GET + POST /api/settings/motion-ingestion (admin
    for write, viewer for read).  Mirrors the notifications/timezone
    settings pattern: stored in the existing Setting key/value table,
    no schema migration needed.  Audited so there's a record of who
    flipped it.

  - 3 new tests pinning the kill-switch shape: default enabled,
    toggle persists across GET, viewer can't flip the switch.

This is the one of the cheap fixes from the SaaS-readiness review.
Costs ~30 lines of code, gives operators an immediate stop-bleeding
lever for a real failure mode.

290 backend tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Sbussiso

backend/app/api/cameras.py

@@ -436,6 +436,57 @@ async def get_notification_settings(
     }
 
 
+@router.get("/settings/motion-ingestion")
+async def get_motion_ingestion_setting(
+    user: AuthUser = Depends(require_view), db: Session = Depends(get_db)
+):
+    """Return whether server-side motion-event ingestion is enabled
+    for this org.
+
+    Defaults to enabled — the kill switch only takes effect after an
+    admin explicitly flips it off via POST.  This is a safety valve
+    for runaway sensors flooding events; under normal operation it
+    stays on and the per-camera recording policy is the granularity
+    operators usually want.
+    """
+    enabled = Setting.get(db, user.org_id, "motion_ingestion_enabled", "true").lower() == "true"
+    return {"motion_ingestion_enabled": enabled}
+
+
+@router.post("/settings/motion-ingestion")
+@limiter.limit("30/minute")
+async def update_motion_ingestion_setting(
+    request: Request,
+    user: AuthUser = Depends(require_admin),
+    db: Session = Depends(get_db),
+):
+    """Toggle server-side motion-event ingestion for the org.
+
+    Body: ``{"enabled": true|false}``.
+
+    When set to false, ``POST /api/cameras/{id}/motion`` short-circuits
+    with ``{"ingested": false}`` and no MotionEvent rows are written.
+    Stops a misbehaving / mis-configured camera or node from flooding
+    the events table without requiring physical access to the node.
+    Audited so there's a record of who flipped it.
+    """
+    payload = await request.json()
+    enabled = bool(payload.get("enabled"))
+    Setting.set(
+        db, user.org_id, "motion_ingestion_enabled", "true" if enabled else "false"
+    )
+    write_audit(
+        db,
+        org_id=user.org_id,
+        event="motion_ingestion_toggled",
+        user_id=user.user_id,
+        username=audit_label(user),
+        details={"enabled": enabled},
+        request=request,
+    )
+    return {"motion_ingestion_enabled": enabled}
+
+
 @router.post("/settings/notifications")
 @limiter.limit("30/minute")
 async def update_notification_settings(

backend/app/api/hls.py

@@ -13,6 +13,7 @@
 from app.core.auth import get_current_user
 from app.core.limiter import limiter
 from app.models import Camera, CameraNode, StreamAccessLog
+from app.models.models import Setting
 from app.models.models import OrgMonthlyUsage
 
 router = APIRouter(prefix="/api/cameras/{camera_id}", tags=["streaming"])
@@ -708,6 +709,17 @@ async def push_motion_event(
     if not camera:
         raise HTTPException(status_code=404, detail="Camera not found")
 
+    # Per-org kill switch.  When an admin disables ingestion (e.g. a
+    # misbehaving sensor is flooding events and you need a server-side
+    # stop without reaching the node), short-circuit before recording
+    # anything.  Returns 200 + ingested:false so the CloudNode treats
+    # this as a successful "by design" rejection and doesn't burn its
+    # retry budget — same behaviour as the plan-cap suspension path.
+    # Default "true" so orgs that never touch the toggle keep the
+    # original always-ingest behaviour.
+    if Setting.get(db, node.org_id, "motion_ingestion_enabled", "true").lower() != "true":
+        return {"success": True, "ingested": False, "reason": "ingestion_disabled"}
+
     body = await request.json()
 
     from app.api.ws import _handle_motion_event
@@ -723,4 +735,4 @@ async def push_motion_event(
         },
     )
 
-    return {"success": True}
+    return {"success": True, "ingested": True}

backend/tests/test_cameras.py

@@ -87,6 +87,40 @@ def test_notification_settings_reflected_in_all_settings(admin_client):
     assert resp.json()["notifications"]["motion_notifications"] is False
 
 
+def test_get_motion_ingestion_default_enabled(admin_client):
+    """Default is on — kill switch only filters after explicit opt-out.
+    Critical to verify because flipping the default would silently
+    disable motion ingestion on every existing org."""
+    resp = admin_client.get("/api/settings/motion-ingestion")
+    assert resp.status_code == 200
+    assert resp.json() == {"motion_ingestion_enabled": True}
+
+
+def test_motion_ingestion_toggle_persists(admin_client):
+    """Admin disables, GET reflects, admin re-enables, GET reflects."""
+    off = admin_client.post("/api/settings/motion-ingestion", json={"enabled": False})
+    assert off.status_code == 200
+    assert off.json()["motion_ingestion_enabled"] is False
+
+    follow = admin_client.get("/api/settings/motion-ingestion")
+    assert follow.json()["motion_ingestion_enabled"] is False
+
+    on = admin_client.post("/api/settings/motion-ingestion", json={"enabled": True})
+    assert on.json()["motion_ingestion_enabled"] is True
+
+    follow_on = admin_client.get("/api/settings/motion-ingestion")
+    assert follow_on.json()["motion_ingestion_enabled"] is True
+
+
+def test_motion_ingestion_toggle_requires_admin(viewer_client):
+    """Members without admin can't flip the kill switch."""
+    resp = viewer_client.post(
+        "/api/settings/motion-ingestion", json={"enabled": False}
+    )
+    # Same rejection shape as other admin-gated settings endpoints.
+    assert resp.status_code in (401, 403, 404)
+
+
 def test_update_notification_settings_requires_admin(viewer_client):
     """Non-admin members can't flip the toggles."""
     resp = viewer_client.post("/api/settings/notifications", json={