Enhance documentation and improve WebSocket event handling

README-compose.md

@@ -12,6 +12,10 @@ Lifecycle scripts (mode-aware) are available:
 - `./scripts/start.sh --prod`
 - `./scripts/status.sh --dev|--prod`
 - `./scripts/stop.sh --dev|--prod [--wipe]`
+- `.\scripts\start.ps1 --dev`
+- `.\scripts\start.ps1 --prod`
+- `.\scripts\status.ps1 --dev|--prod`
+- `.\scripts\stop.ps1 --dev|--prod [--wipe]`
 
 ## 1) Prepare environment and secrets
 

README.md

@@ -52,6 +52,10 @@ Run the entire platform with one command:
 ./scripts/start.sh
 ```
 
+```powershell
+.\scripts\start.ps1
+```
+
 ### Services started:
 
 | Service  | Purpose                                 | Exposed port |
@@ -65,7 +69,7 @@ Run the entire platform with one command:
 
 Open the app at `http://localhost:3000`. The frontend container proxies `/api`, `/auth`, `/health`, `/metrics`, `/docs`, `/redoc`, `/openapi.json` and `/ws/*` (with WebSocket upgrade) to the `api` service, so browser clients only need to talk to port 3000.
 
-### What `start.sh` does automatically
+### What `start.sh` / `start.ps1` do automatically
 
 - verifies required tools (`docker`, `docker compose`, `curl`, `rg`)
 - creates `.env` on first run with generated secrets (`POSTGRES_PASSWORD`, `JWT_SECRET_KEY`)
@@ -92,10 +96,17 @@ docker compose logs -f frontend api worker       # tail live logs
 ./scripts/stop.sh --wipe                         # stop + wipe volumes
 ```
 
+```powershell
+.\scripts\status.ps1                             # compose + health summary
+docker compose logs -f frontend api worker       # tail live logs
+.\scripts\stop.ps1                               # stop stack
+.\scripts\stop.ps1 --wipe                        # stop + wipe volumes
+```
+
 ### Troubleshooting
 
 - If Docker daemon is not running, start Docker Desktop / Docker Engine first.
-- If startup fails after a major change, run `./scripts/stop.sh --wipe` and then `./scripts/start.sh`.
+- If startup fails after a major change, run `./scripts/stop.sh --wipe` (or `.\scripts\stop.ps1 --wipe` on Windows) and then restart.
 - If ports are already in use, stop conflicting services or adjust host port bindings in `compose.yaml` plus overlay files.
 
 ## 🎮 Examples & Demo
@@ -131,7 +142,7 @@ During dev the Vite server listens on `http://localhost:5173`; since `VITE_API_B
 
 If you are new to this codebase, follow this order:
 
-1. **Run the stack first** with `./scripts/start.sh` so you can see the full upload → analysis → results flow end-to-end.
+1. **Run the stack first** with `./scripts/start.sh` (or `.\scripts\start.ps1` on Windows) so you can see the full upload → analysis → results flow end-to-end.
 2. **Read the backend entrypoint** (`backend/app/main.py`) and API router composition (`backend/app/api/router.py`) to understand how routes are wired.
 3. **Study the analysis pipeline** in `backend/app/application/services/analysis_service.py` and worker task orchestration in `backend/app/workers/tasks.py`.
 4. **Review plugin-based checks** under `backend/app/plugins/` (Hadolint, compose runnability, security checks, resource estimation).

backend/app/api/routers/ws.py

@@ -4,11 +4,59 @@
 from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect
 
 from app.core.security import decode_token
+from app.infrastructure.db.repositories import JobRepository
+from app.infrastructure.db.session import SessionLocal
+from app.infrastructure.events.bus import redis_client
 from app.infrastructure.events.bus import subscribe
 
 router = APIRouter(tags=["websockets"])
 
 
+async def _job_belongs_to_user(job_id: uuid.UUID, user_id: uuid.UUID) -> bool:
+    async with SessionLocal() as session:
+        repo = JobRepository(session)
+        job = await repo.get_job(job_id, user_id)
+        return job is not None
+
+
+async def _user_has_container_access(user_id: uuid.UUID, container_id: str) -> bool:
+    pattern = f"deploy:{user_id}:*"
+    cursor = 0
+    while True:
+        cursor, keys = await redis_client.scan(cursor=cursor, match=pattern, count=100)
+        for key in keys:
+            raw_state = await redis_client.get(key)
+            if not raw_state:
+                continue
+            try:
+                state = json.loads(raw_state)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(state, dict):
+                continue
+            ids: set[str] = set()
+            top_id = state.get("container_id")
+            if isinstance(top_id, str) and top_id:
+                ids.add(top_id)
+            top_ids = state.get("container_ids")
+            if isinstance(top_ids, list):
+                for row in top_ids:
+                    if isinstance(row, str) and row:
+                        ids.add(row)
+            containers = state.get("containers")
+            if isinstance(containers, list):
+                for row in containers:
+                    if isinstance(row, dict):
+                        row_id = row.get("id")
+                        if isinstance(row_id, str) and row_id:
+                            ids.add(row_id)
+            if container_id in ids:
+                return True
+        if cursor == 0:
+            break
+    return False
+
+
 @router.websocket("/ws/jobs/{job_id}")
 async def job_updates(
     websocket: WebSocket,
@@ -20,13 +68,18 @@ async def job_updates(
         payload = decode_token(token)
         if payload.get("type") != "access":
             raise ValueError("wrong token type")
-        _ = uuid.UUID(payload["sub"])
-        uuid.UUID(job_id)
-    except Exception:
+        user_id = uuid.UUID(payload["sub"])
+        job_uuid = uuid.UUID(job_id)
+    except (KeyError, TypeError, ValueError):
         await websocket.send_json({"error": "invalid token"})
         await websocket.close(code=4401)
         return
 
+    if not await _job_belongs_to_user(job_uuid, user_id):
+        await websocket.send_json({"error": "forbidden"})
+        await websocket.close(code=4403)
+        return
+
     try:
         async for event in subscribe(f"job:{job_id}:events"):
             await websocket.send_text(json.dumps(event))
@@ -45,11 +98,16 @@ async def container_metrics(
         payload = decode_token(token)
         if payload.get("type") != "access":
             raise ValueError("wrong token type")
-        _ = uuid.UUID(payload["sub"])
-    except Exception:
+        user_id = uuid.UUID(payload["sub"])
+    except (KeyError, TypeError, ValueError):
         await websocket.send_json({"error": "invalid token"})
         await websocket.close(code=4401)
         return
+
+    if not await _user_has_container_access(user_id, container_id):
+        await websocket.send_json({"error": "forbidden"})
+        await websocket.close(code=4403)
+        return
     try:
         async for event in subscribe(f"container:{container_id}:metrics"):
             await websocket.send_text(json.dumps(event))
@@ -72,7 +130,7 @@ async def user_container_metrics(
         if payload.get("sub") != user_id:
             raise ValueError("user mismatch")
         uuid.UUID(user_id)
-    except Exception:
+    except (KeyError, TypeError, ValueError):
         await websocket.send_json({"error": "invalid token"})
         await websocket.close(code=4401)
         return
@@ -81,3 +139,29 @@ async def user_container_metrics(
             await websocket.send_text(json.dumps(event))
     except WebSocketDisconnect:
         return
+
+
+@router.websocket("/ws/users/{user_id}/events")
+async def user_events(
+    websocket: WebSocket,
+    user_id: str,
+    token: str = Query(...),
+) -> None:
+    await websocket.accept()
+    try:
+        payload = decode_token(token)
+        if payload.get("type") != "access":
+            raise ValueError("wrong token type")
+        if payload.get("sub") != user_id:
+            raise ValueError("user mismatch")
+        uuid.UUID(user_id)
+    except (KeyError, TypeError, ValueError):
+        await websocket.send_json({"error": "invalid token"})
+        await websocket.close(code=4401)
+        return
+
+    try:
+        async for event in subscribe(f"user:{user_id}:events"):
+            await websocket.send_text(json.dumps(event))
+    except WebSocketDisconnect:
+        return

backend/app/application/services/analysis_service.py

@@ -64,6 +64,22 @@ def _grade(score: int) -> str:
     return "F"
 
 
+def _build_analysis_started_payload(context: dict[str, Any]) -> dict[str, Any]:
+    filename = context.get("filename")
+    source = context.get("source")
+    dockerfile_content = context.get("dockerfile_content")
+    compose_content = context.get("compose_content")
+    analysis_type = "compose" if isinstance(compose_content, str) and compose_content else "dockerfile"
+    effective_source = compose_content if analysis_type == "compose" else dockerfile_content
+    line_count = len(effective_source.splitlines()) if isinstance(effective_source, str) else 0
+    payload: dict[str, Any] = {"analysis_type": analysis_type, "line_count": max(1, line_count)}
+    if isinstance(filename, str) and filename:
+        payload["filename"] = filename
+    if isinstance(source, str) and source:
+        payload["source"] = source
+    return payload
+
+
 class AnalysisService:
     def __init__(self, session: AsyncSession) -> None:
         self.session = session
@@ -140,7 +156,8 @@ async def analyze_content(
     async def run_job_with_plugins(
         self, user_id: uuid.UUID, job_id: uuid.UUID, context: dict[str, Any], plugin_names: Iterable[str]
     ) -> dict[str, Any]:
-        await publish_event(DomainEvent("user.analysis.started", str(user_id), str(job_id), payload=context))
+        safe_start_payload = _build_analysis_started_payload(context)
+        await publish_event(DomainEvent("user.analysis.started", str(user_id), str(job_id), payload=safe_start_payload))
         await self.repo.update_status(job_id, user_id, JobStatus.running)
         await self.session.commit()
 

backend/app/workers/tasks.py

@@ -88,7 +88,17 @@ async def run_project_analysis(ctx, payload: dict) -> dict:
         repo = JobRepository(session)
         svc = AnalysisService(session)
 
-        await publish_event(DomainEvent("project.analysis_started", str(user_id), str(job_id), payload={"project_path": str(project_path)}))
+        await publish_event(
+            DomainEvent(
+                "project.analysis_started",
+                str(user_id),
+                str(job_id),
+                payload={
+                    "dockerfile_count": len(dockerfiles),
+                    "compose_count": len(compose_files),
+                },
+            )
+        )
         await repo.update_status(job_id, user_id, JobStatus.running)
         await session.commit()
 

backend/tests/test_analysis_service.py

@@ -1,6 +1,6 @@
 from unittest.mock import AsyncMock
 
-from app.application.services.analysis_service import AnalysisService
+from app.application.services.analysis_service import AnalysisService, _build_analysis_started_payload
 
 
 def test_normalize_issue_uses_level_when_severity_missing():
@@ -70,3 +70,20 @@ def test_normalize_issue_dclint_rdjson_code_object():
     assert issue.line == 7
     assert issue.severity == "error"
     assert issue.suggestion == "Bind ports to a host interface."
+
+
+def test_build_analysis_started_payload_omits_source_content():
+    payload = _build_analysis_started_payload(
+        {
+            "filename": "docker-compose.yml",
+            "source": "upload",
+            "compose_content": "services:\n  web:\n    image: nginx:1.27\n",
+            "dockerfile_content": "",
+        }
+    )
+    assert payload["analysis_type"] == "compose"
+    assert payload["line_count"] == 3
+    assert payload["filename"] == "docker-compose.yml"
+    assert payload["source"] == "upload"
+    assert "compose_content" not in payload
+    assert "dockerfile_content" not in payload

backend/tests/test_ws_metrics.py

@@ -1,5 +1,6 @@
 import json
 import uuid
+from unittest.mock import AsyncMock
 
 from app.core.security import create_token
 
@@ -18,6 +19,7 @@ async def fake_subscribe(channel: str):
         }
 
     monkeypatch.setattr("app.api.routers.ws.subscribe", fake_subscribe)
+    monkeypatch.setattr("app.api.routers.ws._user_has_container_access", AsyncMock(return_value=True))
 
     with client.websocket_connect(f"/ws/metrics/web-container?token={token}") as websocket:
         message = websocket.receive_text()
@@ -42,6 +44,7 @@ async def fake_subscribe(channel: str):
         }
 
     monkeypatch.setattr("app.api.routers.ws.subscribe", fake_subscribe)
+    monkeypatch.setattr("app.api.routers.ws._job_belongs_to_user", AsyncMock(return_value=True))
 
     with client.websocket_connect(f"/ws/jobs/{job_id}?token={token}") as websocket:
         message = websocket.receive_text()
@@ -77,3 +80,54 @@ async def fake_subscribe(channel: str):
     ) as websocket:
         payload = websocket.receive_json()
         assert payload["error"] == "invalid token"
+
+
+def test_ws_jobs_rejects_non_owner_subscription(client, monkeypatch):
+    user_id = uuid.uuid4()
+    job_id = uuid.uuid4()
+    token = create_token(str(user_id), "access", 30)
+    monkeypatch.setattr("app.api.routers.ws._job_belongs_to_user", AsyncMock(return_value=False))
+
+    with client.websocket_connect(f"/ws/jobs/{job_id}?token={token}") as websocket:
+        payload = websocket.receive_json()
+        assert payload["error"] == "forbidden"
+
+
+def test_ws_metrics_rejects_non_owner_subscription(client, monkeypatch):
+    user_id = uuid.uuid4()
+    token = create_token(str(user_id), "access", 30)
+    monkeypatch.setattr("app.api.routers.ws._user_has_container_access", AsyncMock(return_value=False))
+
+    with client.websocket_connect(f"/ws/metrics/web-container?token={token}") as websocket:
+        payload = websocket.receive_json()
+        assert payload["error"] == "forbidden"
+
+
+def test_ws_user_events_streams_with_matching_token(client, monkeypatch):
+    user_id = uuid.uuid4()
+    token = create_token(str(user_id), "access", 30)
+
+    async def fake_subscribe(channel: str):
+        assert channel == f"user:{user_id}:events"
+        yield {
+            "event_name": "user.analysis.completed",
+            "user_id": str(user_id),
+            "job_id": "j-1",
+            "payload": {"status": "done"},
+        }
+
+    monkeypatch.setattr("app.api.routers.ws.subscribe", fake_subscribe)
+
+    with client.websocket_connect(f"/ws/users/{user_id}/events?token={token}") as websocket:
+        payload = websocket.receive_json()
+        assert payload["event_name"] == "user.analysis.completed"
+
+
+def test_ws_user_events_rejects_user_mismatch(client):
+    user_id = uuid.uuid4()
+    other_user_id = uuid.uuid4()
+    token = create_token(str(user_id), "access", 30)
+
+    with client.websocket_connect(f"/ws/users/{other_user_id}/events?token={token}") as websocket:
+        payload = websocket.receive_json()
+        assert payload["error"] == "invalid token"