diff --git a/docs/plans/2026-03-12-live-announce-dashboard-impl.md b/docs/plans/2026-03-12-live-announce-dashboard-impl.md new file mode 100644 index 0000000..2528579 --- /dev/null +++ b/docs/plans/2026-03-12-live-announce-dashboard-impl.md @@ -0,0 +1,1261 @@ +# Live Announce Dashboard Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Add a web dashboard that lets Nick announce tracks to IRC in real-time during live shows via WebSocket push, with both Sopel and Limnoria plugins receiving announcements. + +**Architecture:** FastAPI serves a login page + dashboard page as static HTML. A WebSocket endpoint at `/ws/announce` acts as a broadcast hub. The dashboard POSTs to `/admin/announce`, which formats the message and pushes it to all connected WS subscriber bots. Bot plugins run a background thread with a `websocket-client` connection that listens for announcements and emits them to a configured IRC channel. + +**Tech Stack:** FastAPI (WS support built-in), `itsdangerous` (cookie signing), `websocket-client` (bot plugins), Pico CSS (CDN), vanilla JS. + +**Design doc:** `docs/plans/2026-03-12-live-announce-dashboard-design.md` + +--- + +## Task 1: Add optional dashboard config fields + +**Files:** +- Modify: `src/ntr_fetcher/config.py` +- Test: `tests/test_config.py` (create) + +**Step 1: Write the failing test** + +```python +# tests/test_config.py +import os +from ntr_fetcher.config import Settings + + +def test_dashboard_config_absent(monkeypatch): + monkeypatch.setenv("NTR_ADMIN_TOKEN", "tok") + monkeypatch.delenv("NTR_WEB_USER", raising=False) + monkeypatch.delenv("NTR_WEB_PASSWORD", raising=False) + monkeypatch.delenv("NTR_SECRET_KEY", raising=False) + s = Settings() + assert s.web_user is None + assert s.web_password is None + assert s.secret_key is None + assert s.dashboard_enabled is False + + +def test_dashboard_config_present(monkeypatch): + monkeypatch.setenv("NTR_ADMIN_TOKEN", "tok") + monkeypatch.setenv("NTR_WEB_USER", "nick") + monkeypatch.setenv("NTR_WEB_PASSWORD", "secret") + monkeypatch.setenv("NTR_SECRET_KEY", "signme") + s = Settings() + assert s.web_user == "nick" + assert s.web_password == "secret" + assert s.secret_key == "signme" + assert s.dashboard_enabled is True +``` + +**Step 2: Run test to verify it fails** + +Run: `pytest tests/test_config.py -v` +Expected: FAIL — `Settings` has no `web_user` / `dashboard_enabled` attributes. + +**Step 3: Write minimal implementation** + +In `src/ntr_fetcher/config.py`, add three optional fields and a computed property: + +```python +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + model_config = {"env_prefix": "NTR_"} + + port: int = 8000 + host: str = "127.0.0.1" + db_path: str = "./ntr_fetcher.db" + poll_interval_seconds: int = 3600 + admin_token: str + soundcloud_user: str = "nicktherat" + show_day: int = 2 + show_hour: int = 22 + + web_user: str | None = None + web_password: str | None = None + secret_key: str | None = None + + @property + def dashboard_enabled(self) -> bool: + return all([self.web_user, self.web_password, self.secret_key]) +``` + +**Step 4: Run test to verify it passes** + +Run: `pytest tests/test_config.py -v` +Expected: PASS + +**Step 5: Commit** + +```bash +git add src/ntr_fetcher/config.py tests/test_config.py +git commit -m "feat: add optional dashboard config fields" +``` + +--- + +## Task 2: Add `itsdangerous` dependency + +**Files:** +- Modify: `pyproject.toml` + +**Step 1: Add dependency** + +Add `"itsdangerous"` to the `dependencies` list in `pyproject.toml`: + +```toml +dependencies = [ + "fastapi", + "uvicorn[standard]", + "httpx", + "pydantic-settings", + "itsdangerous", +] +``` + +**Step 2: Install** + +Run: `pip install -e ".[dev]"` +Expected: installs `itsdangerous` + +**Step 3: Commit** + +```bash +git add pyproject.toml +git commit -m "chore: add itsdangerous dependency for session cookies" +``` + +--- + +## Task 3: WebSocket manager + +**Files:** +- Create: `src/ntr_fetcher/websocket.py` +- Test: `tests/test_websocket.py` (create) + +**Step 1: Write the failing test** + +```python +# tests/test_websocket.py +import asyncio +import pytest +from ntr_fetcher.websocket import AnnounceManager + + +@pytest.fixture +def manager(): + return AnnounceManager() + + +def test_no_subscribers_initially(manager): + assert manager.subscriber_count == 0 + + +@pytest.mark.asyncio +async def test_subscribe_and_broadcast(manager): + received = [] + + async def fake_send(msg): + received.append(msg) + + class FakeWS: + async def send_json(self, data): + received.append(data) + + ws = FakeWS() + manager.add_subscriber(ws) + assert manager.subscriber_count == 1 + + await manager.broadcast({"type": "announce", "message": "Now Playing: Song #1"}) + assert len(received) == 1 + assert received[0]["message"] == "Now Playing: Song #1" + + manager.remove_subscriber(ws) + assert manager.subscriber_count == 0 + + +@pytest.mark.asyncio +async def test_broadcast_skips_dead_connections(manager): + class DeadWS: + async def send_json(self, data): + raise Exception("connection closed") + + ws = DeadWS() + manager.add_subscriber(ws) + assert manager.subscriber_count == 1 + + await manager.broadcast({"type": "announce", "message": "test"}) + assert manager.subscriber_count == 0 +``` + +**Step 2: Run test to verify it fails** + +Run: `pytest tests/test_websocket.py -v` +Expected: FAIL — `ntr_fetcher.websocket` does not exist. + +**Step 3: Write minimal implementation** + +```python +# src/ntr_fetcher/websocket.py +import logging + +logger = logging.getLogger(__name__) + + +class AnnounceManager: + def __init__(self): + self._subscribers: list = [] + + @property + def subscriber_count(self) -> int: + return len(self._subscribers) + + def add_subscriber(self, websocket) -> None: + self._subscribers.append(websocket) + logger.info("Subscriber connected (%d total)", self.subscriber_count) + + def remove_subscriber(self, websocket) -> None: + self._subscribers = [ws for ws in self._subscribers if ws is not websocket] + logger.info("Subscriber disconnected (%d total)", self.subscriber_count) + + async def broadcast(self, message: dict) -> None: + dead = [] + for ws in self._subscribers: + try: + await ws.send_json(message) + except Exception: + dead.append(ws) + logger.warning("Removing dead subscriber") + for ws in dead: + self.remove_subscriber(ws) + + async def broadcast_status(self) -> None: + await self.broadcast({ + "type": "status", + "subscribers": self.subscriber_count, + }) +``` + +**Step 4: Run test to verify it passes** + +Run: `pytest tests/test_websocket.py -v` +Expected: PASS + +**Step 5: Commit** + +```bash +git add src/ntr_fetcher/websocket.py tests/test_websocket.py +git commit -m "feat: add WebSocket announce manager" +``` + +--- + +## Task 4: Dashboard router — session auth helpers + +**Files:** +- Create: `src/ntr_fetcher/dashboard.py` +- Test: `tests/test_dashboard.py` (create) + +This task covers session cookie creation/validation and the login/logout flow. The announce endpoint and WS handler come in Tasks 5–6. + +**Step 1: Write the failing test** + +```python +# tests/test_dashboard.py +from datetime import datetime, timezone +from unittest.mock import MagicMock, AsyncMock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from ntr_fetcher.dashboard import create_dashboard_router +from ntr_fetcher.db import Database +from ntr_fetcher.websocket import AnnounceManager + + +@pytest.fixture +def db(tmp_path): + database = Database(str(tmp_path / "test.db")) + database.initialize() + return database + + +@pytest.fixture +def manager(): + return AnnounceManager() + + +@pytest.fixture +def app(db, manager): + a = FastAPI() + router = create_dashboard_router( + db=db, + manager=manager, + admin_token="test-token", + web_user="nick", + web_password="secret", + secret_key="test-secret-key", + show_day=2, + show_hour=22, + ) + a.include_router(router) + return a + + +@pytest.fixture +def client(app): + return TestClient(app) + + +def test_dashboard_redirects_without_session(client): + resp = client.get("/dashboard", follow_redirects=False) + assert resp.status_code == 303 + assert "/login" in resp.headers["location"] + + +def test_login_page_renders(client): + resp = client.get("/login") + assert resp.status_code == 200 + assert "login" in resp.text.lower() + + +def test_login_with_valid_credentials(client): + resp = client.post( + "/login", + data={"username": "nick", "password": "secret"}, + follow_redirects=False, + ) + assert resp.status_code == 303 + assert "/dashboard" in resp.headers["location"] + assert "ntr_session" in resp.cookies + + +def test_login_with_invalid_credentials(client): + resp = client.post( + "/login", + data={"username": "nick", "password": "wrong"}, + follow_redirects=False, + ) + assert resp.status_code == 200 + assert "invalid" in resp.text.lower() or "incorrect" in resp.text.lower() + + +def test_dashboard_accessible_with_session(client): + client.post( + "/login", + data={"username": "nick", "password": "secret"}, + follow_redirects=False, + ) + resp = client.get("/dashboard") + assert resp.status_code == 200 + assert "dashboard" in resp.text.lower() + + +def test_logout_clears_session(client): + client.post( + "/login", + data={"username": "nick", "password": "secret"}, + follow_redirects=False, + ) + resp = client.get("/logout", follow_redirects=False) + assert resp.status_code == 303 + assert "/login" in resp.headers["location"] + + resp2 = client.get("/dashboard", follow_redirects=False) + assert resp2.status_code == 303 +``` + +**Step 2: Run test to verify it fails** + +Run: `pytest tests/test_dashboard.py -v` +Expected: FAIL — `ntr_fetcher.dashboard` does not exist. + +**Step 3: Write minimal implementation** + +```python +# src/ntr_fetcher/dashboard.py +import hashlib +import hmac +import json +import logging +import time +from pathlib import Path + +from fastapi import APIRouter, Request, Response, HTTPException, Depends, Form +from fastapi.responses import HTMLResponse, RedirectResponse + +from ntr_fetcher.db import Database +from ntr_fetcher.websocket import AnnounceManager +from ntr_fetcher.week import get_show_week + +logger = logging.getLogger(__name__) + +STATIC_DIR = Path(__file__).parent / "static" +SESSION_MAX_AGE = 86400 # 24 hours + + +def _sign(data: str, secret: str) -> str: + sig = hmac.new(secret.encode(), data.encode(), hashlib.sha256).hexdigest() + return f"{data}.{sig}" + + +def _unsign(signed: str, secret: str, max_age: int = SESSION_MAX_AGE) -> str | None: + if "." not in signed: + return None + data, sig = signed.rsplit(".", 1) + expected = hmac.new(secret.encode(), data.encode(), hashlib.sha256).hexdigest() + if not hmac.compare_digest(sig, expected): + return None + try: + payload = json.loads(data) + if time.time() - payload.get("t", 0) > max_age: + return None + return payload.get("user") + except (json.JSONDecodeError, KeyError): + return None + + +def create_dashboard_router( + db: Database, + manager: AnnounceManager, + admin_token: str, + web_user: str, + web_password: str, + secret_key: str, + show_day: int = 2, + show_hour: int = 22, +) -> APIRouter: + router = APIRouter() + + def _get_session_user(request: Request) -> str | None: + cookie = request.cookies.get("ntr_session") + if not cookie: + return None + return _unsign(cookie, secret_key) + + def _require_session(request: Request) -> str: + user = _get_session_user(request) + if user is None: + raise HTTPException(status_code=303, headers={"Location": "/login"}) + return user + + @router.get("/login", response_class=HTMLResponse) + def login_page(request: Request, error: str = ""): + html = (STATIC_DIR / "login.html").read_text() + if error: + html = html.replace("", '
Invalid username or password
') + return HTMLResponse(html) + + @router.post("/login") + def login_submit( + username: str = Form(...), + password: str = Form(...), + ): + if username == web_user and password == web_password: + payload = json.dumps({"user": username, "t": int(time.time())}) + cookie_value = _sign(payload, secret_key) + response = RedirectResponse(url="/dashboard", status_code=303) + response.set_cookie( + "ntr_session", + cookie_value, + max_age=SESSION_MAX_AGE, + httponly=True, + samesite="lax", + ) + return response + html = (STATIC_DIR / "login.html").read_text() + html = html.replace("", 'Invalid username or password
') + return HTMLResponse(html, status_code=200) + + @router.get("/logout") + def logout(): + response = RedirectResponse(url="/login", status_code=303) + response.delete_cookie("ntr_session") + return response + + @router.get("/dashboard", response_class=HTMLResponse) + def dashboard(request: Request): + user = _get_session_user(request) + if user is None: + return RedirectResponse(url="/login", status_code=303) + html = (STATIC_DIR / "dashboard.html").read_text() + return HTMLResponse(html) + + return router +``` + +**Step 4: Create minimal static HTML stubs** + +Create `src/ntr_fetcher/static/login.html`: + +```html + +