NtR-soudcloud-fetcher/docs/plans/2026-03-12-live-announce-dashboard-impl.md

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

# 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:

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

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:

dependencies = [
    "fastapi",
    "uvicorn[standard]",
    "httpx",
    "pydantic-settings",
    "itsdangerous",
]

Step 2: Install

Run: pip install -e ".[dev]" Expected: installs itsdangerous

Step 3: Commit

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

# 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

# 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

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

# 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

# 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("<!--ERROR-->", '<p class="error">Invalid username or password</p>')
        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("<!--ERROR-->", '<p class="error">Invalid username or password</p>')
        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:

<!DOCTYPE html>
<html><head><title>Login</title></head>
<body>
<!--ERROR-->
<form method="post" action="/login">
<label>Username <input name="username"></label>
<label>Password <input name="password" type="password"></label>
<button type="submit">Login</button>
</form>
</body></html>

Create src/ntr_fetcher/static/dashboard.html:

<!DOCTYPE html>
<html><head><title>Dashboard</title></head>
<body><h1>NtR Playlist Dashboard</h1></body></html>

Step 5: Run test to verify it passes

Run: pytest tests/test_dashboard.py -v Expected: PASS

Step 6: Commit

git add src/ntr_fetcher/dashboard.py src/ntr_fetcher/static/ tests/test_dashboard.py
git commit -m "feat: add dashboard router with session auth"

---

Task 5: Announce endpoint

Files:

• Modify: src/ntr_fetcher/dashboard.py
• Test: tests/test_dashboard.py (append)

Step 1: Write the failing test

Append to tests/test_dashboard.py:

from ntr_fetcher.models import Track


def _seed_show(db):
    from datetime import datetime, timezone
    week_start = datetime(2026, 3, 12, 2, 0, 0, tzinfo=timezone.utc)
    week_end = datetime(2026, 3, 19, 2, 0, 0, tzinfo=timezone.utc)
    show = db.get_or_create_show(week_start, week_end)
    t1 = Track(1, "Song A", "Artist A", "https://soundcloud.com/a/1", None, 180000, "cc-by",
               datetime(2026, 3, 14, 1, 0, 0, tzinfo=timezone.utc), "{}")
    db.upsert_track(t1)
    db.set_show_tracks(show.id, [t1.id])
    return show


def test_announce_with_session(client, db):
    _seed_show(db)
    client.post(
        "/login",
        data={"username": "nick", "password": "secret"},
        follow_redirects=False,
    )
    from unittest.mock import patch
    with patch("ntr_fetcher.dashboard.datetime") as mock_dt:
        mock_dt.now.return_value = datetime(2026, 3, 14, 12, 0, 0, tzinfo=timezone.utc)
        resp = client.post("/admin/announce", json={"show_id": 1, "position": 1})
    assert resp.status_code == 200
    data = resp.json()
    assert data["status"] == "announced"
    assert "Now Playing:" in data["message"]
    assert "Song A" in data["message"]
    assert "Artist A" in data["message"]


def test_announce_with_bearer(client, db):
    _seed_show(db)
    from unittest.mock import patch
    with patch("ntr_fetcher.dashboard.datetime") as mock_dt:
        mock_dt.now.return_value = datetime(2026, 3, 14, 12, 0, 0, tzinfo=timezone.utc)
        resp = client.post(
            "/admin/announce",
            json={"show_id": 1, "position": 1},
            headers={"Authorization": "Bearer test-token"},
        )
    assert resp.status_code == 200
    assert "Now Playing:" in resp.json()["message"]


def test_announce_without_auth(client, db):
    _seed_show(db)
    resp = client.post("/admin/announce", json={"show_id": 1, "position": 1})
    assert resp.status_code == 401

Step 2: Run test to verify it fails

Run: pytest tests/test_dashboard.py::test_announce_with_session -v Expected: FAIL — no /admin/announce route.

Step 3: Write minimal implementation

Add to src/ntr_fetcher/dashboard.py, inside create_dashboard_router, before return router:

from datetime import datetime, timezone
from pydantic import BaseModel

class AnnounceRequest(BaseModel):
    show_id: int
    position: int

@router.post("/admin/announce")
async def announce(body: AnnounceRequest, request: Request):
    # Accept session cookie OR bearer token
    user = _get_session_user(request)
    if user is None:
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer ") or auth_header.removeprefix("Bearer ") != admin_token:
            raise HTTPException(status_code=401, detail="Unauthorized")

    track = db.get_show_track_by_position(body.show_id, body.position)
    if track is None:
        raise HTTPException(status_code=404, detail=f"No track at position {body.position}")

    message = (
        f"Now Playing: Song #{track['position']}: "
        f"{track['title']} by {track['artist']} - {track['permalink_url']}"
    )
    await manager.broadcast({"type": "announce", "message": message})
    return {"status": "announced", "message": message}

Note: the AnnounceRequest model and datetime import should go at module level, not inside the function. Place the AnnounceRequest class near the top of the file. The datetime import is already present from the session code.

Step 4: Run test to verify it passes

Run: pytest tests/test_dashboard.py -v Expected: PASS

Step 5: Commit

git add src/ntr_fetcher/dashboard.py tests/test_dashboard.py
git commit -m "feat: add announce endpoint with dual auth"

---

Task 6: WebSocket handler on the API

Files:

• Modify: src/ntr_fetcher/dashboard.py
• Test: tests/test_dashboard.py (append)

Step 1: Write the failing test

Append to tests/test_dashboard.py:

def test_ws_subscribe_with_valid_token(app):
    with TestClient(app) as c:
        with c.websocket_connect("/ws/announce") as ws:
            ws.send_json({"type": "subscribe", "token": "test-token"})
            data = ws.receive_json()
            assert data["type"] == "status"


def test_ws_subscribe_with_invalid_token(app):
    with TestClient(app) as c:
        with c.websocket_connect("/ws/announce") as ws:
            ws.send_json({"type": "subscribe", "token": "wrong"})
            # Server should close the connection
            import websockets
            with pytest.raises(Exception):
                ws.receive_json()

Step 2: Run test to verify it fails

Run: pytest tests/test_dashboard.py::test_ws_subscribe_with_valid_token -v Expected: FAIL — no /ws/announce route.

Step 3: Write minimal implementation

Add to src/ntr_fetcher/dashboard.py, inside create_dashboard_router:

from fastapi import WebSocket, WebSocketDisconnect

@router.websocket("/ws/announce")
async def ws_announce(websocket: WebSocket):
    await websocket.accept()
    try:
        auth_msg = await websocket.receive_json()
        if auth_msg.get("type") != "subscribe" or auth_msg.get("token") != admin_token:
            await websocket.close(code=4001, reason="Invalid token")
            return

        manager.add_subscriber(websocket)
        await manager.broadcast_status()
        try:
            while True:
                await websocket.receive_text()
        except WebSocketDisconnect:
            pass
    except WebSocketDisconnect:
        pass
    finally:
        manager.remove_subscriber(websocket)
        try:
            await manager.broadcast_status()
        except Exception:
            pass

Step 4: Run test to verify it passes

Run: pytest tests/test_dashboard.py -v Expected: PASS (the invalid-token test may need adjustment — FastAPI test client may raise WebSocketDisconnect or similar; adjust assertion as needed during implementation).

Step 5: Commit

git add src/ntr_fetcher/dashboard.py tests/test_dashboard.py
git commit -m "feat: add WebSocket handler for announce subscribers"

---

Task 7: Wire dashboard into the app

Files:

• Modify: src/ntr_fetcher/api.py
• Modify: src/ntr_fetcher/main.py
• Test: tests/test_api.py (add one test to verify existing behavior unbroken)

Step 1: Write the failing test

Add to tests/test_api.py:

def test_no_dashboard_routes_without_config(client):
    resp = client.get("/dashboard", follow_redirects=False)
    assert resp.status_code == 404


def test_no_login_route_without_config(client):
    resp = client.get("/login")
    assert resp.status_code == 404

Step 2: Run test to verify it fails (or passes if routes aren't mounted yet)

Run: pytest tests/test_api.py::test_no_dashboard_routes_without_config -v Expected: PASS (routes don't exist yet — this is a guard test).

Step 3: Modify create_app signature and mounting

In src/ntr_fetcher/api.py, update create_app to accept optional dashboard params and conditionally mount:

def create_app(
    db: Database,
    poller,
    admin_token: str,
    show_day: int = 2,
    show_hour: int = 22,
    web_user: str | None = None,
    web_password: str | None = None,
    secret_key: str | None = None,
) -> FastAPI:
    app = FastAPI(title="NtR SoundCloud Fetcher")

    # ... existing routes unchanged ...

    if all([web_user, web_password, secret_key]):
        from ntr_fetcher.dashboard import create_dashboard_router
        from ntr_fetcher.websocket import AnnounceManager
        manager = AnnounceManager()
        dashboard_router = create_dashboard_router(
            db=db,
            manager=manager,
            admin_token=admin_token,
            web_user=web_user,
            web_password=web_password,
            secret_key=secret_key,
            show_day=show_day,
            show_hour=show_hour,
        )
        app.include_router(dashboard_router)

    return app

In src/ntr_fetcher/main.py, pass the new params:

app = create_app(
    db=db,
    poller=poller,
    admin_token=settings.admin_token,
    show_day=settings.show_day,
    show_hour=settings.show_hour,
    web_user=settings.web_user,
    web_password=settings.web_password,
    secret_key=settings.secret_key,
)

Step 4: Run full test suite

Run: pytest tests/ -v Expected: ALL PASS — existing tests unbroken, dashboard routes absent without config.

Step 5: Commit

git add src/ntr_fetcher/api.py src/ntr_fetcher/main.py tests/test_api.py
git commit -m "feat: conditionally mount dashboard when config present"

---

Task 8: Styled login page

Files:

• Modify: src/ntr_fetcher/static/login.html

No test needed — this is a visual/HTML-only change. The functional login tests from Task 4 already cover the form behavior.

Step 1: Write the styled login page

Replace src/ntr_fetcher/static/login.html with a Pico CSS dark-themed login form. Key elements:

• <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css">
• <html data-theme="dark">
• Centered card with "NtR Playlist" heading, username/password fields, submit button
• <!--ERROR--> placeholder for error message injection
• Minimal custom CSS for the error message color

<!DOCTYPE html>
<html lang="en" data-theme="dark">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>NtR Login</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css">
    <style>
        body { display: flex; align-items: center; justify-content: center; min-height: 100vh; }
        main { max-width: 400px; width: 100%; }
        .error { color: var(--pico-color-red-500, #e53935); margin-bottom: 1rem; }
    </style>
</head>
<body>
    <main>
        <article>
            <header><h2>NtR Playlist</h2></header>
            <!--ERROR-->
            <form method="post" action="/login">
                <label>Username
                    <input name="username" autocomplete="username" required>
                </label>
                <label>Password
                    <input name="password" type="password" autocomplete="current-password" required>
                </label>
                <button type="submit">Log in</button>
            </form>
        </article>
    </main>
</body>
</html>

Step 2: Run existing tests to verify nothing broke

Run: pytest tests/test_dashboard.py -v Expected: PASS

Step 3: Commit

git add src/ntr_fetcher/static/login.html
git commit -m "feat: styled login page with Pico CSS dark theme"

---

Task 9: Styled dashboard page

Files:

• Modify: src/ntr_fetcher/static/dashboard.html

This is the main dashboard with playlist tables and announce buttons. The page uses vanilla JS to:

1. Fetch playlists from /playlist and /shows?limit=2 + /shows/{id} on load
2. Open a read-only WebSocket for connection status
3. POST to /admin/announce when an "Announce" button is clicked

Step 1: Write the dashboard HTML

Replace src/ntr_fetcher/static/dashboard.html. Key elements:

• Pico CSS dark theme
• Header with title, subscriber status dot, logout link
• "Current Show" section: episode heading, track table, announce buttons
• "Previous Show" section: <details> element (collapsed by default), same table layout
• JS: fetch('/playlist') for current show data, fetch('/shows?limit=2') then fetch('/shows/{id}') for previous show
• JS: new WebSocket(...) connecting to /ws/announce with admin token from a <meta> tag or fetched from a session-protected endpoint. Simpler approach: the dashboard HTML is served by the router, which can inject the admin token into a <script> block as a JS variable. Alternative (safer): add a small /dashboard/ws-token endpoint that returns the admin token if the user has a valid session. This avoids putting the token in the HTML source.

Actually, re-reading the design: the WS endpoint authenticates via the admin token. The dashboard needs to know this token to connect. Two options:

• The server injects it into the HTML as a template variable
• A small API endpoint returns it to authenticated sessions

Use the template-injection approach since we're already reading the HTML file and can do a simple string replacement. The token is only visible to authenticated users (the /dashboard route requires a session).

<!DOCTYPE html>
<html lang="en" data-theme="dark">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>NtR Playlist Dashboard</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css">
    <style>
        .status-dot { display: inline-block; width: 10px; height: 10px; border-radius: 50%; margin-right: 6px; }
        .status-dot.connected { background: #4caf50; }
        .status-dot.disconnected { background: #888; }
        header nav { display: flex; align-items: center; justify-content: space-between; }
        .announce-btn { padding: 4px 12px; font-size: 0.85rem; }
        .announce-btn.success { background: #4caf50; }
        .announce-btn.warning { background: #ff9800; }
        table { width: 100%; }
        .toast { position: fixed; bottom: 20px; right: 20px; padding: 12px 20px; border-radius: 8px;
                 background: #333; color: #fff; display: none; z-index: 100; }
        .toast.error { background: #e53935; }
    </style>
</head>
<body>
    <main class="container">
        <nav>
            <div>
                <span class="status-dot disconnected" id="status-dot"></span>
                <strong>NtR Playlist Dashboard</strong>
                <small id="sub-count">(connecting...)</small>
            </div>
            <a href="/logout">Logout</a>
        </nav>
        <section id="current-show"><h3>Loading current show...</h3></section>
        <details id="prev-show-details">
            <summary><h3 style="display:inline">Previous Show</h3></summary>
            <section id="prev-show"><p>Loading...</p></section>
        </details>
        <div class="toast" id="toast"></div>
    </main>
    <script>
        const WS_TOKEN = "{{WS_TOKEN}}";
        // JS implementation: fetch playlists, render tables, handle announce clicks, WS status
        // (full JS provided below)
    </script>
</body>
</html>

The full JS in the <script> block handles:

• loadCurrentShow() — fetches /playlist, renders track table with announce buttons
• loadPreviousShow() — fetches /shows?limit=2, if 2 results fetches /shows/{prev_id}, renders table
• announce(showId, position, btn) — POSTs to /admin/announce, handles button state
• connectWS() — connects to ws://{host}/ws/announce, sends subscribe, listens for status updates, auto-reconnects
• showToast(msg, isError) — brief toast notification

Update dashboard.py to inject the token:

@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()
    html = html.replace("{{WS_TOKEN}}", admin_token)
    return HTMLResponse(html)

Step 2: Run tests

Run: pytest tests/ -v Expected: ALL PASS

Step 3: Commit

git add src/ntr_fetcher/static/dashboard.html src/ntr_fetcher/dashboard.py
git commit -m "feat: styled dashboard with playlist tables and announce buttons"

---

Task 10: Limnoria plugin — WS client + announce config

Files:

• Modify: plugins/limnoria/NtrPlaylist/plugin.py
• Modify: plugins/limnoria/NtrPlaylist/config.py

No automated test (Limnoria's test framework is separate and requires a running bot instance). Manual testing instructions provided.

Step 1: Add config values

In plugins/limnoria/NtrPlaylist/config.py, add two new registry values:

conf.registerGlobalValue(
    NtrPlaylist,
    "wsUrl",
    registry.String(
        "ws://127.0.0.1:8000/ws/announce",
        """WebSocket URL for receiving announce commands from the dashboard.""",
    ),
)

conf.registerGlobalValue(
    NtrPlaylist,
    "announceChannel",
    registry.String(
        "#sewerchat",
        """IRC channel to send announce messages to.""",
    ),
)

Step 2: Add WS client thread to plugin

In plugins/limnoria/NtrPlaylist/plugin.py:

1. Add import json, threading, time to existing imports.
2. Add a _ws_listener method to the NtrPlaylist class that runs in a daemon thread.
3. Start the thread in __init__ and stop it in die().

def __init__(self, irc):
    super().__init__(irc)
    self._irc = irc
    self._ws_stop = threading.Event()
    self._ws_thread = threading.Thread(target=self._ws_listener, daemon=True)
    self._ws_thread.start()

def die(self):
    self._ws_stop.set()
    super().die()

def _ws_listener(self):
    import websocket  # websocket-client library
    from supybot import ircmsgs

    backoff = 5
    max_backoff = 60

    while not self._ws_stop.is_set():
        ws_url = self.registryValue("wsUrl")
        token = self.registryValue("adminToken")
        channel = self.registryValue("announceChannel")

        if not ws_url or not token:
            LOGGER.warning("wsUrl or adminToken not configured, WS listener sleeping")
            self._ws_stop.wait(30)
            continue

        try:
            ws = websocket.WebSocket()
            ws.connect(ws_url, timeout=10)
            ws.send(json.dumps({"type": "subscribe", "token": token}))
            LOGGER.info("Connected to announce WebSocket at %s", ws_url)
            backoff = 5

            while not self._ws_stop.is_set():
                ws.settimeout(5)
                try:
                    raw = ws.recv()
                    if not raw:
                        break
                    data = json.loads(raw)
                    if data.get("type") == "announce" and "message" in data:
                        msg = ircmsgs.privmsg(channel, data["message"])
                        self._irc.queueMsg(msg)
                        LOGGER.info("Announced to %s: %s", channel, data["message"])
                except websocket.WebSocketTimeoutException:
                    continue
                except websocket.WebSocketConnectionClosedException:
                    break
        except Exception:
            LOGGER.exception("WS listener error")
        finally:
            try:
                ws.close()
            except Exception:
                pass

        if not self._ws_stop.is_set():
            LOGGER.info("Reconnecting in %ds", backoff)
            self._ws_stop.wait(backoff)
            backoff = min(backoff * 2, max_backoff)

Step 3: Commit

git add plugins/limnoria/NtrPlaylist/plugin.py plugins/limnoria/NtrPlaylist/config.py
git commit -m "feat: add WS announce listener to Limnoria plugin"

---

Task 11: Sopel plugin — WS client + announce config

Files:

• Modify: plugins/sopel/ntr_playlist.py

Feature-parity with the Limnoria plugin.

Step 1: Add config values

Add to NtrPlaylistSection:

ws_url = types.ValidatedAttribute("ws_url", default="ws://127.0.0.1:8000/ws/announce")
announce_channel = types.ValidatedAttribute("announce_channel", default="#sewerchat")

Step 2: Add WS client thread

Add a _ws_listener function and start/stop lifecycle via setup and shutdown hooks:

_ws_stop = None
_ws_thread = None


def setup(bot):
    global _ws_stop, _ws_thread
    bot.settings.define_section("ntr_playlist", NtrPlaylistSection)
    _ws_stop = threading.Event()
    _ws_thread = threading.Thread(target=_ws_listener, args=(bot,), daemon=True)
    _ws_thread.start()


def shutdown(bot):
    global _ws_stop
    if _ws_stop:
        _ws_stop.set()


def _ws_listener(bot):
    import websocket

    backoff = 5
    max_backoff = 60

    while not _ws_stop.is_set():
        ws_url = bot.settings.ntr_playlist.ws_url
        token = bot.settings.ntr_playlist.admin_token
        channel = bot.settings.ntr_playlist.announce_channel

        if not ws_url or not token:
            LOGGER.warning("ws_url or admin_token not configured, WS listener sleeping")
            _ws_stop.wait(30)
            continue

        try:
            ws = websocket.WebSocket()
            ws.connect(ws_url, timeout=10)
            ws.send(json.dumps({"type": "subscribe", "token": token}))
            LOGGER.info("Connected to announce WebSocket at %s", ws_url)
            backoff = 5

            while not _ws_stop.is_set():
                ws.settimeout(5)
                try:
                    raw = ws.recv()
                    if not raw:
                        break
                    data = json.loads(raw)
                    if data.get("type") == "announce" and "message" in data:
                        bot.say(data["message"], channel)
                        LOGGER.info("Announced to %s: %s", channel, data["message"])
                except websocket.WebSocketTimeoutException:
                    continue
                except websocket.WebSocketConnectionClosedException:
                    break
        except Exception:
            LOGGER.exception("WS listener error")
        finally:
            try:
                ws.close()
            except Exception:
                pass

        if not _ws_stop.is_set():
            LOGGER.info("Reconnecting in %ds", backoff)
            _ws_stop.wait(backoff)
            backoff = min(backoff * 2, max_backoff)

Add import threading to the module imports.

Step 3: Commit

git add plugins/sopel/ntr_playlist.py
git commit -m "feat: add WS announce listener to Sopel plugin"

---

Task 12: Update docs

Files:

• Modify: docs/api.md
• Modify: README.md

Step 1: Update docs/api.md

Add new sections for:

• GET /login, POST /login, GET /logout, GET /dashboard
• POST /admin/announce — request/response shapes
• WS /ws/announce — protocol description (subscribe message, announce message, status message)
• Dashboard configuration table

Step 2: Update README.md

Add a "Dashboard" section with:

• The three env vars needed
• Brief description of the feature
• New endpoint table rows

Step 3: Commit

git add docs/api.md README.md
git commit -m "docs: add dashboard and announce API documentation"

---

Task 13: Run full test suite and lint

Step 1: Run all tests

Run: pytest tests/ -v Expected: ALL PASS

Step 2: Run linter

Run: ruff check src/ tests/ Expected: no errors (fix any that appear)

Step 3: Final commit (if any lint fixes)

git add -A
git commit -m "chore: lint fixes"