spencermandClaude Fable 5.1
6b6db0230f
The Project Management screen (/pm, view only) with Overview, Projects & units, Bottlenecks & aging, Hours vs target, Quality & issues and Find jobs, the period picker, health per unit and per project, drill down, Excel (CSV) per table and for all, and the PESCO production status report PDF with the controlled-copy stamp; every export logged as "report exported". The find bar on every home screen: the full search set, the dropdowns, quick flags with Mine, the flag kinds, Board and List with sorts, filters saved per browser; every filter applied by the server on the list and the boards, with facets. Presence in packets with role, lock and revoke drop; the motion switch with reduced motion honoured, card glides, glow and count-up; Back restores the scroll spot; sheet placement and the iPad layout pass. The data book compared page by page with the old build and six layout differences fixed. Tests for the PM aggregates, search, filters, presence and exports; Playwright scenarios for the PM screen, the find bar and presence at desktop and iPad viewports. Docs updated; PLAN.md questions 38 to 42. Co-Authored-By: Claude Fable 5.1 <[email protected]>
267 lines
9.7 KiB
Python
267 lines
9.7 KiB
Python
"""Server-sent events: one LISTEN connection per process, fanned out to every subscriber queue, plus presence.
|
|
|
|
A thread holds a sync psycopg connection (app role) doing LISTEN ps_events and pushes each NOTIFY payload
|
|
to the event loop with call_soon_threadsafe. It reconnects with backoff when the database drops. Presence is
|
|
kept in process (one uvicorn worker): one entry per (user, screen, packet) with the display name and the role
|
|
label. A locked session drops out of presence until it is unlocked; a revoked one drops out at once. A browser
|
|
that goes away drops out when its stream ends (the stream's `finally`, at the latest on the next ping).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import threading
|
|
import uuid
|
|
from collections.abc import AsyncIterator, Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CHANNEL = "ps_events"
|
|
PING_SECONDS = 25.0
|
|
QUEUE_MAX = 1000
|
|
BACKOFF_MAX = 30.0
|
|
|
|
Item = tuple[str, dict[str, Any]]
|
|
|
|
|
|
def publish(conn: Any, payload: dict[str, Any]) -> None:
|
|
"""NOTIFY ps_events with a JSON payload. Inside a transaction it is delivered on commit.
|
|
|
|
`payload["event"]` names the SSE event (default `changed`); the rest is the event data.
|
|
"""
|
|
conn.execute("SELECT ps_notify(%s, %s)", (CHANNEL, json.dumps(payload, default=str, separators=(",", ":"))))
|
|
|
|
|
|
@dataclass(eq=False)
|
|
class Subscriber:
|
|
queue: asyncio.Queue[Item]
|
|
loop: asyncio.AbstractEventLoop
|
|
session_id: str
|
|
user_id: str
|
|
display_name: str
|
|
screen: str | None = None
|
|
packet_id: int | None = None # the packet the subscriber has open, for "Also here" on the packet page
|
|
role: str = "" # the label shown beside the name: the screen open, else the person's primary role
|
|
since: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
locked: bool = False # the session is locked: left out of presence until it is unlocked
|
|
revoked: bool = False # the session was revoked: left out of presence while the stream winds down
|
|
id: str = field(default_factory=lambda: uuid.uuid4().hex)
|
|
closed: bool = False
|
|
|
|
@property
|
|
def present(self) -> bool:
|
|
return not (self.closed or self.locked or self.revoked)
|
|
|
|
def put(self, item: Item) -> None:
|
|
"""Called on the event loop."""
|
|
try:
|
|
self.queue.put_nowait(item)
|
|
except asyncio.QueueFull:
|
|
log.warning("event queue full for a subscriber; dropping one event")
|
|
|
|
|
|
class EventHub:
|
|
def __init__(self, connect: Callable[[], Any] | None = None):
|
|
self._connect = connect
|
|
self._subs: dict[str, Subscriber] = {}
|
|
self._lock = threading.Lock()
|
|
self._stop = threading.Event()
|
|
self._thread: threading.Thread | None = None
|
|
self.connected = False
|
|
|
|
# ------------------------------------------------ subscribers
|
|
|
|
def subscribe(
|
|
self,
|
|
*,
|
|
session_id: Any,
|
|
user_id: Any,
|
|
display_name: str,
|
|
screen: str | None = None,
|
|
packet_id: int | None = None,
|
|
role: str = "",
|
|
locked: bool = False,
|
|
) -> Subscriber:
|
|
"""Register a subscriber on the running loop and broadcast presence."""
|
|
sub = Subscriber(
|
|
queue=asyncio.Queue(maxsize=QUEUE_MAX),
|
|
loop=asyncio.get_running_loop(),
|
|
session_id=str(session_id),
|
|
user_id=str(user_id),
|
|
display_name=display_name,
|
|
screen=screen,
|
|
packet_id=packet_id,
|
|
role=role,
|
|
locked=locked,
|
|
)
|
|
with self._lock:
|
|
self._subs[sub.id] = sub
|
|
self._broadcast_presence()
|
|
return sub
|
|
|
|
def unsubscribe(self, sub: Subscriber) -> None:
|
|
sub.closed = True
|
|
with self._lock:
|
|
removed = self._subs.pop(sub.id, None)
|
|
if removed is not None:
|
|
self._broadcast_presence()
|
|
|
|
def _users(self) -> list[dict[str, Any]]:
|
|
"""One entry per (user, screen, packet) among the present subscribers, the earliest `since` kept."""
|
|
with self._lock:
|
|
subs = [s for s in self._subs.values() if s.present]
|
|
seen: dict[tuple[str, str | None, int | None], dict[str, Any]] = {}
|
|
for s in sorted(subs, key=lambda x: x.since):
|
|
seen.setdefault(
|
|
(s.user_id, s.screen, s.packet_id),
|
|
{
|
|
"user_id": s.user_id,
|
|
"display_name": s.display_name,
|
|
"role": s.role,
|
|
"screen": s.screen,
|
|
"packet_id": s.packet_id,
|
|
"since": s.since.isoformat(),
|
|
},
|
|
)
|
|
return sorted(seen.values(), key=lambda u: (u["display_name"].lower(), u["screen"] or "", u["packet_id"] or 0))
|
|
|
|
def presence(self) -> dict[str, Any]:
|
|
"""{users: [{user_id, display_name, role, screen, packet_id, since}]}, locked and revoked sessions left
|
|
out."""
|
|
return {"users": self._users()}
|
|
|
|
def presence_by_screen(self) -> dict[str, list[dict[str, Any]]]:
|
|
"""The same users grouped by the screen they have open ("none" for a stream without a screen), for the
|
|
Admin "who's in" view."""
|
|
out: dict[str, list[dict[str, Any]]] = {}
|
|
for u in self._users():
|
|
out.setdefault(u["screen"] or "none", []).append(u)
|
|
return out
|
|
|
|
def _send(self, sub: Subscriber, item: Item) -> None:
|
|
try:
|
|
sub.loop.call_soon_threadsafe(sub.put, item)
|
|
except RuntimeError: # loop closed
|
|
pass
|
|
|
|
def _broadcast_presence(self) -> None:
|
|
item: Item = ("presence", self.presence())
|
|
with self._lock:
|
|
subs = list(self._subs.values())
|
|
for s in subs:
|
|
self._send(s, item)
|
|
|
|
def _deliver(self, payload: str) -> None:
|
|
"""Fan one NOTIFY payload out. `session` events go only to the subscribers of that session."""
|
|
try:
|
|
data = json.loads(payload)
|
|
except ValueError:
|
|
log.warning("ignoring a non-JSON ps_events payload")
|
|
return
|
|
if not isinstance(data, dict):
|
|
return
|
|
name = str(data.pop("event", "changed"))
|
|
with self._lock:
|
|
subs = list(self._subs.values())
|
|
changed = False
|
|
if name == "session":
|
|
target = str(data.get("session_id"))
|
|
subs = [s for s in subs if s.session_id == target]
|
|
changed = self._mark(subs, str(data.get("action") or ""))
|
|
for s in subs:
|
|
self._send(s, (name, data))
|
|
if changed:
|
|
self._broadcast_presence()
|
|
|
|
def _mark(self, subs: list[Subscriber], action: str) -> bool:
|
|
"""Apply a session event to that session's subscribers. True when presence changed."""
|
|
changed = False
|
|
with self._lock:
|
|
for s in subs:
|
|
before = s.present
|
|
if action == "locked":
|
|
s.locked = True
|
|
elif action == "unlocked":
|
|
s.locked = False
|
|
elif action == "revoked":
|
|
s.revoked = True
|
|
changed = changed or before != s.present
|
|
return changed
|
|
|
|
# ------------------------------------------------ listener thread
|
|
|
|
def start(self) -> None:
|
|
if self._connect is None or (self._thread and self._thread.is_alive()):
|
|
return
|
|
self._stop.clear()
|
|
self._thread = threading.Thread(target=self._run, name="ps-events", daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self, timeout: float = 5.0) -> None:
|
|
self._stop.set()
|
|
if self._thread is not None:
|
|
self._thread.join(timeout)
|
|
self._thread = None
|
|
|
|
def _run(self) -> None:
|
|
assert self._connect is not None
|
|
delay = 1.0
|
|
while not self._stop.is_set():
|
|
conn = None
|
|
try:
|
|
conn = self._connect()
|
|
conn.execute(f"LISTEN {CHANNEL}")
|
|
self.connected = True
|
|
delay = 1.0
|
|
log.info("events hub listening on %s", CHANNEL)
|
|
while not self._stop.is_set():
|
|
for n in conn.notifies(timeout=1.0):
|
|
self._deliver(n.payload)
|
|
except Exception as e:
|
|
if not self._stop.is_set():
|
|
log.warning("events hub: %s; retrying in %.0f s", type(e).__name__, delay)
|
|
finally:
|
|
self.connected = False
|
|
if conn is not None:
|
|
with contextlib.suppress(Exception):
|
|
conn.close()
|
|
if self._stop.wait(delay):
|
|
break
|
|
delay = min(delay * 2, BACKOFF_MAX)
|
|
|
|
|
|
def format_sse(event: str, data: dict[str, Any]) -> str:
|
|
return f"event: {event}\ndata: {json.dumps(data, default=str, separators=(',', ':'))}\n\n"
|
|
|
|
|
|
async def stream(
|
|
hub: EventHub,
|
|
sub: Subscriber,
|
|
*,
|
|
hello: dict[str, Any],
|
|
is_disconnected: Callable[[], Awaitable[bool]],
|
|
ping_seconds: float = PING_SECONDS,
|
|
) -> AsyncIterator[str]:
|
|
"""The SSE body: `hello`, then queued events, with a `: ping` comment when idle. Ends on own revoke."""
|
|
try:
|
|
yield format_sse("hello", hello)
|
|
while True:
|
|
if await is_disconnected():
|
|
break
|
|
try:
|
|
name, data = await asyncio.wait_for(sub.queue.get(), timeout=ping_seconds)
|
|
except TimeoutError:
|
|
yield ": ping\n\n"
|
|
continue
|
|
yield format_sse(name, data)
|
|
if name == "session" and data.get("session_id") == sub.session_id and data.get("action") == "revoked":
|
|
break
|
|
finally:
|
|
hub.unsubscribe(sub)
|