Private
Public Access
Compare commits
6
Commits
v2026.9.4
...
v2026.9.28
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c54c1bf97d | ||
|
|
3349cadea2 | ||
|
|
047f47d524 | ||
|
|
850d9ec0fa | ||
|
|
86584660fb | ||
|
|
6f67f36578 |
No files matched your search
+6
-2
@@ -58,8 +58,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|||||||
# ASGITransport, used by the plain REST test fixtures -- only
|
# ASGITransport, used by the plain REST test fixtures -- only
|
||||||
# TestClient's websocket_connect-based tests actually run lifespan).
|
# TestClient's websocket_connect-based tests actually run lifespan).
|
||||||
# The background listener task genuinely needs a running event loop
|
# The background listener task genuinely needs a running event loop
|
||||||
# though, so that part stays here.
|
# though, so that part stays here. run_forever() (not listen() directly)
|
||||||
listener_task = asyncio.create_task(app.state.broadcaster.listen())
|
# -- #76: a bare listen() here died permanently and silently the moment
|
||||||
|
# its Redis connection ever dropped (e.g. Redis itself restarting),
|
||||||
|
# with nothing to notice or reconnect; run_forever() supervises it and
|
||||||
|
# restarts on failure instead.
|
||||||
|
listener_task = asyncio.create_task(app.state.broadcaster.run_forever())
|
||||||
|
|
||||||
yield
|
yield
|
||||||
|
|
||||||
|
|||||||
@@ -1,13 +1,26 @@
|
|||||||
|
import asyncio
|
||||||
|
import contextlib
|
||||||
import json
|
import json
|
||||||
|
import logging
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from redis.asyncio import Redis
|
from redis.asyncio import Redis
|
||||||
|
|
||||||
from app.ws.connection_manager import ConnectionManager
|
from app.ws.connection_manager import ConnectionManager
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
ROOM_CHANNEL_PREFIX = "room:"
|
ROOM_CHANNEL_PREFIX = "room:"
|
||||||
USER_CHANNEL_PREFIX = "user:"
|
USER_CHANNEL_PREFIX = "user:"
|
||||||
|
|
||||||
|
# #76: a run shorter than this before dying doesn't count as "stable" for
|
||||||
|
# resetting the backoff in run_forever() below -- long enough that a real,
|
||||||
|
# working connection has clearly settled in, short enough not to leave a
|
||||||
|
# just-recovered listener exposed to a full backoff climb if Redis blips
|
||||||
|
# again shortly after reconnecting.
|
||||||
|
STABLE_RUN_SECONDS = 60
|
||||||
|
MAX_BACKOFF_SECONDS = 30
|
||||||
|
|
||||||
|
|
||||||
class Broadcaster:
|
class Broadcaster:
|
||||||
"""Cross-instance message fan-out (ARCHITECTURE.md phase 5).
|
"""Cross-instance message fan-out (ARCHITECTURE.md phase 5).
|
||||||
@@ -36,6 +49,43 @@ class Broadcaster:
|
|||||||
async def publish_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
async def publish_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
||||||
await self._redis.publish(f"{USER_CHANNEL_PREFIX}{user_id}", json.dumps(payload))
|
await self._redis.publish(f"{USER_CHANNEL_PREFIX}{user_id}", json.dumps(payload))
|
||||||
|
|
||||||
|
async def run_forever(self) -> None:
|
||||||
|
"""Supervises listen(), restarting it if it ever dies.
|
||||||
|
|
||||||
|
#76's real root cause, found after the per-message try/except below
|
||||||
|
turned out not to be enough: this is the *one* pubsub listener for
|
||||||
|
the whole process (see main.py's lifespan), created once at startup
|
||||||
|
with nothing watching over it afterward. A connection-level failure
|
||||||
|
-- Redis restarting, a network blip between the app and Redis --
|
||||||
|
raises from the `async for` iteration in listen() itself, outside
|
||||||
|
that per-message guard, and previously killed this task permanently
|
||||||
|
and silently: HTTP/REST and login kept working fine (neither
|
||||||
|
touches this), but live message delivery was gone for every room
|
||||||
|
and every user on this instance until the whole process was
|
||||||
|
restarted. Confirmed live in production: Redis got restarted by an
|
||||||
|
OS update, and `PUBSUB NUMPAT` against it afterward showed zero
|
||||||
|
pattern subscriptions across all 4 gunicorn workers -- every one of
|
||||||
|
them had silently lost its listener with nothing to bring it back.
|
||||||
|
"""
|
||||||
|
backoff_seconds = 1.0
|
||||||
|
while True:
|
||||||
|
started_at = asyncio.get_event_loop().time()
|
||||||
|
try:
|
||||||
|
await self.listen()
|
||||||
|
logger.warning("Broadcaster.listen() exited without error -- restarting")
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
raise
|
||||||
|
except Exception:
|
||||||
|
logger.error("Broadcaster.listen() failed -- restarting", exc_info=True)
|
||||||
|
ran_for = asyncio.get_event_loop().time() - started_at
|
||||||
|
# Sleep at the *current* backoff first, then grow it for next
|
||||||
|
# time -- computing the next value before sleeping would skip
|
||||||
|
# straight past the first (shortest) retry delay.
|
||||||
|
await asyncio.sleep(backoff_seconds)
|
||||||
|
backoff_seconds = (
|
||||||
|
1.0 if ran_for > STABLE_RUN_SECONDS else min(backoff_seconds * 2, MAX_BACKOFF_SECONDS)
|
||||||
|
)
|
||||||
|
|
||||||
async def listen(self) -> None:
|
async def listen(self) -> None:
|
||||||
pubsub = self._redis.pubsub()
|
pubsub = self._redis.pubsub()
|
||||||
await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
||||||
@@ -43,6 +93,16 @@ class Broadcaster:
|
|||||||
async for message in pubsub.listen():
|
async for message in pubsub.listen():
|
||||||
if message["type"] != "pmessage":
|
if message["type"] != "pmessage":
|
||||||
continue
|
continue
|
||||||
|
# #76: this is the one pubsub listener for the whole process
|
||||||
|
# (see main.py's lifespan) -- ConnectionManager.broadcast/
|
||||||
|
# send_to_user already guard against one dead socket taking
|
||||||
|
# the rest of a single fan-out down, but this belt-and-
|
||||||
|
# suspenders catch is for anything else unexpected (a
|
||||||
|
# malformed payload, e.g.) doing the same. An uncaught
|
||||||
|
# exception here previously meant *this whole async for*
|
||||||
|
# loop died silently -- no more live delivery to any room on
|
||||||
|
# this instance, for anyone, until the process restarted.
|
||||||
|
try:
|
||||||
channel = message["channel"]
|
channel = message["channel"]
|
||||||
payload = json.loads(message["data"])
|
payload = json.loads(message["data"])
|
||||||
if channel.startswith(ROOM_CHANNEL_PREFIX):
|
if channel.startswith(ROOM_CHANNEL_PREFIX):
|
||||||
@@ -51,6 +111,19 @@ class Broadcaster:
|
|||||||
elif channel.startswith(USER_CHANNEL_PREFIX):
|
elif channel.startswith(USER_CHANNEL_PREFIX):
|
||||||
user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
|
user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
|
||||||
await self._manager.send_to_user(user_id, payload)
|
await self._manager.send_to_user(user_id, payload)
|
||||||
|
except Exception:
|
||||||
|
logger.error("Error handling pubsub message on channel %s", message.get("channel"), exc_info=True)
|
||||||
finally:
|
finally:
|
||||||
|
# #76: reached via a connection that's already dead just as
|
||||||
|
# often as a clean shutdown (that's the whole failure mode
|
||||||
|
# run_forever() now recovers from) -- punsubscribe/aclose over
|
||||||
|
# a broken connection would themselves raise, masking whatever
|
||||||
|
# actually killed the loop above and, since run_forever() only
|
||||||
|
# starts backing off *after* listen() returns/raises, delaying
|
||||||
|
# the reconnect for no benefit (a fresh pubsub object is about
|
||||||
|
# to be created on retry regardless of whether this cleanup
|
||||||
|
# succeeded).
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
await pubsub.aclose()
|
await pubsub.aclose()
|
||||||
+33
-1
@@ -1,3 +1,4 @@
|
|||||||
|
import asyncio
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
||||||
@@ -31,6 +32,18 @@ router = APIRouter(tags=["ws"])
|
|||||||
|
|
||||||
WS_UNAUTHENTICATED = 4401
|
WS_UNAUTHENTICATED = 4401
|
||||||
|
|
||||||
|
# #76: comfortably under the ~60s idle-connection timeout common to reverse
|
||||||
|
# proxies/load balancers sitting in front of this endpoint, and short enough
|
||||||
|
# that the frontend's own zombie-detection watchdog (useChatSocket.ts) can
|
||||||
|
# notice a dead connection and reconnect well within a session -- a socket
|
||||||
|
# can go silently dead (no close frame at all, e.g. a laptop sleeping or a
|
||||||
|
# NAT dropping an idle mapping) without this, and no live message reaches it
|
||||||
|
# again until *something* else finally triggers a close, which used to be
|
||||||
|
# "switch rooms and back" purely by accident (that remounts the message
|
||||||
|
# list, which refetches history over plain REST -- unrelated to the socket
|
||||||
|
# itself actually recovering).
|
||||||
|
WS_PING_INTERVAL_SECONDS = 30
|
||||||
|
|
||||||
|
|
||||||
class ClientEnvelope(BaseModel):
|
class ClientEnvelope(BaseModel):
|
||||||
type: str
|
type: str
|
||||||
@@ -124,7 +137,18 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
raw = await websocket.receive_json()
|
try:
|
||||||
|
raw = await asyncio.wait_for(
|
||||||
|
websocket.receive_json(), timeout=WS_PING_INTERVAL_SECONDS
|
||||||
|
)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
# Nothing received in a while -- ping rather than assume the
|
||||||
|
# connection's still good. A genuinely dead socket fails this
|
||||||
|
# send (raising, caught below same as any other disconnect);
|
||||||
|
# a merely quiet one just gets proded and the loop continues
|
||||||
|
# waiting.
|
||||||
|
await websocket.send_json({"type": "ping"})
|
||||||
|
continue
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
envelope = ClientEnvelope.model_validate(raw)
|
envelope = ClientEnvelope.model_validate(raw)
|
||||||
@@ -146,6 +170,14 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
|
|||||||
joined_rooms.add(envelope.room_id)
|
joined_rooms.add(envelope.room_id)
|
||||||
await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)})
|
await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)})
|
||||||
|
|
||||||
|
elif envelope.type == "pong":
|
||||||
|
# #76: the client's reply to our "ping" above -- nothing
|
||||||
|
# to do with it. Its only purpose is resetting the
|
||||||
|
# frontend's own liveness watchdog; the timeout on the
|
||||||
|
# receive_json() above already gets reset by receiving
|
||||||
|
# any frame at all, this one included.
|
||||||
|
pass
|
||||||
|
|
||||||
elif envelope.type == "leave":
|
elif envelope.type == "leave":
|
||||||
if envelope.room_id is None:
|
if envelope.room_id is None:
|
||||||
await websocket.send_json({"type": "error", "detail": "room_id required"})
|
await websocket.send_json({"type": "error", "detail": "room_id required"})
|
||||||
|
|||||||
@@ -1,8 +1,11 @@
|
|||||||
|
import logging
|
||||||
import uuid
|
import uuid
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
|
|
||||||
from fastapi import WebSocket
|
from fastapi import WebSocket
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class ConnectionManager:
|
class ConnectionManager:
|
||||||
"""Local, single-process WebSocket socket registry.
|
"""Local, single-process WebSocket socket registry.
|
||||||
@@ -43,8 +46,22 @@ class ConnectionManager:
|
|||||||
|
|
||||||
async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
|
async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
|
||||||
for websocket in list(self._rooms.get(room_id, ())):
|
for websocket in list(self._rooms.get(room_id, ())):
|
||||||
|
# #76: one dead/racing socket (e.g. its own disconnect cleanup
|
||||||
|
# hasn't finished removing it from _rooms yet) must not take the
|
||||||
|
# rest of this broadcast down with it -- this runs inside
|
||||||
|
# Broadcaster.listen()'s single, never-restarted per-process
|
||||||
|
# pubsub loop, so an unhandled exception here previously meant
|
||||||
|
# one bad socket silently killed *live delivery for every room
|
||||||
|
# on this instance*, not just the one socket, until the process
|
||||||
|
# was restarted.
|
||||||
|
try:
|
||||||
await websocket.send_json(payload)
|
await websocket.send_json(payload)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Dropping dead socket during broadcast to room %s", room_id, exc_info=True)
|
||||||
|
|
||||||
async def send_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
async def send_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
||||||
for websocket in list(self._users.get(user_id, ())):
|
for websocket in list(self._users.get(user_id, ())):
|
||||||
|
try:
|
||||||
await websocket.send_json(payload)
|
await websocket.send_json(payload)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Dropping dead socket sending to user %s", user_id, exc_info=True)
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "ds-chat"
|
name = "ds-chat"
|
||||||
version = "2026.9.4"
|
version = "2026.9.28"
|
||||||
description = "DS Chat backend service"
|
description = "DS Chat backend service"
|
||||||
license = { text = "AGPL-3.0-or-later" }
|
license = { text = "AGPL-3.0-or-later" }
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
|
|||||||
@@ -0,0 +1,61 @@
|
|||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.ws.broadcaster import Broadcaster
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_forever_restarts_after_listen_raises(monkeypatch):
|
||||||
|
# #76: listen() dying (a dropped Redis connection, most commonly -- e.g.
|
||||||
|
# Redis itself restarting) must not permanently kill live delivery for
|
||||||
|
# the whole process. run_forever() has to notice and restart it, not
|
||||||
|
# just let the background task disappear silently while everything else
|
||||||
|
# (login, REST endpoints) keeps working fine.
|
||||||
|
broadcaster = Broadcaster(redis=None, manager=None)
|
||||||
|
calls = 0
|
||||||
|
|
||||||
|
async def fake_listen():
|
||||||
|
nonlocal calls
|
||||||
|
calls += 1
|
||||||
|
if calls < 3:
|
||||||
|
raise ConnectionError("redis went away")
|
||||||
|
# Stop the loop once the retry behavior below is proven, rather
|
||||||
|
# than looping forever in a test.
|
||||||
|
raise asyncio.CancelledError()
|
||||||
|
|
||||||
|
monkeypatch.setattr(broadcaster, "listen", fake_listen)
|
||||||
|
sleep_calls = []
|
||||||
|
|
||||||
|
async def fake_sleep(seconds):
|
||||||
|
sleep_calls.append(seconds)
|
||||||
|
|
||||||
|
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
|
||||||
|
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await broadcaster.run_forever()
|
||||||
|
|
||||||
|
assert calls == 3
|
||||||
|
# Slept once after each of the first two failures (exponential:
|
||||||
|
# 1s, then 2s) -- not after the third, which is the CancelledError
|
||||||
|
# that stops the loop before reaching the sleep.
|
||||||
|
assert sleep_calls == [1.0, 2.0]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_forever_does_not_swallow_cancellation(monkeypatch):
|
||||||
|
# A genuine shutdown (the app's own lifespan cancels this task) must
|
||||||
|
# propagate immediately, not get treated as just another failure to
|
||||||
|
# retry after a backoff.
|
||||||
|
broadcaster = Broadcaster(redis=None, manager=None)
|
||||||
|
|
||||||
|
async def fake_listen():
|
||||||
|
raise asyncio.CancelledError()
|
||||||
|
|
||||||
|
monkeypatch.setattr(broadcaster, "listen", fake_listen)
|
||||||
|
|
||||||
|
async def fail_if_called(seconds):
|
||||||
|
raise AssertionError("should not sleep/retry on cancellation")
|
||||||
|
|
||||||
|
monkeypatch.setattr(asyncio, "sleep", fail_if_called)
|
||||||
|
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await broadcaster.run_forever()
|
||||||
@@ -109,6 +109,24 @@ def test_history_tiebreaks_identical_timestamps_by_id(ws_client):
|
|||||||
assert [m["content"] for m in history] == [m["content"] for m in expected_order]
|
assert [m["content"] for m in history] == [m["content"] for m in expected_order]
|
||||||
|
|
||||||
|
|
||||||
|
def test_ws_sends_heartbeat_ping(ws_client, monkeypatch):
|
||||||
|
# #76: with nothing else sent on the connection, a "ping" must arrive
|
||||||
|
# proactively once the idle timeout elapses -- shrunk here so the test
|
||||||
|
# doesn't actually wait the real 30s. Replying "pong" (the frontend's
|
||||||
|
# own behavior) must be accepted silently and leave the connection
|
||||||
|
# fully usable afterward, not treated as an unknown message type.
|
||||||
|
monkeypatch.setattr("app.ws.chat.WS_PING_INTERVAL_SECONDS", 0.05)
|
||||||
|
username = _unique("alice")
|
||||||
|
_register(ws_client, username=username)
|
||||||
|
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
||||||
|
|
||||||
|
with ws_client.websocket_connect("/ws/chat") as ws:
|
||||||
|
assert ws.receive_json() == {"type": "ping"}
|
||||||
|
ws.send_json({"type": "pong"})
|
||||||
|
ws.send_json({"type": "join", "room_id": room["id"]})
|
||||||
|
assert ws.receive_json() == {"type": "joined", "room_id": room["id"]}
|
||||||
|
|
||||||
|
|
||||||
def test_ws_message_without_join_errors(ws_client):
|
def test_ws_message_without_join_errors(ws_client):
|
||||||
_register(ws_client, username=_unique("alice"))
|
_register(ws_client, username=_unique("alice"))
|
||||||
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
||||||
|
|||||||
Generated
+953
-11
File diff suppressed because it is too large.
Load diff
@@ -1,14 +1,15 @@
|
|||||||
{
|
{
|
||||||
"name": "frontend",
|
"name": "frontend",
|
||||||
"private": true,
|
"private": true,
|
||||||
"version": "2026.9.4",
|
"version": "2026.9.28",
|
||||||
"license": "AGPL-3.0-or-later",
|
"license": "AGPL-3.0-or-later",
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"dev": "vite",
|
"dev": "vite",
|
||||||
"build": "tsc -b && vite build",
|
"build": "tsc -b && vite build",
|
||||||
"lint": "oxlint",
|
"lint": "oxlint",
|
||||||
"preview": "vite preview"
|
"preview": "vite preview",
|
||||||
|
"test": "vitest run"
|
||||||
},
|
},
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"markdown-to-jsx": "^9.10.2",
|
"markdown-to-jsx": "^9.10.2",
|
||||||
@@ -18,13 +19,16 @@
|
|||||||
"vite-plugin-pwa": "^1.3.0"
|
"vite-plugin-pwa": "^1.3.0"
|
||||||
},
|
},
|
||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
|
"@testing-library/react": "^16.3.3",
|
||||||
"@types/node": "^24.13.3",
|
"@types/node": "^24.13.3",
|
||||||
"@types/react": "^19.2.17",
|
"@types/react": "^19.2.17",
|
||||||
"@types/react-dom": "^19.2.3",
|
"@types/react-dom": "^19.2.3",
|
||||||
"@vitejs/plugin-react": "^6.0.4",
|
"@vitejs/plugin-react": "^6.0.4",
|
||||||
|
"jsdom": "^30.1.1",
|
||||||
"oxlint": "^1.75.0",
|
"oxlint": "^1.75.0",
|
||||||
"typescript": "~6.0.2",
|
"typescript": "~6.0.2",
|
||||||
"vite": "^8.2.0",
|
"vite": "^8.2.0",
|
||||||
|
"vitest": "^5.0.2",
|
||||||
"workbox-cacheable-response": "^7.4.1",
|
"workbox-cacheable-response": "^7.4.1",
|
||||||
"workbox-expiration": "^7.4.1",
|
"workbox-expiration": "^7.4.1",
|
||||||
"workbox-precaching": "^7.4.1",
|
"workbox-precaching": "^7.4.1",
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
import { renderHook } from '@testing-library/react'
|
||||||
|
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||||
|
import * as desktopBridge from '../lib/desktopBridge'
|
||||||
|
import { useDesktopRoomBridge } from './useDesktopRoomBridge'
|
||||||
|
|
||||||
|
describe('useDesktopRoomBridge', () => {
|
||||||
|
beforeEach(() => {
|
||||||
|
vi.spyOn(desktopBridge, 'notifyRoomOpened')
|
||||||
|
vi.spyOn(desktopBridge, 'notifyUnreadCount')
|
||||||
|
})
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('calls roomOpened once when entering a room', () => {
|
||||||
|
renderHook(({ roomId, count }) => useDesktopRoomBridge(roomId, count), {
|
||||||
|
initialProps: { roomId: 'room-1', count: 0 },
|
||||||
|
})
|
||||||
|
expect(desktopBridge.notifyRoomOpened).toHaveBeenCalledTimes(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('does not call roomOpened again on a rerender with the same roomId', () => {
|
||||||
|
const { rerender } = renderHook(({ roomId, count }) => useDesktopRoomBridge(roomId, count), {
|
||||||
|
initialProps: { roomId: 'room-1', count: 0 },
|
||||||
|
})
|
||||||
|
expect(desktopBridge.notifyRoomOpened).toHaveBeenCalledTimes(1)
|
||||||
|
|
||||||
|
// Unrelated rerender -- unreadRoomCount changed, roomId did not.
|
||||||
|
rerender({ roomId: 'room-1', count: 3 })
|
||||||
|
rerender({ roomId: 'room-1', count: 3 })
|
||||||
|
expect(desktopBridge.notifyRoomOpened).toHaveBeenCalledTimes(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('calls roomOpened again on a genuine transition to a different room', () => {
|
||||||
|
const { rerender } = renderHook(({ roomId, count }) => useDesktopRoomBridge(roomId, count), {
|
||||||
|
initialProps: { roomId: 'room-1', count: 0 },
|
||||||
|
})
|
||||||
|
rerender({ roomId: 'room-2', count: 0 })
|
||||||
|
expect(desktopBridge.notifyRoomOpened).toHaveBeenCalledTimes(2)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('does not call roomOpened when there is no room id', () => {
|
||||||
|
renderHook(({ roomId, count }) => useDesktopRoomBridge(roomId, count), {
|
||||||
|
initialProps: { roomId: undefined as string | undefined, count: 0 },
|
||||||
|
})
|
||||||
|
expect(desktopBridge.notifyRoomOpened).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('sends the unread count whenever it changes, independent of roomOpened', () => {
|
||||||
|
const { rerender } = renderHook(({ roomId, count }) => useDesktopRoomBridge(roomId, count), {
|
||||||
|
initialProps: { roomId: undefined as string | undefined, count: 0 },
|
||||||
|
})
|
||||||
|
expect(desktopBridge.notifyUnreadCount).toHaveBeenCalledWith(0)
|
||||||
|
|
||||||
|
rerender({ roomId: undefined, count: 2 })
|
||||||
|
expect(desktopBridge.notifyUnreadCount).toHaveBeenCalledWith(2)
|
||||||
|
expect(desktopBridge.notifyRoomOpened).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
|
describe('desktop bridge functions with no window.dsDesktop', () => {
|
||||||
|
it('remain optional -- calling them does not throw when the bridge is absent', () => {
|
||||||
|
expect(window.dsDesktop).toBeUndefined()
|
||||||
|
expect(() => desktopBridge.notifyRoomOpened()).not.toThrow()
|
||||||
|
expect(() => desktopBridge.notifyUnreadCount(5)).not.toThrow()
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -0,0 +1,26 @@
|
|||||||
|
import { useEffect } from 'react'
|
||||||
|
import { notifyRoomOpened, notifyUnreadCount } from '../lib/desktopBridge'
|
||||||
|
|
||||||
|
// Tells the Electron desktop wrapper (a) when the user has genuinely
|
||||||
|
// entered a room, so it can clear that room's tray indicator, and (b) the
|
||||||
|
// current unread-room count, so the tray badge stays accurate. Both are
|
||||||
|
// no-ops in every browser build (see desktopBridge.ts).
|
||||||
|
//
|
||||||
|
// roomOpened must fire only on a genuine transition to a real room id --
|
||||||
|
// never merely because the window was shown, focused, restored, or opened
|
||||||
|
// from the tray, and never when no room is open at all. Depending on just
|
||||||
|
// the id (a primitive, not the room object) is what makes this safe to
|
||||||
|
// call from a component that rerenders often: an unrelated rerender that
|
||||||
|
// leaves activeRoomId unchanged doesn't rerun the effect, so there's no
|
||||||
|
// separate de-duplication to get wrong here -- it falls out of React's own
|
||||||
|
// effect-dependency comparison.
|
||||||
|
export function useDesktopRoomBridge(activeRoomId: string | undefined, unreadRoomCount: number): void {
|
||||||
|
useEffect(() => {
|
||||||
|
if (!activeRoomId) return
|
||||||
|
notifyRoomOpened()
|
||||||
|
}, [activeRoomId])
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
notifyUnreadCount(unreadRoomCount)
|
||||||
|
}, [unreadRoomCount])
|
||||||
|
}
|
||||||
@@ -15,6 +15,12 @@ declare global {
|
|||||||
interface Window {
|
interface Window {
|
||||||
dsDesktop?: {
|
dsDesktop?: {
|
||||||
setUnreadCount?(unreadRoomCount: number): void
|
setUnreadCount?(unreadRoomCount: number): void
|
||||||
|
// Tells the tray/dock icon the user has genuinely entered a room --
|
||||||
|
// the desktop wrapper's own signal to clear that room's tray
|
||||||
|
// indicator. Deliberately not fired just because the window was
|
||||||
|
// shown/focused/restored, or opened from the tray: none of those mean
|
||||||
|
// the user actually looked at the room the unread message is in.
|
||||||
|
roomOpened?(): void
|
||||||
showNotification?(notification: DesktopNotificationRequest): void
|
showNotification?(notification: DesktopNotificationRequest): void
|
||||||
onNotificationClick?(callback: (roomId: string) => void): () => void
|
onNotificationClick?(callback: (roomId: string) => void): () => void
|
||||||
}
|
}
|
||||||
@@ -38,6 +44,16 @@ export function isDesktopClickListenerSupported(): boolean {
|
|||||||
return typeof window.dsDesktop?.onNotificationClick === 'function'
|
return typeof window.dsDesktop?.onNotificationClick === 'function'
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Both no-op silently if the bridge or this specific method isn't present,
|
||||||
|
// same as showDesktopNotification below -- callers never need to guard.
|
||||||
|
export function notifyRoomOpened(): void {
|
||||||
|
window.dsDesktop?.roomOpened?.()
|
||||||
|
}
|
||||||
|
|
||||||
|
export function notifyUnreadCount(unreadRoomCount: number): void {
|
||||||
|
window.dsDesktop?.setUnreadCount?.(unreadRoomCount)
|
||||||
|
}
|
||||||
|
|
||||||
// No-ops silently if the bridge or this specific method isn't present --
|
// No-ops silently if the bridge or this specific method isn't present --
|
||||||
// callers don't need to guard, matching the rest of this module's
|
// callers don't need to guard, matching the rest of this module's
|
||||||
// capability-detect-per-method philosophy (see #49's "feature-test each
|
// capability-detect-per-method philosophy (see #49's "feature-test each
|
||||||
|
|||||||
@@ -0,0 +1,56 @@
|
|||||||
|
import { describe, expect, it } from 'vitest'
|
||||||
|
import type { MyRoomItem } from '../types'
|
||||||
|
import { clearRoomIndicators } from './roomIndicators'
|
||||||
|
|
||||||
|
function makeRoom(overrides: Partial<MyRoomItem> & { id: string }): MyRoomItem {
|
||||||
|
return {
|
||||||
|
name: overrides.id,
|
||||||
|
description: null,
|
||||||
|
is_private: false,
|
||||||
|
is_dm: false,
|
||||||
|
is_archived: false,
|
||||||
|
owner_id: 'owner',
|
||||||
|
created_at: '2026-01-01T00:00:00Z',
|
||||||
|
role: 'member',
|
||||||
|
has_unread: false,
|
||||||
|
has_mention: false,
|
||||||
|
dm_partner: null,
|
||||||
|
email_notifications: false,
|
||||||
|
...overrides,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('clearRoomIndicators', () => {
|
||||||
|
it('clears has_unread and has_mention on the matching room', () => {
|
||||||
|
const rooms = [makeRoom({ id: 'room-1', has_unread: true, has_mention: true })]
|
||||||
|
const result = clearRoomIndicators(rooms, 'room-1')
|
||||||
|
expect(result[0].has_unread).toBe(false)
|
||||||
|
expect(result[0].has_mention).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('preserves unread state on every other room', () => {
|
||||||
|
const rooms = [
|
||||||
|
makeRoom({ id: 'room-1', has_unread: true, has_mention: true }),
|
||||||
|
makeRoom({ id: 'room-2', has_unread: true, has_mention: false }),
|
||||||
|
makeRoom({ id: 'room-3', has_unread: false, has_mention: false }),
|
||||||
|
]
|
||||||
|
const result = clearRoomIndicators(rooms, 'room-1')
|
||||||
|
const room2 = result.find((r) => r.id === 'room-2')
|
||||||
|
const room3 = result.find((r) => r.id === 'room-3')
|
||||||
|
expect(room2?.has_unread).toBe(true)
|
||||||
|
expect(room3?.has_unread).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('leaves untouched rooms as the same object reference', () => {
|
||||||
|
const room2 = makeRoom({ id: 'room-2', has_unread: true })
|
||||||
|
const rooms = [makeRoom({ id: 'room-1', has_unread: true }), room2]
|
||||||
|
const result = clearRoomIndicators(rooms, 'room-1')
|
||||||
|
expect(result[1]).toBe(room2)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('is a no-op if the room id is not found', () => {
|
||||||
|
const rooms = [makeRoom({ id: 'room-1', has_unread: true })]
|
||||||
|
const result = clearRoomIndicators(rooms, 'does-not-exist')
|
||||||
|
expect(result[0].has_unread).toBe(true)
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
import type { MyRoomItem } from '../types'
|
||||||
|
|
||||||
|
// Pulled out of ChatShellPage as a pure function purely so it's directly
|
||||||
|
// unit-testable without rendering the whole page -- behavior is unchanged.
|
||||||
|
// Only touches the matching room; every other room's object reference (and
|
||||||
|
// therefore its unread state) is left untouched.
|
||||||
|
export function clearRoomIndicators(rooms: MyRoomItem[], roomId: string): MyRoomItem[] {
|
||||||
|
return rooms.map((r) => (r.id === roomId ? { ...r, has_unread: false, has_mention: false } : r))
|
||||||
|
}
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
import { useCallback, useEffect, useState } from 'react'
|
import { useCallback, useEffect, useMemo, useState } from 'react'
|
||||||
import { useNavigate, useParams } from 'react-router-dom'
|
import { useNavigate, useParams } from 'react-router-dom'
|
||||||
import { NetworkError } from '../api/client'
|
import { NetworkError } from '../api/client'
|
||||||
import { listMyRooms, listRoomMembers } from '../api/rooms'
|
import { listMyRooms, listRoomMembers } from '../api/rooms'
|
||||||
@@ -12,7 +12,9 @@ import { Sidebar } from '../components/Sidebar'
|
|||||||
import { TopBar } from '../components/TopBar'
|
import { TopBar } from '../components/TopBar'
|
||||||
import { useAuth } from '../context/AuthContext'
|
import { useAuth } from '../context/AuthContext'
|
||||||
import { useChatSocketContext } from '../context/ChatSocketContext'
|
import { useChatSocketContext } from '../context/ChatSocketContext'
|
||||||
|
import { useDesktopRoomBridge } from '../hooks/useDesktopRoomBridge'
|
||||||
import { MOBILE_BREAKPOINT, useWindowWidth } from '../hooks/useWindowWidth'
|
import { MOBILE_BREAKPOINT, useWindowWidth } from '../hooks/useWindowWidth'
|
||||||
|
import { clearRoomIndicators } from '../lib/roomIndicators'
|
||||||
import type { MyRoomItem, RoomMember } from '../types'
|
import type { MyRoomItem, RoomMember } from '../types'
|
||||||
import './ChatShellPage.css'
|
import './ChatShellPage.css'
|
||||||
|
|
||||||
@@ -60,10 +62,13 @@ export function ChatShellPage() {
|
|||||||
|
|
||||||
const socket = useChatSocketContext()
|
const socket = useChatSocketContext()
|
||||||
|
|
||||||
const clearRoomIndicators = useCallback((id: string) => {
|
const handleRoomRead = useCallback((id: string) => {
|
||||||
setRooms((prev) => prev.map((r) => (r.id === id ? { ...r, has_unread: false, has_mention: false } : r)))
|
setRooms((prev) => clearRoomIndicators(prev, id))
|
||||||
}, [])
|
}, [])
|
||||||
|
|
||||||
|
const unreadRoomCount = useMemo(() => rooms.filter((r) => r.has_unread).length, [rooms])
|
||||||
|
useDesktopRoomBridge(activeRoom?.id, unreadRoomCount)
|
||||||
|
|
||||||
useEffect(
|
useEffect(
|
||||||
() =>
|
() =>
|
||||||
socket.subscribe((envelope) => {
|
socket.subscribe((envelope) => {
|
||||||
@@ -134,7 +139,7 @@ export function ChatShellPage() {
|
|||||||
onToggleInfo={() => setInfoOpen((v) => !v)}
|
onToggleInfo={() => setInfoOpen((v) => !v)}
|
||||||
infoOpen={infoOpen}
|
infoOpen={infoOpen}
|
||||||
socket={socket}
|
socket={socket}
|
||||||
onRoomRead={clearRoomIndicators}
|
onRoomRead={handleRoomRead}
|
||||||
/>
|
/>
|
||||||
) : (
|
) : (
|
||||||
!isMobile && (
|
!isMobile && (
|
||||||
|
|||||||
@@ -10,6 +10,13 @@ interface UseChatSocketOptions {
|
|||||||
const RECONNECT_BASE_DELAY_MS = 1000
|
const RECONNECT_BASE_DELAY_MS = 1000
|
||||||
const RECONNECT_MAX_DELAY_MS = 30000
|
const RECONNECT_MAX_DELAY_MS = 30000
|
||||||
|
|
||||||
|
// #76: the backend pings every 30s (WS_PING_INTERVAL_SECONDS in chat.py) --
|
||||||
|
// this needs to comfortably tolerate one missed ping plus ordinary network
|
||||||
|
// jitter before declaring the connection dead, not fire on the very first
|
||||||
|
// late beat.
|
||||||
|
const ZOMBIE_TIMEOUT_MS = 65000
|
||||||
|
const ZOMBIE_CHECK_INTERVAL_MS = 10000
|
||||||
|
|
||||||
const desktopMode = isDesktopNotificationsSupported()
|
const desktopMode = isDesktopNotificationsSupported()
|
||||||
|
|
||||||
// Room join/leave (live message delivery) depends only on visibility --
|
// Room join/leave (live message delivery) depends only on visibility --
|
||||||
@@ -51,6 +58,16 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
// live WebSocket delivery the user can't actually see will do the job.
|
// live WebSocket delivery the user can't actually see will do the job.
|
||||||
const desiredRoomsRef = useRef(new Set<string>())
|
const desiredRoomsRef = useRef(new Set<string>())
|
||||||
const isVisibleRef = useRef(isVisible())
|
const isVisibleRef = useRef(isVisible())
|
||||||
|
// #76: last time *anything* arrived on the socket (a real message, a
|
||||||
|
// "ping", doesn't matter which) -- the only signal that a connection
|
||||||
|
// sitting at readyState OPEN is actually still alive rather than a
|
||||||
|
// zombie. Without this, a connection that dies without ever sending a
|
||||||
|
// close frame (laptop sleep, a NAT silently dropping an idle mapping)
|
||||||
|
// looks identical to a healthy-but-quiet one: onclose never fires, so the
|
||||||
|
// already-correct reconnect+rejoin logic below never runs, and new
|
||||||
|
// messages just stop arriving until something unrelated (e.g. switching
|
||||||
|
// rooms, which refetches history over plain REST) papers over it.
|
||||||
|
const lastActivityRef = useRef(Date.now())
|
||||||
|
|
||||||
const sendRoomFrame = useCallback((type: 'join' | 'leave', roomId: string) => {
|
const sendRoomFrame = useCallback((type: 'join' | 'leave', roomId: string) => {
|
||||||
const ws = socketRef.current
|
const ws = socketRef.current
|
||||||
@@ -99,6 +116,7 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
if (hasConnectedBefore) checkForUpdate()
|
if (hasConnectedBefore) checkForUpdate()
|
||||||
hasConnectedBefore = true
|
hasConnectedBefore = true
|
||||||
reconnectDelay = RECONNECT_BASE_DELAY_MS
|
reconnectDelay = RECONNECT_BASE_DELAY_MS
|
||||||
|
lastActivityRef.current = Date.now()
|
||||||
setConnected(true)
|
setConnected(true)
|
||||||
// Re-join whatever rooms were joined before a reconnect -- the
|
// Re-join whatever rooms were joined before a reconnect -- the
|
||||||
// server has no memory of a dropped connection's prior state. Only
|
// server has no memory of a dropped connection's prior state. Only
|
||||||
@@ -122,7 +140,17 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
|
|
||||||
ws.onmessage = (event) => {
|
ws.onmessage = (event) => {
|
||||||
if (socketRef.current !== ws) return
|
if (socketRef.current !== ws) return
|
||||||
const envelope = JSON.parse(event.data) as ServerEnvelope
|
lastActivityRef.current = Date.now()
|
||||||
|
const parsed = JSON.parse(event.data)
|
||||||
|
// #76: the server's heartbeat -- purely transport-level, not a
|
||||||
|
// domain event, so it's answered and swallowed right here instead
|
||||||
|
// of being forwarded to subscribers (ServerEnvelope's own type
|
||||||
|
// doesn't include it for exactly that reason).
|
||||||
|
if (parsed.type === 'ping') {
|
||||||
|
ws.send(JSON.stringify({ type: 'pong' }))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
const envelope = parsed as ServerEnvelope
|
||||||
for (const handler of subscribersRef.current) handler(envelope)
|
for (const handler of subscribersRef.current) handler(envelope)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -143,10 +171,9 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
if (stopped) return
|
if (stopped) return
|
||||||
|
|
||||||
// Unexpected close -- a deploy restarting the app server, a brief
|
// Unexpected close -- a deploy restarting the app server, a brief
|
||||||
// network blip, or (absent any app-level ping/pong) an idle
|
// network blip, or an idle connection getting recycled by a reverse
|
||||||
// connection getting recycled by a reverse proxy in front of it.
|
// proxy in front of it. Retry with exponential backoff instead of
|
||||||
// Retry with exponential backoff instead of leaving the chat
|
// leaving the chat silently dead until the user manually reloads.
|
||||||
// silently dead until the user manually reloads.
|
|
||||||
reconnectTimer = setTimeout(connect, reconnectDelay)
|
reconnectTimer = setTimeout(connect, reconnectDelay)
|
||||||
reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_DELAY_MS)
|
reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_DELAY_MS)
|
||||||
}
|
}
|
||||||
@@ -154,9 +181,21 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
|
|
||||||
connect()
|
connect()
|
||||||
|
|
||||||
|
// #76: catches the case onclose can't -- a connection that dies without
|
||||||
|
// ever sending a close frame at all (readyState stays OPEN forever) is
|
||||||
|
// otherwise invisible to this hook. Forcing a close here just routes it
|
||||||
|
// through the exact same onclose/reconnect/rejoin path as a normal
|
||||||
|
// disconnect, rather than needing separate recovery logic of its own.
|
||||||
|
const zombieCheckTimer = setInterval(() => {
|
||||||
|
if (Date.now() - lastActivityRef.current > ZOMBIE_TIMEOUT_MS) {
|
||||||
|
socketRef.current?.close()
|
||||||
|
}
|
||||||
|
}, ZOMBIE_CHECK_INTERVAL_MS)
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
stopped = true
|
stopped = true
|
||||||
if (reconnectTimer) clearTimeout(reconnectTimer)
|
if (reconnectTimer) clearTimeout(reconnectTimer)
|
||||||
|
clearInterval(zombieCheckTimer)
|
||||||
socketRef.current?.close()
|
socketRef.current?.close()
|
||||||
socketRef.current = null
|
socketRef.current = null
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,13 @@
|
|||||||
|
import { defineConfig } from 'vitest/config'
|
||||||
|
import react from '@vitejs/plugin-react'
|
||||||
|
|
||||||
|
// Deliberately separate from vite.config.ts, not an extension of it -- that
|
||||||
|
// file's VitePWA plugin does real build-time work (service worker
|
||||||
|
// injection, manifest generation) that has nothing to do with running unit
|
||||||
|
// tests and would just slow every test run down for no benefit.
|
||||||
|
export default defineConfig({
|
||||||
|
plugins: [react()],
|
||||||
|
test: {
|
||||||
|
environment: 'jsdom',
|
||||||
|
},
|
||||||
|
})
|
||||||
Reference in new issue
Block a user