feat(gui): add streams view (1b-6)

Add streams list and edit routes with live JetStream data:
- GET /streams: list all streams with live size/messages
- POST /streams/{name}: update max_age_s with validation

Features:
- Live data from JetStream (bytes, messages, timestamps)
- Graceful degradation when NATS unavailable
- Preset chip buttons (1d, 7d, 14d, 30d, 365d)
- Custom days input with Save button
- Current selection highlighted
- Managed by supervisor badge
- Audit logging with before/after max_age_s

Files:
- src/central/gui/audit.py: add STREAM_UPDATE constant
- src/central/gui/routes.py: add streams_list and streams_update handlers
- src/central/gui/templates/base.html: add Streams nav link
- src/central/gui/templates/streams_list.html: new template
- tests/test_streams.py: 9 tests covering all requirements

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-05-17 23:36:48 +00:00
commit ca853fbba9
5 changed files with 651 additions and 0 deletions

View file

@ -10,6 +10,7 @@ AUTH_LOGOUT = "auth.logout"
AUTH_PASSWORD_CHANGE = "auth.password_change"
OPERATOR_CREATE = "operator.create"
ADAPTER_UPDATE = "adapter.update"
STREAM_UPDATE = "stream.update"
async def write_audit(

View file

@ -22,6 +22,7 @@ from central.gui.audit import (
AUTH_LOGOUT,
AUTH_PASSWORD_CHANGE,
OPERATOR_CREATE,
STREAM_UPDATE,
write_audit,
)
from central.gui.db import get_pool
@ -872,3 +873,196 @@ async def adapters_edit_submit(
)
return RedirectResponse(url="/adapters", status_code=302)
# =============================================================================
# Streams routes
# =============================================================================
@router.get("/streams", response_class=HTMLResponse)
async def streams_list(
request: Request,
csrf_protect: CsrfProtect = Depends(),
) -> HTMLResponse:
"""List all streams with live data."""
from central.gui.nats import get_js
templates = _get_templates()
pool = get_pool()
operator = request.state.operator
js = get_js()
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT name, max_age_s, max_bytes, managed_max_bytes, updated_at
FROM config.streams
ORDER BY name
"""
)
streams = []
for row in rows:
stream_data = {
"name": row["name"],
"max_age_s": row["max_age_s"],
"max_bytes_cfg": row["max_bytes"],
"managed_max_bytes": row["managed_max_bytes"],
"live_bytes": None,
"live_messages": None,
"live_first_ts": None,
"live_last_ts": None,
"error": None,
}
# Fetch live data from JetStream
if js is not None:
try:
info = await js.stream_info(row["name"])
stream_data["live_bytes"] = info.state.bytes
stream_data["live_messages"] = info.state.messages
stream_data["live_first_ts"] = info.state.first_ts
stream_data["live_last_ts"] = info.state.last_ts
except Exception:
stream_data["error"] = "unavailable"
else:
stream_data["error"] = "NATS unavailable"
streams.append(stream_data)
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
response = templates.TemplateResponse(
request=request,
name="streams_list.html",
context={
"operator": operator,
"csrf_token": csrf_token,
"streams": streams,
},
)
csrf_protect.set_csrf_cookie(signed_token, response)
return response
@router.post("/streams/{name}", response_class=HTMLResponse)
async def streams_update(
request: Request,
name: str,
csrf_protect: CsrfProtect = Depends(),
) -> Response:
"""Update stream max_age_s."""
from central.gui.nats import get_js
templates = _get_templates()
pool = get_pool()
operator = request.state.operator
# Validate CSRF
await csrf_protect.validate_csrf(request)
form = await request.form()
max_age_s_str = form.get("max_age_s", "").strip()
errors: dict[str, str] = {}
# Parse max_age_s
try:
max_age_s = int(max_age_s_str)
except (ValueError, TypeError):
max_age_s = 0
errors[name] = "max_age_s must be a valid integer"
# Validate range: 1 hour to 5 years
MIN_AGE = 3600 # 1 hour
MAX_AGE = 5 * 365 * 24 * 3600 # 5 years (157680000)
if not errors and (max_age_s < MIN_AGE or max_age_s > MAX_AGE):
errors[name] = f"max_age_s must be between {MIN_AGE} (1 hour) and {MAX_AGE} (5 years)"
async with pool.acquire() as conn:
# Check stream exists
row = await conn.fetchrow(
"SELECT name, max_age_s FROM config.streams WHERE name = $1",
name,
)
if row is None:
return Response(status_code=404, content="Stream not found")
if errors:
# Re-render with errors
js = get_js()
rows = await conn.fetch(
"""
SELECT name, max_age_s, max_bytes, managed_max_bytes, updated_at
FROM config.streams
ORDER BY name
"""
)
streams = []
for r in rows:
stream_data = {
"name": r["name"],
"max_age_s": r["max_age_s"],
"max_bytes_cfg": r["max_bytes"],
"managed_max_bytes": r["managed_max_bytes"],
"live_bytes": None,
"live_messages": None,
"live_first_ts": None,
"live_last_ts": None,
"error": None,
}
if js is not None:
try:
info = await js.stream_info(r["name"])
stream_data["live_bytes"] = info.state.bytes
stream_data["live_messages"] = info.state.messages
stream_data["live_first_ts"] = info.state.first_ts
stream_data["live_last_ts"] = info.state.last_ts
except Exception:
stream_data["error"] = "unavailable"
else:
stream_data["error"] = "NATS unavailable"
streams.append(stream_data)
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
response = templates.TemplateResponse(
request=request,
name="streams_list.html",
context={
"operator": operator,
"csrf_token": csrf_token,
"streams": streams,
"errors": errors,
},
)
csrf_protect.set_csrf_cookie(signed_token, response)
return response
old_max_age_s = row["max_age_s"]
# Update stream
await conn.execute(
"""
UPDATE config.streams
SET max_age_s = $1, updated_at = now()
WHERE name = $2
""",
max_age_s,
name,
)
# Write audit log
await write_audit(
conn,
STREAM_UPDATE,
operator_id=operator.id,
target=name,
before={"max_age_s": old_max_age_s},
after={"max_age_s": max_age_s},
)
return RedirectResponse(url="/streams", status_code=302)

View file

@ -17,6 +17,7 @@
{% if operator %}
<li><a href="/">Dashboard</a></li>
<li><a href="/adapters">Adapters</a></li>
<li><a href="/streams">Streams</a></li>
<li>{{ operator.username }}</li>
<li><a href="/change-password">Change Password</a></li>
<li>

View file

@ -0,0 +1,71 @@
{% extends "base.html" %}
{% block title %}Central — Streams{% endblock %}
{% block content %}
<h1>Streams</h1>
<div class="grid">
{% for stream in streams %}
<article>
<header>
<strong>{{ stream.name }}</strong>
{% if stream.managed_max_bytes %}
<span style="float: right; font-size: 0.8em; background: var(--pico-secondary-background); padding: 0.2em 0.5em; border-radius: 4px;">Managed by supervisor</span>
{% endif %}
</header>
<div style="margin-bottom: 1rem;">
<strong>Live Data:</strong>
{% if stream.error %}
<span style="color: var(--pico-color-red-500);">({{ stream.error }})</span>
{% else %}
<ul style="margin: 0.5rem 0;">
<li>Size: {{ stream.live_bytes|default(0)|filesizeformat if stream.live_bytes is not none else '(unavailable)' }}</li>
<li>Messages: {{ stream.live_messages if stream.live_messages is not none else '(unavailable)' }}</li>
<li>First message: {{ stream.live_first_ts.isoformat() if stream.live_first_ts else '(none)' }}</li>
<li>Last message: {{ stream.live_last_ts.isoformat() if stream.live_last_ts else '(none)' }}</li>
</ul>
{% endif %}
</div>
<div style="margin-bottom: 1rem;">
<strong>Configuration:</strong>
<ul style="margin: 0.5rem 0;">
<li>Max age: {{ (stream.max_age_s / 86400)|round(1) }} days ({{ stream.max_age_s }}s)</li>
<li>Max bytes (config): {{ stream.max_bytes_cfg|filesizeformat }}</li>
</ul>
</div>
{% if errors and errors[stream.name] %}
<p style="color: var(--pico-color-red-500);">{{ errors[stream.name] }}</p>
{% endif %}
<div>
<strong>Set Retention:</strong>
<div style="display: flex; flex-wrap: wrap; gap: 0.5rem; margin-top: 0.5rem;">
{% set presets = [(1, '1 day'), (7, '7 days'), (14, '14 days'), (30, '30 days'), (365, '365 days')] %}
{% for days, label in presets %}
{% set preset_seconds = days * 86400 %}
<form action="/streams/{{ stream.name }}" method="post" style="margin: 0;">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<input type="hidden" name="max_age_s" value="{{ preset_seconds }}">
<button type="submit" class="{% if stream.max_age_s == preset_seconds %}contrast{% else %}outline{% endif %}" style="padding: 0.3em 0.6em; font-size: 0.9em;">
{{ label }}
</button>
</form>
{% endfor %}
</div>
<form action="/streams/{{ stream.name }}" method="post" style="margin-top: 0.5rem; display: flex; gap: 0.5rem; align-items: center;">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label for="custom_days_{{ stream.name }}" style="margin: 0;">Custom:</label>
<input type="number" id="custom_days_{{ stream.name }}" name="custom_days" min="1" max="1825" placeholder="days" style="width: 100px;" onchange="this.form.max_age_s.value = this.value * 86400;">
<input type="hidden" name="max_age_s" value="">
<button type="submit" class="outline" style="padding: 0.3em 0.6em;">Save</button>
</form>
</div>
</article>
{% endfor %}
</div>
{% endblock %}