refac

backend/open_webui/routers/channels.py

@@ -8,6 +8,8 @@ from pydantic import BaseModel
 
 
 from open_webui.socket.main import (
+    emit_to_users,
+    enter_room_for_users,
     sio,
     get_user_ids_from_room,
 )
@@ -156,6 +158,20 @@ async def get_dm_channel_by_user_id(
     try:
         existing_channel = Channels.get_dm_channel_by_user_ids([user.id, user_id])
         if existing_channel:
+            participant_ids = [
+                member.user_id
+                for member in Channels.get_members_by_channel_id(existing_channel.id)
+            ]
+
+            await emit_to_users(
+                "events:channel",
+                {"data": {"type": "channel:created"}},
+                participant_ids,
+            )
+            await enter_room_for_users(
+                f"channel:{existing_channel.id}", participant_ids
+            )
+
             Channels.update_member_active_status(existing_channel.id, user.id, True)
             return ChannelModel(**existing_channel.model_dump())
 
@@ -167,7 +183,23 @@ async def get_dm_channel_by_user_id(
             ),
             user.id,
         )
-        return ChannelModel(**channel.model_dump())
+
+        if channel:
+            participant_ids = [
+                member.user_id
+                for member in Channels.get_members_by_channel_id(channel.id)
+            ]
+
+            await emit_to_users(
+                "events:channel",
+                {"data": {"type": "channel:created"}},
+                participant_ids,
+            )
+            await enter_room_for_users(f"channel:{channel.id}", participant_ids)
+
+            return ChannelModel(**channel.model_dump())
+        else:
+            raise Exception("Error creating channel")
     except Exception as e:
         log.exception(e)
         raise HTTPException(
@@ -205,11 +237,42 @@ async def create_new_channel(
                 [user.id, *form_data.user_ids]
             )
             if existing_channel:
+                participant_ids = [
+                    member.user_id
+                    for member in Channels.get_members_by_channel_id(
+                        existing_channel.id
+                    )
+                ]
+                await emit_to_users(
+                    "events:channel",
+                    {"data": {"type": "channel:created"}},
+                    participant_ids,
+                )
+                await enter_room_for_users(
+                    f"channel:{existing_channel.id}", participant_ids
+                )
+
                 Channels.update_member_active_status(existing_channel.id, user.id, True)
                 return ChannelModel(**existing_channel.model_dump())
 
         channel = Channels.insert_new_channel(form_data, user.id)
-        return ChannelModel(**channel.model_dump())
+
+        if channel:
+            participant_ids = [
+                member.user_id
+                for member in Channels.get_members_by_channel_id(channel.id)
+            ]
+
+            await emit_to_users(
+                "events:channel",
+                {"data": {"type": "channel:created"}},
+                participant_ids,
+            )
+            await enter_room_for_users(f"channel:{channel.id}", participant_ids)
+
+            return ChannelModel(**channel.model_dump())
+        else:
+            raise Exception("Error creating channel")
     except Exception as e:
         log.exception(e)
         raise HTTPException(

backend/open_webui/socket/main.py

@@ -253,6 +253,38 @@ def get_user_ids_from_room(room):
     return active_user_ids
 
 
+async def emit_to_users(event: str, data: dict, user_ids: list[str]):
+    """
+    Send a message to specific users using their user:{id} rooms.
+
+    Args:
+        event (str): The event name to emit.
+        data (dict): The payload/data to send.
+        user_ids (list[str]): The target users' IDs.
+    """
+    try:
+        for user_id in user_ids:
+            await sio.emit(event, data, room=f"user:{user_id}")
+    except Exception as e:
+        log.debug(f"Failed to emit event {event} to users {user_ids}: {e}")
+
+
+async def enter_room_for_users(room: str, user_ids: list[str]):
+    """
+    Make all sessions of a user join a specific room.
+    Args:
+        room (str): The room to join.
+        user_ids (list[str]): The target user's IDs.
+    """
+    try:
+        for user_id in user_ids:
+            session_ids = get_session_ids_from_room(f"user:{user_id}")
+            for sid in session_ids:
+                await sio.enter_room(sid, room)
+    except Exception as e:
+        log.debug(f"Failed to make users {user_ids} join room {room}: {e}")
+
+
 @sio.on("usage")
 async def usage(sid, data):
     if sid in SESSION_POOL: