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
|
||||
# TestClient's websocket_connect-based tests actually run lifespan).
|
||||
# The background listener task genuinely needs a running event loop
|
||||
# though, so that part stays here.
|
||||
listener_task = asyncio.create_task(app.state.broadcaster.listen())
|
||||
# though, so that part stays here. run_forever() (not listen() directly)
|
||||
# -- #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
|
||||
|
||||
|
||||
@@ -1,13 +1,26 @@
|
||||
import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
|
||||
from redis.asyncio import Redis
|
||||
|
||||
from app.ws.connection_manager import ConnectionManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ROOM_CHANNEL_PREFIX = "room:"
|
||||
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:
|
||||
"""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:
|
||||
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:
|
||||
pubsub = self._redis.pubsub()
|
||||
await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
||||
@@ -43,6 +93,16 @@ class Broadcaster:
|
||||
async for message in pubsub.listen():
|
||||
if message["type"] != "pmessage":
|
||||
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"]
|
||||
payload = json.loads(message["data"])
|
||||
if channel.startswith(ROOM_CHANNEL_PREFIX):
|
||||
@@ -51,6 +111,19 @@ class Broadcaster:
|
||||
elif channel.startswith(USER_CHANNEL_PREFIX):
|
||||
user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
|
||||
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:
|
||||
# #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}*")
|
||||
with contextlib.suppress(Exception):
|
||||
await pubsub.aclose()
|
||||
+33
-1
@@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
||||
@@ -31,6 +32,18 @@ router = APIRouter(tags=["ws"])
|
||||
|
||||
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):
|
||||
type: str
|
||||
@@ -124,7 +137,18 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
|
||||
|
||||
try:
|
||||
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:
|
||||
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)
|
||||
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":
|
||||
if envelope.room_id is None:
|
||||
await websocket.send_json({"type": "error", "detail": "room_id required"})
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
import logging
|
||||
import uuid
|
||||
from collections import defaultdict
|
||||
|
||||
from fastapi import WebSocket
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ConnectionManager:
|
||||
"""Local, single-process WebSocket socket registry.
|
||||
@@ -43,8 +46,22 @@ class ConnectionManager:
|
||||
|
||||
async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
|
||||
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)
|
||||
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:
|
||||
for websocket in list(self._users.get(user_id, ())):
|
||||
try:
|
||||
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]
|
||||
name = "ds-chat"
|
||||
version = "2026.9.4"
|
||||
version = "2026.9.28"
|
||||
description = "DS Chat backend service"
|
||||
license = { text = "AGPL-3.0-or-later" }
|
||||
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]
|
||||
|
||||
|
||||
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):
|
||||
_register(ws_client, username=_unique("alice"))
|
||||
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",
|
||||
"private": true,
|
||||
"version": "2026.9.4",
|
||||
"version": "2026.9.28",
|
||||
"license": "AGPL-3.0-or-later",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"dev": "vite",
|
||||
"build": "tsc -b && vite build",
|
||||
"lint": "oxlint",
|
||||
"preview": "vite preview"
|
||||
"preview": "vite preview",
|
||||
"test": "vitest run"
|
||||
},
|
||||
"dependencies": {
|
||||
"markdown-to-jsx": "^9.10.2",
|
||||
@@ -18,13 +19,16 @@
|
||||
"vite-plugin-pwa": "^1.3.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@testing-library/react": "^16.3.3",
|
||||
"@types/node": "^24.13.3",
|
||||
"@types/react": "^19.2.17",
|
||||
"@types/react-dom": "^19.2.3",
|
||||
"@vitejs/plugin-react": "^6.0.4",
|
||||
"jsdom": "^30.1.1",
|
||||
"oxlint": "^1.75.0",
|
||||
"typescript": "~6.0.2",
|
||||
"vite": "^8.2.0",
|
||||
"vitest": "^5.0.2",
|
||||
"workbox-cacheable-response": "^7.4.1",
|
||||
"workbox-expiration": "^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 {
|
||||
dsDesktop?: {
|
||||
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
|
||||
onNotificationClick?(callback: (roomId: string) => void): () => void
|
||||
}
|
||||
@@ -38,6 +44,16 @@ export function isDesktopClickListenerSupported(): boolean {
|
||||
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 --
|
||||
// callers don't need to guard, matching the rest of this module's
|
||||
// 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 { NetworkError } from '../api/client'
|
||||
import { listMyRooms, listRoomMembers } from '../api/rooms'
|
||||
@@ -12,7 +12,9 @@ import { Sidebar } from '../components/Sidebar'
|
||||
import { TopBar } from '../components/TopBar'
|
||||
import { useAuth } from '../context/AuthContext'
|
||||
import { useChatSocketContext } from '../context/ChatSocketContext'
|
||||
import { useDesktopRoomBridge } from '../hooks/useDesktopRoomBridge'
|
||||
import { MOBILE_BREAKPOINT, useWindowWidth } from '../hooks/useWindowWidth'
|
||||
import { clearRoomIndicators } from '../lib/roomIndicators'
|
||||
import type { MyRoomItem, RoomMember } from '../types'
|
||||
import './ChatShellPage.css'
|
||||
|
||||
@@ -60,10 +62,13 @@ export function ChatShellPage() {
|
||||
|
||||
const socket = useChatSocketContext()
|
||||
|
||||
const clearRoomIndicators = useCallback((id: string) => {
|
||||
setRooms((prev) => prev.map((r) => (r.id === id ? { ...r, has_unread: false, has_mention: false } : r)))
|
||||
const handleRoomRead = useCallback((id: string) => {
|
||||
setRooms((prev) => clearRoomIndicators(prev, id))
|
||||
}, [])
|
||||
|
||||
const unreadRoomCount = useMemo(() => rooms.filter((r) => r.has_unread).length, [rooms])
|
||||
useDesktopRoomBridge(activeRoom?.id, unreadRoomCount)
|
||||
|
||||
useEffect(
|
||||
() =>
|
||||
socket.subscribe((envelope) => {
|
||||
@@ -134,7 +139,7 @@ export function ChatShellPage() {
|
||||
onToggleInfo={() => setInfoOpen((v) => !v)}
|
||||
infoOpen={infoOpen}
|
||||
socket={socket}
|
||||
onRoomRead={clearRoomIndicators}
|
||||
onRoomRead={handleRoomRead}
|
||||
/>
|
||||
) : (
|
||||
!isMobile && (
|
||||
|
||||
@@ -10,6 +10,13 @@ interface UseChatSocketOptions {
|
||||
const RECONNECT_BASE_DELAY_MS = 1000
|
||||
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()
|
||||
|
||||
// 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.
|
||||
const desiredRoomsRef = useRef(new Set<string>())
|
||||
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 ws = socketRef.current
|
||||
@@ -99,6 +116,7 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
||||
if (hasConnectedBefore) checkForUpdate()
|
||||
hasConnectedBefore = true
|
||||
reconnectDelay = RECONNECT_BASE_DELAY_MS
|
||||
lastActivityRef.current = Date.now()
|
||||
setConnected(true)
|
||||
// Re-join whatever rooms were joined before a reconnect -- the
|
||||
// server has no memory of a dropped connection's prior state. Only
|
||||
@@ -122,7 +140,17 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
||||
|
||||
ws.onmessage = (event) => {
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -143,10 +171,9 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
||||
if (stopped) return
|
||||
|
||||
// Unexpected close -- a deploy restarting the app server, a brief
|
||||
// network blip, or (absent any app-level ping/pong) an idle
|
||||
// connection getting recycled by a reverse proxy in front of it.
|
||||
// Retry with exponential backoff instead of leaving the chat
|
||||
// silently dead until the user manually reloads.
|
||||
// network blip, or an idle connection getting recycled by a reverse
|
||||
// proxy in front of it. Retry with exponential backoff instead of
|
||||
// leaving the chat silently dead until the user manually reloads.
|
||||
reconnectTimer = setTimeout(connect, reconnectDelay)
|
||||
reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_DELAY_MS)
|
||||
}
|
||||
@@ -154,9 +181,21 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
||||
|
||||
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 () => {
|
||||
stopped = true
|
||||
if (reconnectTimer) clearTimeout(reconnectTimer)
|
||||
clearInterval(zombieCheckTimer)
|
||||
socketRef.current?.close()
|
||||
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