perf: eliminate pre-read load in socket event emitter

Author: Algorithm5838

backend/open_webui/socket/main.py

@@ -808,25 +808,15 @@ async def __event_emitter__(event_data):
                 )
 
             elif event_type == 'message':
-                message = await asyncio.to_thread(
-                    Chats.get_message_by_id_and_message_id,
+                delta = event_data.get('data', {}).get('content', '')
+                await asyncio.to_thread(
+                    Chats.upsert_message_to_chat_by_id_and_message_id,
                     request_info['chat_id'],
                     request_info['message_id'],
+                    {},
+                    merge_fn=lambda existing: {'content': existing.get('content', '') + delta},
                 )
 
-                if message:
-                    content = message.get('content', '')
-                    content += event_data.get('data', {}).get('content', '')
-
-                    await asyncio.to_thread(
-                        Chats.upsert_message_to_chat_by_id_and_message_id,
-                        request_info['chat_id'],
-                        request_info['message_id'],
-                        {
-                            'content': content,
-                        },
-                    )
-
             elif event_type == 'replace':
                 content = event_data.get('data', {}).get('content', '')
 
@@ -840,62 +830,34 @@ async def __event_emitter__(event_data):
                 )
 
             elif event_type == 'embeds':
-                message = await asyncio.to_thread(
-                    Chats.get_message_by_id_and_message_id,
-                    request_info['chat_id'],
-                    request_info['message_id'],
-                )
-
-                embeds = event_data.get('data', {}).get('embeds', [])
-                embeds.extend(message.get('embeds', []))
-
+                new_embeds = event_data.get('data', {}).get('embeds', [])
                 await asyncio.to_thread(
                     Chats.upsert_message_to_chat_by_id_and_message_id,
                     request_info['chat_id'],
                     request_info['message_id'],
-                    {
-                        'embeds': embeds,
-                    },
+                    {},
+                    merge_fn=lambda existing: {'embeds': new_embeds + existing.get('embeds', [])},
                 )
 
             elif event_type == 'files':
-                message = await asyncio.to_thread(
-                    Chats.get_message_by_id_and_message_id,
-                    request_info['chat_id'],
-                    request_info['message_id'],
-                )
-
-                files = event_data.get('data', {}).get('files', [])
-                files.extend(message.get('files', []))
-
+                new_files = event_data.get('data', {}).get('files', [])
                 await asyncio.to_thread(
                     Chats.upsert_message_to_chat_by_id_and_message_id,
                     request_info['chat_id'],
                     request_info['message_id'],
-                    {
-                        'files': files,
-                    },
+                    {},
+                    merge_fn=lambda existing: {'files': new_files + existing.get('files', [])},
                 )
 
             elif event_type in ('source', 'citation'):
                 data = event_data.get('data', {})
                 if data.get('type') is None:
-                    message = await asyncio.to_thread(
-                        Chats.get_message_by_id_and_message_id,
-                        request_info['chat_id'],
-                        request_info['message_id'],
-                    )
-
-                    sources = message.get('sources', [])
-                    sources.append(data)
-
                     await asyncio.to_thread(
                         Chats.upsert_message_to_chat_by_id_and_message_id,
                         request_info['chat_id'],
                         request_info['message_id'],
-                        {
-                            'sources': sources,
-                        },
+                        {},
+                        merge_fn=lambda existing: {'sources': existing.get('sources', []) + [data]},
                     )
 
     if 'user_id' in request_info and 'chat_id' in request_info and 'message_id' in request_info: