"""Debug / observability endpoints — view recent activity in a browser."""
from collections import deque
from datetime import datetime, timezone
from typing import Any, Deque, Dict
from fastapi import APIRouter
from fastapi.responses import HTMLResponse
router = APIRouter(prefix="/debug", tags=["debug"])
# In-memory ring buffer of the last N received tank-allocation commands.
_MAX = 100
_recent: Deque[Dict[str, Any]] = deque(maxlen=_MAX)
def record_allocation(payload: Dict[str, Any], event_id: str) -> None:
_recent.appendleft({
"received_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"event_id": event_id,
"payload": payload,
})
@router.get("/allocations.json")
async def allocations_json() -> dict:
return {"count": len(_recent), "items": list(_recent)}
@router.get("/allocations", response_class=HTMLResponse)
async def allocations_page() -> str:
rows = ""
for item in _recent:
p = item["payload"]
meta = p.get("metadata") or {}
rows += f"""
| {item['received_at']} |
{p.get('job_id','')} |
{p.get('licence_number','')} |
{p.get('tank_id','')} |
{meta.get('j_cust_code','')} |
{'⚠️' if p.get('is_dangerous_goods') else ''} |
{item['event_id']} |
"""
if not rows:
rows = '| No allocations received yet |
'
return f"""
BTL FastLane – Recent Allocations
| Received (UTC) |
Job ID |
Licence |
Tank ID |
Customer |
DG |
Event ID |
{rows}
"""