feat: receive and store motion detection events from CloudNodes

Handles the new "event" WS message type with "motion_detected"
command, persisting MotionEvent rows per org. Adds REST endpoints
for querying motion events with camera filtering, time windows,
and per-camera aggregate stats.

- MotionEvent model (motion_events table)
- WS handler for motion_detected events
- GET /api/motion/events + /api/motion/events/stats
- Motion events included in log retention cleanup
- 5 new tests (26 total, all passing)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Sbussiso

backend/app/api/motion.py

@@ -0,0 +1,95 @@
+"""
+Motion Events API — query motion detection events reported by CloudNodes.
+"""
+
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.auth import AuthUser, require_view
+from app.core.database import get_db
+from app.models.models import MotionEvent
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/motion", tags=["motion"])
+
+
+@router.get("/events")
+async def list_motion_events(
+    camera_id: Optional[str] = None,
+    hours: int = Query(default=24, le=168),
+    limit: int = Query(default=100, le=500),
+    offset: int = Query(default=0, ge=0),
+    user: AuthUser = Depends(require_view),
+    db: Session = Depends(get_db),
+):
+    """List recent motion events, optionally filtered by camera."""
+    since = datetime.now(tz=timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
+
+    query = db.query(MotionEvent).filter(
+        MotionEvent.org_id == user.org_id,
+        MotionEvent.timestamp >= since,
+    )
+
+    if camera_id:
+        query = query.filter(MotionEvent.camera_id == camera_id)
+
+    total = query.count()
+    events = (
+        query.order_by(MotionEvent.timestamp.desc())
+        .offset(offset)
+        .limit(limit)
+        .all()
+    )
+
+    return {
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+        "hours": hours,
+        "events": [e.to_dict() for e in events],
+    }
+
+
+@router.get("/events/stats")
+async def motion_stats(
+    hours: int = Query(default=24, le=168),
+    user: AuthUser = Depends(require_view),
+    db: Session = Depends(get_db),
+):
+    """Aggregate motion stats per camera over the given time window."""
+    from sqlalchemy import func
+
+    since = datetime.now(tz=timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
+
+    rows = (
+        db.query(
+            MotionEvent.camera_id,
+            func.count(MotionEvent.id).label("count"),
+            func.max(MotionEvent.score).label("peak_score"),
+            func.max(MotionEvent.timestamp).label("latest"),
+        )
+        .filter(
+            MotionEvent.org_id == user.org_id,
+            MotionEvent.timestamp >= since,
+        )
+        .group_by(MotionEvent.camera_id)
+        .all()
+    )
+
+    return {
+        "hours": hours,
+        "cameras": [
+            {
+                "camera_id": r.camera_id,
+                "event_count": r.count,
+                "peak_score": r.peak_score,
+                "latest": r.latest.isoformat() if r.latest else None,
+            }
+            for r in rows
+        ],
+    }

backend/app/api/ws.py

@@ -29,7 +29,7 @@
 from sqlalchemy.orm import Session
 
 from app.core.database import SessionLocal
-from app.models import CameraNode, Camera
+from app.models import CameraNode, Camera, MotionEvent
 
 logger = logging.getLogger(__name__)
 
@@ -162,6 +162,14 @@ async def node_websocket(
                 if correlation_id:
                     manager.resolve_command(correlation_id, data.get("payload", {}))
 
+            elif msg_type == "event":
+                command = data.get("command")
+                payload = data.get("payload", {})
+                if command == "motion_detected":
+                    await _handle_motion_event(node_id, org_id, payload)
+                else:
+                    logger.debug("Unhandled event command from node %s: %s", node_id, command)
+
             else:
                 logger.warning("Unknown WS message type from node %s: %s", node_id, msg_type)
                 await ws.send_json({
@@ -217,3 +225,48 @@ async def _handle_heartbeat(node_id: str, node_db_id: int, org_id: str, payload:
         db.rollback()
     finally:
         db.close()
+
+
+# ── Motion Event Handler ─────────────────────────────────────────────
+
+async def _handle_motion_event(node_id: str, org_id: str, payload: dict):
+    """Persist a motion detection event reported by a CloudNode."""
+    camera_id = payload.get("camera_id")
+    score = payload.get("score")
+    segment_seq = payload.get("segment_seq")
+    event_ts = payload.get("timestamp")  # ISO 8601 from node
+
+    if not camera_id or score is None:
+        logger.warning("Motion event from node %s missing camera_id or score", node_id)
+        return
+
+    ts = None
+    if event_ts:
+        try:
+            ts = datetime.fromisoformat(event_ts).replace(tzinfo=None)
+        except (ValueError, TypeError):
+            pass
+    if ts is None:
+        ts = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+
+    db: Session = SessionLocal()
+    try:
+        event = MotionEvent(
+            org_id=org_id,
+            camera_id=camera_id,
+            node_id=node_id,
+            score=int(score),
+            segment_seq=int(segment_seq) if segment_seq is not None else None,
+            timestamp=ts,
+        )
+        db.add(event)
+        db.commit()
+        logger.info(
+            "Motion event: camera=%s score=%d%% node=%s",
+            camera_id, int(score), node_id,
+        )
+    except Exception as e:
+        logger.error("Failed to save motion event: %s", e)
+        db.rollback()
+    finally:
+        db.close()

backend/app/main.py

@@ -13,7 +13,7 @@
 from app.core.config import settings
 from app.core.database import Base, engine, SessionLocal
 from app.core.limiter import limiter
-from app.api import cameras, webhooks, nodes, audit, hls, ws, install, mcp_keys, mcp_activity, incidents
+from app.api import cameras, webhooks, nodes, audit, hls, ws, install, mcp_keys, mcp_activity, incidents, motion
 from app.mcp.server import mcp
 
 logger = logging.getLogger(__name__)
@@ -105,6 +105,7 @@ async def security_headers(request: Request, call_next):
 app.include_router(mcp_keys.router)
 app.include_router(mcp_activity.router)
 app.include_router(incidents.router)
+app.include_router(motion.router)
 
 # Mount MCP server at /mcp
 app.mount("/mcp", mcp_app)
@@ -121,7 +122,7 @@ async def security_headers(request: Request, call_next):
 async def _log_cleanup_loop():
     """Background task: delete old logs and free segment caches
     for cameras that have been offline."""
-    from app.models.models import StreamAccessLog, McpActivityLog, AuditLog, Camera
+    from app.models.models import StreamAccessLog, McpActivityLog, AuditLog, MotionEvent, Camera
 
     while True:
         await asyncio.sleep(LOG_CLEANUP_INTERVAL_HOURS * 3600)
@@ -134,12 +135,13 @@ async def _log_cleanup_loop():
                 stream_count = db.query(StreamAccessLog).filter(StreamAccessLog.accessed_at < cutoff).delete()
                 mcp_count = db.query(McpActivityLog).filter(McpActivityLog.timestamp < cutoff).delete()
                 audit_count = db.query(AuditLog).filter(AuditLog.timestamp < cutoff).delete()
+                motion_count = db.query(MotionEvent).filter(MotionEvent.timestamp < cutoff).delete()
                 db.commit()
-                total = stream_count + mcp_count + audit_count
+                total = stream_count + mcp_count + audit_count + motion_count
                 if total > 0:
                     logger.info(
-                        "[Cleanup] Deleted %d old logs (stream=%d, mcp=%d, audit=%d, retention=%dd)",
-                        total, stream_count, mcp_count, audit_count, LOG_RETENTION_DAYS,
+                        "[Cleanup] Deleted %d old logs (stream=%d, mcp=%d, audit=%d, motion=%d, retention=%dd)",
+                        total, stream_count, mcp_count, audit_count, motion_count, LOG_RETENTION_DAYS,
                     )
             finally:
                 db.close()

backend/app/models/__init__.py

@@ -8,6 +8,7 @@
     McpActivityLog,
     Incident,
     IncidentEvidence,
+    MotionEvent,
 )
 
 __all__ = [
@@ -20,4 +21,5 @@
     "McpActivityLog",
     "Incident",
     "IncidentEvidence",
+    "MotionEvent",
 ]

backend/app/models/models.py

@@ -359,3 +359,36 @@ def to_dict(self) -> dict:
         }
 
 
+class MotionEvent(Base):
+    """A motion detection event reported by a CloudNode.
+
+    Created when a node's FFmpeg scene-change analysis exceeds the
+    configured threshold for a camera segment.
+    """
+
+    __tablename__ = "motion_events"
+
+    id = Column(Integer, primary_key=True)
+    org_id = Column(String(100), nullable=False, index=True)
+    camera_id = Column(String(100), nullable=False, index=True)
+    node_id = Column(String(100), nullable=False, index=True)
+    score = Column(Integer, nullable=False)  # 0-100 (normalised)
+    segment_seq = Column(Integer, nullable=True)
+    timestamp = Column(
+        DateTime,
+        default=lambda: datetime.now(tz=timezone.utc).replace(tzinfo=None),
+        index=True,
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "org_id": self.org_id,
+            "camera_id": self.camera_id,
+            "node_id": self.node_id,
+            "score": self.score,
+            "segment_seq": self.segment_seq,
+            "timestamp": self.timestamp.isoformat() if self.timestamp else None,
+        }
+
+

backend/tests/test_motion.py

@@ -0,0 +1,89 @@
+"""Tests for motion detection event endpoints."""
+
+from datetime import datetime, timedelta, timezone
+from app.models.models import MotionEvent
+
+
+def test_list_motion_events_empty(viewer_client):
+    """No motion events returns empty list."""
+    resp = viewer_client.get("/api/motion/events")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["total"] == 0
+    assert data["events"] == []
+
+
+def test_list_motion_events_with_data(viewer_client, db):
+    """Motion events are returned in reverse chronological order."""
+    now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+    for i in range(3):
+        db.add(MotionEvent(
+            org_id="org_test123",
+            camera_id="cam_front",
+            node_id="node_1",
+            score=50 + i * 10,
+            segment_seq=100 + i,
+            timestamp=now - timedelta(minutes=10 - i),
+        ))
+    db.commit()
+
+    resp = viewer_client.get("/api/motion/events")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["total"] == 3
+    assert len(data["events"]) == 3
+    # Most recent first
+    assert data["events"][0]["score"] == 70
+    assert data["events"][2]["score"] == 50
+
+
+def test_list_motion_events_camera_filter(viewer_client, db):
+    """Filtering by camera_id returns only matching events."""
+    now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+    db.add(MotionEvent(org_id="org_test123", camera_id="cam_front", node_id="n1", score=80, timestamp=now))
+    db.add(MotionEvent(org_id="org_test123", camera_id="cam_back", node_id="n1", score=60, timestamp=now))
+    db.commit()
+
+    resp = viewer_client.get("/api/motion/events?camera_id=cam_front")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["total"] == 1
+    assert data["events"][0]["camera_id"] == "cam_front"
+
+
+def test_motion_events_org_isolation(viewer_client, db):
+    """Events from other orgs are not visible."""
+    now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+    db.add(MotionEvent(org_id="org_test123", camera_id="cam_1", node_id="n1", score=90, timestamp=now))
+    db.add(MotionEvent(org_id="org_other", camera_id="cam_2", node_id="n2", score=85, timestamp=now))
+    db.commit()
+
+    resp = viewer_client.get("/api/motion/events")
+    data = resp.json()
+    assert data["total"] == 1
+    assert data["events"][0]["camera_id"] == "cam_1"
+
+
+def test_motion_stats(viewer_client, db):
+    """Stats endpoint returns per-camera aggregates."""
+    now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
+    for score in [40, 60, 80]:
+        db.add(MotionEvent(
+            org_id="org_test123", camera_id="cam_front", node_id="n1",
+            score=score, timestamp=now - timedelta(minutes=score),
+        ))
+    db.add(MotionEvent(
+        org_id="org_test123", camera_id="cam_back", node_id="n1",
+        score=95, timestamp=now,
+    ))
+    db.commit()
+
+    resp = viewer_client.get("/api/motion/events/stats")
+    assert resp.status_code == 200
+    data = resp.json()
+    cameras = {c["camera_id"]: c for c in data["cameras"]}
+    assert len(cameras) == 2
+    assert cameras["cam_front"]["event_count"] == 3
+    assert cameras["cam_front"]["peak_score"] == 80
+    assert cameras["cam_back"]["event_count"] == 1
+    assert cameras["cam_back"]["peak_score"] == 95