"""
MeshBay Node — local web UI (localhost:18000).
Minimal FastAPI app providing:
GET / → status page (HTML)
GET /api/status → JSON status
GET /api/files → JSON file list from the group index
GET /api/config → JSON config summary (no secrets)
Served only on 127.0.0.1 — not exposed to the network.
"""
import asyncio
import json
import logging
from typing import TYPE_CHECKING
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from meshbay_node import __version__
if TYPE_CHECKING:
from meshbay_node.chat.store import ChatStore
from meshbay_node.indexer import GroupIndex
log = logging.getLogger(__name__)
def create_ui_app(state: dict) -> FastAPI:
"""
Create the UI FastAPI app.
state dict is updated by the daemon and read by UI endpoints:
state["status"] : str — "starting" | "running" | "error"
state["group_id"] : str
state["group_name"] : str
state["hub_url"] : str
state["username"] : str
state["node_port"] : int
state["index"] : GroupIndex | None
state["endpoint_hint"]: str | None
"""
app = FastAPI(
title="MeshBay Node UI",
version=__version__,
docs_url=None, # disable Swagger on local UI
redoc_url=None,
)
@app.get("/api/status")
async def api_status():
index = state.get("index")
return {
"version": __version__,
"status": state.get("status", "starting"),
"hub_url": state.get("hub_url", ""),
"username": state.get("username", ""),
"group_id": state.get("group_id", ""),
"group_name": state.get("group_name", ""),
"node_port": state.get("node_port", 0),
"endpoint_hint": state.get("endpoint_hint"),
"file_count": index.count if index else 0,
"index_version": index.version if index else 0,
}
@app.get("/api/files")
async def api_files():
index = state.get("index")
if not index:
return {"files": []}
return {
"files": [
{
"id": e.id[:16] + "…",
"name": e.name,
"path": e.path,
"size": e.size,
"type": e.type,
"duration": e.duration,
}
for e in index.entries
]
}
@app.get("/", response_class=HTMLResponse)
async def root():
index = state.get("index")
status = state.get("status", "starting")
file_count = index.count if index else 0
status_color = {"running": "#22c55e", "error": "#ef4444"}.get(status, "#f59e0b")
files_html = ""
if index:
rows = "".join(
f"| {e.name} | {e.type} | "
f"{e.size // 1024} KB | {e.path or '/'} |
"
for e in index.entries
)
files_html = f"""
Files ({file_count})
"""
return f"""
MeshBay Node
🔗 MeshBay Node {status}
Hub: {state.get("hub_url", "—")} |
User: {state.get("username", "—")} |
Group: {state.get("group_name") or state.get("group_id") or "—"} |
Port: {state.get("node_port", "—")}
Endpoint: {state.get("endpoint_hint") or "unknown"}
{files_html}
MeshBay Node v{__version__} — JSON status
— JSON files — Chat
"""
# ── Chat endpoints ───────────────────────────────────────────────────────
_chat_subscribers: list[WebSocket] = []
@app.get("/api/chat/history")
async def chat_history(since: float = 0, limit: int = 100):
chat_store = state.get("chat_store")
if not chat_store:
return {"messages": []}
msgs = await chat_store.get_messages(since=since, limit=limit)
return {
"messages": [
{
"id": m.id,
"sender_id": m.sender_id,
"iteration": m.iteration,
"timestamp": m.timestamp,
"thread_id": m.thread_id,
}
for m in msgs
]
}
@app.websocket("/ws/chat")
async def chat_websocket(ws: WebSocket):
"""WebSocket for real-time chat push to the local UI."""
await ws.accept()
_chat_subscribers.append(ws)
try:
while True:
await ws.receive_text()
except WebSocketDisconnect:
pass
finally:
_chat_subscribers.remove(ws)
async def broadcast_chat_to_ui(msg: dict) -> None:
"""Push a chat message to all connected UI WebSocket clients."""
payload = json.dumps(msg)
dead = []
for ws in _chat_subscribers:
try:
await ws.send_text(payload)
except Exception:
dead.append(ws)
for ws in dead:
_chat_subscribers.remove(ws)
app.broadcast_chat = broadcast_chat_to_ui
@app.get("/chat", response_class=HTMLResponse)
async def chat_page():
return f"""
MeshBay Chat
MeshBay Chat
Back to status
"""
return app