fix: stop a stream on close, count only real users, record where a node is

**Closing the viewer left the node working.** Nothing told it to stop:
the player dropped its handlers, which only made the browser deaf. ffmpeg
kept running and held one of the node's two transcode slots until the
credit timeout expired two minutes later — which is why the next video
answered "server busy". `stream_stop` ends it at once, and the viewer
also drops its queue, ends the MediaSource and revokes the object URL on
the way out, any of which could be holding megabytes of decrypted video.

While there: `file_chunk` replies were matched to their requests by
arrival order, which was true by luck rather than by construction. The
reply now names the file it belongs to and is matched on that and the
chunk index; a chunk nobody is waiting for is dropped instead of being
handed to whatever request happens to be oldest.

**The administration panel counted its own history.** A deleted account
is tombstoned so the connection log stays readable, and every count and
list treated that row as a user — including a group's member count, and
the member list of the group itself. They do not any more.

**Where a node is.** `endpoint_hint` is what a node believes its address
to be, learned from a STUN server and sent to us: useful for reaching it,
and a claim. The announcement that carries it is signed with the node key
over a fresh timestamp, so the address that request *arrives from* is the
address of whoever holds that key — that is now recorded on the node row
and shown in a Nodes tab, next to the hint, with the difference spelled
out. Clients get the same treatment: `webrtc_offer` is logged with the
address the hub saw when a browser starts a peer connection.

Verified against the live deployment: the node's row reads 90.112.206.172
after a restart, and in e2e a stopped stream goes quiet in one message
and the next one starts immediately instead of being refused.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 60 insertions, 6 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/admin.py b/packages/meshbay-hub/src/meshbay_hub/api/admin.py
index 785731d..7ee05ca 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/admin.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/admin.py
@@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.auth import decrypt_email
 from meshbay_hub.api.deps import require_admin, require_moderator
-from meshbay_hub.api.revocation import get_connected_node_count
+from meshbay_hub.api.revocation import get_connected_node_count, is_node_connected
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import Group, GroupMember, IPLog, Node, User
 
@@ -42,8 +42,18 @@ async def admin_stats(
     current_user: User = Depends(require_moderator),
     db: AsyncSession = Depends(get_db),
 ):
-    user_count = (await db.execute(select(func.count()).select_from(User))).scalar_one()
-    group_count = (await db.execute(select(func.count()).select_from(Group))).scalar_one()
+    # Deleted accounts are tombstoned rather than dropped, so that the
+    # connection log stays readable. They are not users any more and must not be
+    # counted as any: a hub whose user count only ever rises is measuring its
+    # own history, not its population.
+    user_count = (await db.execute(
+        select(func.count()).select_from(User)
+        .where(User.status != "deleted"))).scalar_one()
+    # Groups are not tombstoned — deleting one removes the row — so every group
+    # here is a group. A revoked one is suspended by moderation and still shown
+    # in the list, so counting it keeps the two consistent.
+    group_count = (await db.execute(
+        select(func.count()).select_from(Group))).scalar_one()
     node_count = (await db.execute(select(func.count()).select_from(Node))).scalar_one()
     return {
         "users": user_count,
@@ -63,14 +73,16 @@ async def admin_list_users(
     offset: int = 0,
     limit: int = Query(default=50, le=200),
 ):
-    query = select(User).order_by(User.created_at.desc())
+    query = (select(User).where(User.status != "deleted")
+             .order_by(User.created_at.desc()))
     if q:
         query = query.where(User.username.ilike(f"%{q}%"))
     query = query.offset(offset).limit(limit)
     result = await db.execute(query)
     users = result.scalars().all()
 
-    total_query = select(func.count()).select_from(User)
+    total_query = (select(func.count()).select_from(User)
+                   .where(User.status != "deleted"))
     if q:
         total_query = total_query.where(User.username.ilike(f"%{q}%"))
     total = (await db.execute(total_query)).scalar_one()
@@ -222,9 +234,13 @@ async def admin_list_groups(
     query = (
         select(
             Group,
-            func.count(GroupMember.user_id).label("member_count"),
+            func.count(User.id).label("member_count"),
         )
+        # Members, not rows: a deleted account's membership is removed with it,
+        # but joining through User keeps the count honest if one ever survives.
         .outerjoin(GroupMember, Group.id == GroupMember.group_id)
+        .outerjoin(User, (User.id == GroupMember.user_id)
+                   & (User.status != "deleted"))
         .group_by(Group.id)
         .order_by(Group.created_at.desc())
         .offset(offset)
@@ -288,6 +304,44 @@ async def admin_patch_group(
 
 # ── IP Audit Logs ────────────────────────────────────────────────────────────
 
+@router.get("/nodes")
+async def admin_list_nodes(
+    current_user: User = Depends(require_moderator),
+    db: AsyncSession = Depends(get_db),
+    limit: int = Query(default=100, le=200),
+):
+    """
+    Registered nodes, with the address the hub saw them announce from.
+
+    `observed_ip` is the one to answer a question with: it comes from the
+    connection that carried a valid Ed25519 signature over a fresh timestamp, so
+    it is the address of whoever holds the node key. `endpoint_hint` is what the
+    node believes its own address to be, discovered through a STUN server and
+    sent to us — useful for reaching it, and not evidence of anything.
+    """
+    rows = (await db.execute(
+        select(Node, User.username)
+        .outerjoin(User, User.id == Node.user_id)
+        .order_by(Node.announced_at.desc())
+        .limit(limit))).all()
+    return {
+        "nodes": [
+            {
+                "id": n.id,
+                "user_id": n.user_id,
+                "username": uname or "",
+                "pk_node": n.pk_node,
+                "observed_ip": n.observed_ip or "",
+                "endpoint_hint": n.endpoint_hint or "",
+                "last_seen": n.last_seen.isoformat() if n.last_seen else "",
+                "announced_at": n.announced_at.isoformat(),
+                "online": is_node_connected(n.id),
+            }
+            for n, uname in rows
+        ],
+    }
+
+
 @router.get("/logs")
 async def admin_list_logs(
     current_user: User = Depends(require_moderator),