feat(integration): Phase 3 — motion SSE for Home Assistant binary_sensors

GET /api/integration/motion/stream — a Server-Sent Events feed of motion
detections across all of the org's cameras, the source for HA motion
binary_sensors. Reuses the org-wide motion pipeline the dashboard already
consumes (nodes push motion over WS → broadcast), emitting the same
{type:"motion", camera_id, node_id, score, timestamp} events with a 25s
keepalive.

Crucially it uses a SEPARATE subscriber pool (integration_motion_broadcaster)
from the dashboard, so a persistent HA connection never consumes a dashboard
SSE slot — the motion publisher in ws.py now fans out to both pools. Capped
at INTEGRATION_MAX_SSE_SUBSCRIBERS (10) streams per org, independent of the
dashboard's per-tier cap. No plan gate (all tiers).

Tests: broadcaster is a distinct instance + delivers; the two pools are
independent (filling one doesn't block the other); cap is per-org; endpoint
401 without a key and 429 when the pool is full. Full suite 660 passed.

With this the backend HA story is complete: cameras (LAN-direct video +
snapshots + recording) + motion. Remaining: Phase 1b key UI; Phase 2b
off-LAN proxy; Phase 4 the HACS component.

Author: Sbussiso

backend/app/api/integration.py

@@ -10,11 +10,14 @@
 ``app/api/mcp_keys.py`` for why the two key kinds can't cross surfaces.
 """
 
+import asyncio
 import hashlib
+import json
 import logging
 import secrets
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
 
 from app.core.audit import audit_label, write_audit
@@ -31,6 +34,13 @@
 
 KEY_PREFIX = "osi_"
 
+# Per-org cap on concurrent integration motion-SSE streams. This is a
+# SEPARATE pool from the dashboard (see integration_motion_broadcaster in
+# motion.py), so an HA connection never consumes a dashboard subscriber
+# slot. A small fixed cap is plenty — a home runs one or two HA instances —
+# and bounds memory against a scripted connect loop.
+INTEGRATION_MAX_SSE_SUBSCRIBERS = 10
+
 
 def _generate_key() -> str:
     """Generate a random integration API key: ``osi_`` + 32 hex chars."""
@@ -355,3 +365,63 @@ async def integration_status(
             "items": node_items,
         },
     }
+
+
+@router.get("/motion/stream")
+@limiter.limit("60/minute")
+async def integration_motion_stream(
+    request: Request,
+    user: AuthUser = Depends(require_integration_org),
+):
+    """Server-Sent Events feed of motion detections across ALL of the org's
+    cameras — the source for Home Assistant motion ``binary_sensor``s.
+
+    Reuses the same org-wide motion pipeline the dashboard consumes (nodes
+    push motion over WS → broadcast), but via a SEPARATE subscriber pool
+    (``integration_motion_broadcaster``) so a persistent HA connection never
+    eats into the dashboard's per-tier SSE cap.
+
+    Each event is ``{type:"motion", camera_id, node_id, score, timestamp}``;
+    a ``": keepalive"`` comment is sent every 25s to hold the connection
+    open. Capped at ``INTEGRATION_MAX_SSE_SUBSCRIBERS`` streams per org (429
+    past that).
+    """
+    from app.api.motion import integration_motion_broadcaster
+
+    org_id = user.org_id
+    queue = integration_motion_broadcaster.subscribe(
+        org_id, INTEGRATION_MAX_SSE_SUBSCRIBERS
+    )
+    if queue is None:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"Too many open integration motion streams for this org "
+                f"(cap: {INTEGRATION_MAX_SSE_SUBSCRIBERS}). Close unused "
+                f"connections and retry."
+            ),
+        )
+
+    async def event_generator():
+        try:
+            yield f"data: {json.dumps({'type': 'connected', 'org_id': org_id})}\n\n"
+            while True:
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=25.0)
+                    yield f"data: {json.dumps(event)}\n\n"
+                except TimeoutError:
+                    yield ": keepalive\n\n"
+        except asyncio.CancelledError:
+            pass
+        finally:
+            integration_motion_broadcaster.unsubscribe(org_id, queue)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )

backend/app/api/motion.py

@@ -92,6 +92,13 @@ def unsubscribe(self, org_id: str, q: asyncio.Queue):
 # Singleton — imported by ws.py to broadcast events.
 motion_broadcaster = MotionBroadcaster()
 
+# Second, independent pool for the Home Assistant integration SSE
+# (app/api/integration.py). The motion publisher in ws.py fans out to BOTH,
+# so integration subscribers get the same org-wide events — but a persistent
+# HA connection lives in its own pool and never consumes a dashboard SSE
+# subscriber slot (the dashboard cap and the integration cap are separate).
+integration_motion_broadcaster = MotionBroadcaster()
+
 
 @router.get("/events")
 async def list_motion_events(

backend/app/api/ws.py

@@ -591,17 +591,24 @@ async def _handle_motion_event(node_id: str, org_id: str, payload: dict):
             camera_id, score_int, node_id,
         )
 
-        # Broadcast to the motion SSE stream so live dashboards show
+        # Broadcast to the motion SSE streams so live dashboards show
         # toasts immediately; the inbox notification below handles the
         # durable history.
-        from app.api.motion import motion_broadcaster
-        motion_broadcaster.notify(org_id, {
+        from app.api.motion import (
+            integration_motion_broadcaster,
+            motion_broadcaster,
+        )
+        motion_payload = {
             "type": "motion",
             "camera_id": camera_id,
             "node_id": node_id,
             "score": score_int,
             "timestamp": ts.isoformat(),
-        })
+        }
+        motion_broadcaster.notify(org_id, motion_payload)
+        # Mirror to the Home Assistant integration SSE (separate pool so HA
+        # doesn't consume dashboard subscriber slots — see motion.py).
+        integration_motion_broadcaster.notify(org_id, motion_payload)
 
 
         # Also emit an inbox notification so the user can see motion

backend/tests/test_integration_motion.py

@@ -0,0 +1,101 @@
+"""Phase 3 integration motion-SSE tests.
+
+The integration motion feed reuses the org-wide motion pipeline but via a
+SEPARATE subscriber pool so a persistent Home Assistant connection never
+consumes a dashboard SSE slot. These cover the separate-pool design, the
+per-org cap, and the endpoint's auth + 429 (the streaming success path is
+left to the dashboard SSE's own coverage — it's an identical generator —
+since a TestClient GET on an infinite stream would block).
+"""
+
+import hashlib
+
+import pytest
+
+from app.api.integration import INTEGRATION_MAX_SSE_SUBSCRIBERS
+from app.api.motion import integration_motion_broadcaster, motion_broadcaster
+from app.models.models import McpApiKey
+
+ORG = "org_test123"
+RAW_KEY = "osi_phase3testkey00000000000000000000"
+
+
+@pytest.fixture(autouse=True)
+def _clear_pools():
+    """Both broadcasters are module singletons — reset their subscriber
+    pools around each test so cap/separation assertions are deterministic."""
+    integration_motion_broadcaster._subscribers.clear()
+    motion_broadcaster._subscribers.clear()
+    yield
+    integration_motion_broadcaster._subscribers.clear()
+    motion_broadcaster._subscribers.clear()
+
+
+def _make_integration_key(db, org=ORG, raw=RAW_KEY):
+    db.add(McpApiKey(
+        org_id=org,
+        key_hash=hashlib.sha256(raw.encode()).hexdigest(),
+        name="Home Assistant",
+        kind="integration",
+    ))
+    db.commit()
+
+
+# ── Separate-pool design ────────────────────────────────────────────
+
+def test_integration_broadcaster_is_a_separate_instance():
+    assert integration_motion_broadcaster is not motion_broadcaster
+
+
+def test_integration_broadcaster_delivers_events():
+    q = integration_motion_broadcaster.subscribe(ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS)
+    payload = {"type": "motion", "camera_id": "cam_a", "score": 80}
+    integration_motion_broadcaster.notify(ORG, payload)
+    assert q.get_nowait() == payload
+
+
+def test_pools_are_independent():
+    """Filling the integration pool to its cap must NOT block the dashboard
+    pool (and vice versa) — HA connections and dashboard tabs don't contend."""
+    for _ in range(INTEGRATION_MAX_SSE_SUBSCRIBERS):
+        assert integration_motion_broadcaster.subscribe(
+            ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS
+        ) is not None
+    # Integration pool is now full...
+    assert integration_motion_broadcaster.subscribe(
+        ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS
+    ) is None
+    # ...but the dashboard pool is entirely untouched.
+    assert motion_broadcaster.subscribe(ORG, 5) is not None
+
+
+def test_cap_is_per_org():
+    """One org filling its integration pool doesn't affect another org."""
+    for _ in range(INTEGRATION_MAX_SSE_SUBSCRIBERS):
+        integration_motion_broadcaster.subscribe(ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS)
+    assert integration_motion_broadcaster.subscribe(ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS) is None
+    # A different org has its own budget.
+    assert integration_motion_broadcaster.subscribe(
+        "org_other", INTEGRATION_MAX_SSE_SUBSCRIBERS
+    ) is not None
+
+
+# ── Endpoint ────────────────────────────────────────────────────────
+
+def test_motion_stream_requires_key(unauthenticated_client):
+    # 401 is raised by the auth dependency before any streaming begins.
+    assert unauthenticated_client.get("/api/integration/motion/stream").status_code == 401
+
+
+def test_motion_stream_429_when_pool_full(unauthenticated_client, db):
+    """A valid key still 429s when the org's integration pool is full — the
+    429 is raised before the StreamingResponse, so the GET returns cleanly."""
+    _make_integration_key(db)
+    for _ in range(INTEGRATION_MAX_SSE_SUBSCRIBERS):
+        integration_motion_broadcaster.subscribe(ORG, INTEGRATION_MAX_SSE_SUBSCRIBERS)
+
+    resp = unauthenticated_client.get(
+        "/api/integration/motion/stream",
+        headers={"Authorization": f"Bearer {RAW_KEY}"},
+    )
+    assert resp.status_code == 429