Private
Public Access
ksmithandClaude Sonnet 5
1d9fe25410
Two related fixes: 1. A hidden DM's un-hide-on-new-message path only cleared RoomMembership.hidden_at in the DB -- it never told an already-open client to refresh. The only existing signal for that room (unread_update) does setRooms(prev => prev.map(...)), which is a no-op for a room that isn't in `prev` at all -- exactly what a hidden DM is. Now broadcasts the same room_added signal a brand new DM gets (via UPDATE ... RETURNING to know exactly who was un-hidden), reusing the fix already established for that class of bug. 2. While debugging #1's test, found the actual root cause behind the deploy-blocking migrations from earlier this session: every WebSocket connection shares one AsyncSession for its entire lifetime, and SQLAlchemy opens a transaction implicitly on first use. Nothing ever committed it -- not the initial auth lookup, not any of the several read-then-continue branches in the message loop (join/message/edit/reaction all check membership this way). A connection that's just sitting open (which for a real user can be hours) was holding that transaction open the entire time, which is exactly what blocked ALTER TABLE twice in production this session (confirmed both times via pg_stat_activity -- idle in transaction for 30+ minutes on this exact query shape). Now commits once after connection setup and once after every frame via a try/finally wrapping the whole dispatch, so no exit path (including the many `continue`s) can leave a transaction open while idling on the next receive_json(). Verified end-to-end in the browser (a hidden DM reappears in an already-open tab with zero reload when the other person messages again) and via a new WS-level test reproducing the exact scenario. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
232 lines
9.8 KiB
Python
232 lines
9.8 KiB
Python
import uuid
|
|
|
|
from app.schemas.user import UserCreate
|
|
from app.services.auth_service import register_user
|
|
|
|
|
|
def _unique(prefix: str) -> str:
|
|
return f"{prefix}-{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _recv(ws) -> dict:
|
|
"""Reads the next frame, transparently discarding member_updated
|
|
presence-change broadcasts -- another connection in the same room going
|
|
online/offline is real, expected noise these tests aren't about."""
|
|
while True:
|
|
msg = ws.receive_json()
|
|
if msg.get("type") != "member_updated":
|
|
return msg
|
|
|
|
|
|
def _fake_send_email(monkeypatch):
|
|
calls = []
|
|
|
|
async def fake(db, to, subject, body):
|
|
calls.append({"to": to, "subject": subject, "body": body})
|
|
|
|
monkeypatch.setattr("app.services.room_service.send_email", fake)
|
|
return calls
|
|
|
|
|
|
def _register_ws(ws_client, username: str) -> dict:
|
|
async def _seed():
|
|
async with ws_client.session_factory() as session:
|
|
await register_user(
|
|
session,
|
|
UserCreate(username=username, email=f"{username}@example.com", password="password123"),
|
|
)
|
|
|
|
ws_client.portal.call(_seed)
|
|
resp = ws_client.post(
|
|
"/api/auth/login", json={"username_or_email": username, "password": "password123"}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return resp.json()
|
|
|
|
|
|
def test_message_fans_out_across_instances(ws_client_factory):
|
|
# Two independent app instances -- separate ConnectionManager, separate
|
|
# Redis pubsub subscription, separate everything except the Postgres and
|
|
# Redis they're both pointed at -- the same way two app-server processes
|
|
# behind Nginx would be. Proves delivery actually crosses Redis, not
|
|
# just in-process delivery within a single ConnectionManager.
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
|
|
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
instance2.post(f"/api/rooms/{room['id']}/join")
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
bob_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert bob_ws.receive_json()["type"] == "joined"
|
|
|
|
with instance1.websocket_connect("/ws/chat") as alice_ws:
|
|
alice_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert alice_ws.receive_json()["type"] == "joined"
|
|
alice_ws.send_json(
|
|
{"type": "message", "room_id": room["id"], "content": "hi from instance 1"}
|
|
)
|
|
assert alice_ws.receive_json()["type"] == "message"
|
|
|
|
received = _recv(bob_ws)
|
|
assert received["type"] == "message"
|
|
assert received["content"] == "hi from instance 1"
|
|
assert received["username"] == alice["username"]
|
|
|
|
|
|
def test_presence_is_shared_across_instances(ws_client_factory, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr("app.services.push_service.webpush", lambda **kw: calls.append(kw))
|
|
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
|
|
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
instance2.post(f"/api/rooms/{room['id']}/join")
|
|
instance2.post(
|
|
"/api/push/subscribe",
|
|
json={
|
|
"endpoint": f"https://push.example.com/ep-{bob['id']}",
|
|
"keys": {"p256dh": "p256dh-bob", "auth": "auth-bob"},
|
|
},
|
|
)
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
bob_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert bob_ws.receive_json()["type"] == "joined"
|
|
|
|
with instance1.websocket_connect("/ws/chat") as alice_ws:
|
|
alice_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert alice_ws.receive_json()["type"] == "joined"
|
|
alice_ws.send_json({"type": "message", "room_id": room["id"], "content": "hi"})
|
|
assert alice_ws.receive_json()["type"] == "message"
|
|
|
|
# bob is connected -- just on the other instance -- so he should
|
|
# get the broadcast via Redis, not a push notification. If
|
|
# presence were still process-local (pre-phase-5 behavior) he'd
|
|
# look offline to instance1 and get a redundant push.
|
|
assert _recv(bob_ws)["type"] == "message"
|
|
|
|
# Sync barrier: the handler processes frames strictly
|
|
# sequentially, so a second (idempotent) join only acks once the
|
|
# "message" frame's full handling -- including the offline-push
|
|
# step -- has completed on instance1.
|
|
alice_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert alice_ws.receive_json()["type"] == "joined"
|
|
|
|
assert calls == []
|
|
|
|
|
|
def test_add_member_notifies_target_user_via_websocket(ws_client_factory, monkeypatch):
|
|
# Bob is only ever "connected," never "joined" -- proving the room_added
|
|
# signal reaches him on his own per-user channel, independent of (and
|
|
# necessarily before) ever joining the room's own channel, which he
|
|
# can't do until this signal tells his client the room exists at all.
|
|
_fake_send_email(monkeypatch)
|
|
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
|
|
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
resp = instance1.post(f"/api/rooms/{room['id']}/members", json={"user_id": bob["id"]})
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
received = bob_ws.receive_json()
|
|
assert received == {"type": "room_added", "room_id": room["id"]}
|
|
|
|
|
|
def test_start_dm_notifies_other_participant_via_websocket(ws_client_factory):
|
|
# A production report: bob had no idea a DM existed until he reloaded --
|
|
# find_or_create_dm was creating the room/membership correctly but never
|
|
# sending this signal, unlike every other "you're now in a room" path
|
|
# (add_member, above). Same shape as that test: bob is only ever
|
|
# "connected," never "joined," proving the signal alone is what tells
|
|
# his client the room exists at all.
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
resp = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]})
|
|
assert resp.status_code == 201, resp.text
|
|
room = resp.json()
|
|
|
|
received = bob_ws.receive_json()
|
|
assert received == {"type": "room_added", "room_id": room["id"]}
|
|
|
|
|
|
def test_new_message_notifies_recipient_who_hid_the_dm_via_websocket(ws_client_factory):
|
|
# A second production report on the same underlying gap: hiding a DM
|
|
# correctly clears out of GET /rooms/mine, but when the other person
|
|
# messages again, the *only* existing signal for that (unread_update)
|
|
# does `setRooms(prev => prev.map(...))` -- a no-op for a room that
|
|
# isn't in `prev` at all, which a hidden DM by definition isn't. Needs
|
|
# the same room_added signal a brand new DM gets, not just a DB-level
|
|
# un-hide.
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
|
|
room = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]}).json()
|
|
resp = instance2.post(f"/api/rooms/{room['id']}/hide")
|
|
assert resp.status_code == 204
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
with instance1.websocket_connect("/ws/chat") as alice_ws:
|
|
alice_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert alice_ws.receive_json()["type"] == "joined"
|
|
alice_ws.send_json({"type": "message", "room_id": room["id"], "content": "you there?"})
|
|
alice_ws.receive_json()
|
|
# Sync barrier (see test_mentions.py's identical helper): the
|
|
# message ack only proves the room-level broadcast happened,
|
|
# not that broadcast_new_message's own continuation (which
|
|
# un-hides the room and publishes room_added) has finished --
|
|
# a second frame's own ack proves that before this connection
|
|
# closes underneath it.
|
|
alice_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert alice_ws.receive_json()["type"] == "joined"
|
|
|
|
received = bob_ws.receive_json()
|
|
assert received == {"type": "room_added", "room_id": room["id"]}
|
|
|
|
|
|
def test_profile_update_notifies_room_members_via_websocket(ws_client_factory, monkeypatch):
|
|
# Only reaches clients that have the room's own channel joined --
|
|
# exactly the case where a stale avatar/display name would actually be
|
|
# visible on screen (a room the user has open right now).
|
|
_fake_send_email(monkeypatch)
|
|
|
|
instance1 = ws_client_factory()
|
|
instance2 = ws_client_factory()
|
|
|
|
alice = _register_ws(instance1, _unique("alice"))
|
|
room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
|
|
|
|
bob = _register_ws(instance2, _unique("bob"))
|
|
instance2.post(f"/api/rooms/{room['id']}/join")
|
|
|
|
with instance2.websocket_connect("/ws/chat") as bob_ws:
|
|
bob_ws.send_json({"type": "join", "room_id": room["id"]})
|
|
assert bob_ws.receive_json()["type"] == "joined"
|
|
|
|
resp = instance1.patch("/api/auth/me", json={"display_name": "Alice Updated"})
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
received = bob_ws.receive_json()
|
|
assert received == {"type": "member_updated", "room_id": room["id"], "user_id": alice["id"]}
|