diff --git a/backend/app/services/message_events.py b/backend/app/services/message_events.py index ac026c7..f2b0c32 100644 --- a/backend/app/services/message_events.py +++ b/backend/app/services/message_events.py @@ -143,13 +143,24 @@ async def broadcast_new_message( # A no-op for a regular room (hidden_at is only ever set on a DM's # membership row -- see RoomMembership.hidden_at) -- new activity # un-hiding a DM someone closed matches find_or_create_dm's own - # un-hide-on-reopen behavior. - await db.execute( + # un-hide-on-reopen behavior. `.returning` so we know exactly who was + # un-hidden -- their client needs the same room_added signal a brand + # new DM does (see broadcast_room_added's docstring): the room wasn't + # in their already-loaded room list at all, so unread_update's plain + # setRooms(prev => prev.map(...)) can't make it reappear -- there's + # nothing in `prev` for it to match. + unhidden_result = await db.execute( update(RoomMembership) .where(RoomMembership.room_id == room_id, RoomMembership.hidden_at.is_not(None)) .values(hidden_at=None) + .returning(RoomMembership.user_id) ) + unhidden_user_ids = list(unhidden_result.scalars().all()) await db.commit() + for unhidden_user_id in unhidden_user_ids: + await broadcaster.publish_to_user( + unhidden_user_id, {"type": "room_added", "room_id": str(room_id)} + ) await _notify_offline_members(db, broadcaster, presence, room_id, sender, message) await dispatch_event(db, "message.created", room_id, payload) _maybe_fetch_link_preview(broadcaster, room_id, message) diff --git a/backend/app/ws/chat.py b/backend/app/ws/chat.py index b62b991..206874a 100644 --- a/backend/app/ws/chat.py +++ b/backend/app/ws/chat.py @@ -89,157 +89,177 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db) # visible. if await global_presence.connect(user.id): await broadcast_member_updated(db, broadcaster, user.id) + # This session is shared for the connection's entire lifetime (which can + # be hours) -- SQLAlchemy opens a transaction implicitly on first use, + # and every read above (the auth lookup, broadcast_member_updated's own + # query) leaves it open with nothing to ever close it otherwise. Left + # uncommitted, that transaction sits "idle in transaction" holding locks + # for as long as the socket stays open -- confirmed in production + # blocking unrelated schema migrations on the same tables for 30+ + # minutes. Committing here, and again after every frame below, means + # the connection is never sitting on an open transaction while merely + # waiting for the next one. + await db.commit() try: while True: raw = await websocket.receive_json() try: - envelope = ClientEnvelope.model_validate(raw) - except ValidationError: - await websocket.send_json({"type": "error", "detail": "Malformed message"}) - continue - - if envelope.type == "join": - if envelope.room_id is None: - await websocket.send_json({"type": "error", "detail": "room_id required"}) - continue - if not await _is_room_member(db, envelope.room_id, user.id): - await websocket.send_json( - {"type": "error", "detail": "Not a member of this room"} - ) - continue - manager.join(envelope.room_id, websocket) - await presence.join(envelope.room_id, user.id) - joined_rooms.add(envelope.room_id) - await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)}) - - elif envelope.type == "leave": - if envelope.room_id is None: - await websocket.send_json({"type": "error", "detail": "room_id required"}) - continue - manager.leave(envelope.room_id, websocket) - await presence.leave(envelope.room_id, user.id) - joined_rooms.discard(envelope.room_id) - - elif envelope.type == "message": - if envelope.room_id is None or ( - not envelope.content - and envelope.image_id is None - and envelope.file_id is None - ): - await websocket.send_json( - { - "type": "error", - "detail": "room_id and content or image_id or file_id required", - } - ) - continue - if _missing_scope(api_token, "write:messages"): - await websocket.send_json( - {"type": "error", "detail": "Token missing required scope: write:messages"} - ) - continue - if envelope.room_id not in joined_rooms or not await _is_room_member( - db, envelope.room_id, user.id - ): - await websocket.send_json( - {"type": "error", "detail": "Not a member of this room"} - ) - continue - image_id = None - if envelope.image_id is not None: - image = await db.get(MessageImage, envelope.image_id) - if image is None or image.room_id != envelope.room_id: - await websocket.send_json({"type": "error", "detail": "Invalid image"}) - continue - image_id = image.id - file_id = None - if envelope.file_id is not None: - message_file = await db.get(MessageFile, envelope.file_id) - if message_file is None or message_file.room_id != envelope.room_id: - await websocket.send_json({"type": "error", "detail": "Invalid file"}) - continue - file_id = message_file.id - message = await create_message( - db, envelope.room_id, user.id, envelope.content, image_id, file_id - ) - # Sending implies having seen the room as of now -- without - # this, GET /rooms/mine would show the sender's own room as - # unread the instant they send into it (last_read_at isn't - # otherwise bumped until the frontend's own message echo - # triggers a mark-read call, which is a real but avoidable - # race). Deliberately not done in create_message() itself: - # the incoming-webhook path also calls it, and a webhook's - # attributed sender may not actually be watching. - await mark_room_read(db, envelope.room_id, user.id) - await broadcast_new_message(db, broadcaster, presence, envelope.room_id, message, user) - - elif envelope.type == "edit": - if envelope.room_id is None or envelope.message_id is None or not envelope.content: - await websocket.send_json( - {"type": "error", "detail": "room_id, message_id, and content required"} - ) - continue - if _missing_scope(api_token, "write:messages"): - await websocket.send_json( - {"type": "error", "detail": "Token missing required scope: write:messages"} - ) - continue - if envelope.room_id not in joined_rooms or not await _is_room_member( - db, envelope.room_id, user.id - ): - await websocket.send_json( - {"type": "error", "detail": "Not a member of this room"} - ) - continue try: - message = await edit_message(db, envelope.message_id, user.id, envelope.content) - except MessageNotFoundError: - await websocket.send_json({"type": "error", "detail": "Message not found"}) + envelope = ClientEnvelope.model_validate(raw) + except ValidationError: + await websocket.send_json({"type": "error", "detail": "Malformed message"}) continue - except NotMessageAuthorError: - await websocket.send_json( - {"type": "error", "detail": "You can only edit your own messages"} - ) - continue - await broadcast_message_update(db, broadcaster, envelope.room_id, message) - elif envelope.type == "reaction": - if ( - envelope.room_id is None - or envelope.message_id is None - or not envelope.emoji - or len(envelope.emoji) > 8 - ): - await websocket.send_json( - {"type": "error", "detail": "room_id, message_id, and emoji required"} - ) - continue - if _missing_scope(api_token, "write:messages"): - await websocket.send_json( - {"type": "error", "detail": "Token missing required scope: write:messages"} - ) - continue - if envelope.room_id not in joined_rooms or not await _is_room_member( - db, envelope.room_id, user.id - ): - await websocket.send_json( - {"type": "error", "detail": "Not a member of this room"} - ) - continue - target_message = await db.get(Message, envelope.message_id) - if target_message is None or target_message.room_id != envelope.room_id: - await websocket.send_json({"type": "error", "detail": "Message not found"}) - continue - reactions = await toggle_reaction(db, envelope.message_id, user.id, envelope.emoji) - await broadcast_reaction_update( - broadcaster, envelope.room_id, envelope.message_id, reactions - ) + if envelope.type == "join": + if envelope.room_id is None: + await websocket.send_json({"type": "error", "detail": "room_id required"}) + continue + if not await _is_room_member(db, envelope.room_id, user.id): + await websocket.send_json( + {"type": "error", "detail": "Not a member of this room"} + ) + continue + manager.join(envelope.room_id, websocket) + await presence.join(envelope.room_id, user.id) + joined_rooms.add(envelope.room_id) + await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)}) - else: - await websocket.send_json( - {"type": "error", "detail": f"Unknown message type: {envelope.type}"} - ) + elif envelope.type == "leave": + if envelope.room_id is None: + await websocket.send_json({"type": "error", "detail": "room_id required"}) + continue + manager.leave(envelope.room_id, websocket) + await presence.leave(envelope.room_id, user.id) + joined_rooms.discard(envelope.room_id) + + elif envelope.type == "message": + if envelope.room_id is None or ( + not envelope.content + and envelope.image_id is None + and envelope.file_id is None + ): + await websocket.send_json( + { + "type": "error", + "detail": "room_id and content or image_id or file_id required", + } + ) + continue + if _missing_scope(api_token, "write:messages"): + await websocket.send_json( + {"type": "error", "detail": "Token missing required scope: write:messages"} + ) + continue + if envelope.room_id not in joined_rooms or not await _is_room_member( + db, envelope.room_id, user.id + ): + await websocket.send_json( + {"type": "error", "detail": "Not a member of this room"} + ) + continue + image_id = None + if envelope.image_id is not None: + image = await db.get(MessageImage, envelope.image_id) + if image is None or image.room_id != envelope.room_id: + await websocket.send_json({"type": "error", "detail": "Invalid image"}) + continue + image_id = image.id + file_id = None + if envelope.file_id is not None: + message_file = await db.get(MessageFile, envelope.file_id) + if message_file is None or message_file.room_id != envelope.room_id: + await websocket.send_json({"type": "error", "detail": "Invalid file"}) + continue + file_id = message_file.id + message = await create_message( + db, envelope.room_id, user.id, envelope.content, image_id, file_id + ) + # Sending implies having seen the room as of now -- + # without this, GET /rooms/mine would show the sender's + # own room as unread the instant they send into it + # (last_read_at isn't otherwise bumped until the + # frontend's own message echo triggers a mark-read + # call, which is a real but avoidable race). + # Deliberately not done in create_message() itself: the + # incoming-webhook path also calls it, and a webhook's + # attributed sender may not actually be watching. + await mark_room_read(db, envelope.room_id, user.id) + await broadcast_new_message(db, broadcaster, presence, envelope.room_id, message, user) + + elif envelope.type == "edit": + if envelope.room_id is None or envelope.message_id is None or not envelope.content: + await websocket.send_json( + {"type": "error", "detail": "room_id, message_id, and content required"} + ) + continue + if _missing_scope(api_token, "write:messages"): + await websocket.send_json( + {"type": "error", "detail": "Token missing required scope: write:messages"} + ) + continue + if envelope.room_id not in joined_rooms or not await _is_room_member( + db, envelope.room_id, user.id + ): + await websocket.send_json( + {"type": "error", "detail": "Not a member of this room"} + ) + continue + try: + message = await edit_message(db, envelope.message_id, user.id, envelope.content) + except MessageNotFoundError: + await websocket.send_json({"type": "error", "detail": "Message not found"}) + continue + except NotMessageAuthorError: + await websocket.send_json( + {"type": "error", "detail": "You can only edit your own messages"} + ) + continue + await broadcast_message_update(db, broadcaster, envelope.room_id, message) + + elif envelope.type == "reaction": + if ( + envelope.room_id is None + or envelope.message_id is None + or not envelope.emoji + or len(envelope.emoji) > 8 + ): + await websocket.send_json( + {"type": "error", "detail": "room_id, message_id, and emoji required"} + ) + continue + if _missing_scope(api_token, "write:messages"): + await websocket.send_json( + {"type": "error", "detail": "Token missing required scope: write:messages"} + ) + continue + if envelope.room_id not in joined_rooms or not await _is_room_member( + db, envelope.room_id, user.id + ): + await websocket.send_json( + {"type": "error", "detail": "Not a member of this room"} + ) + continue + target_message = await db.get(Message, envelope.message_id) + if target_message is None or target_message.room_id != envelope.room_id: + await websocket.send_json({"type": "error", "detail": "Message not found"}) + continue + reactions = await toggle_reaction(db, envelope.message_id, user.id, envelope.emoji) + await broadcast_reaction_update( + broadcaster, envelope.room_id, envelope.message_id, reactions + ) + + else: + await websocket.send_json( + {"type": "error", "detail": f"Unknown message type: {envelope.type}"} + ) + finally: + # See the comment on the pre-loop commit above -- guarantees + # every single frame, on every exit path (including the + # many `continue`s above, which still run a `finally` + # before actually looping), leaves nothing open while this + # blocks on the next receive_json(). + await db.commit() except WebSocketDisconnect: pass diff --git a/backend/tests/test_broadcast.py b/backend/tests/test_broadcast.py index be73a1a..87fdaf1 100644 --- a/backend/tests/test_broadcast.py +++ b/backend/tests/test_broadcast.py @@ -168,6 +168,43 @@ def test_start_dm_notifies_other_participant_via_websocket(ws_client_factory): 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