fix(telegram-bot): OOM due to incorrect chat messages query filter

Co-authored-by: Makito <5277268+sumimakito@users.noreply.github.com>

services/telegram-bot/src/bots/telegram/index.ts

@@ -50,20 +50,19 @@ async function handleLoopStep(state: BotSelf, msgs?: LLMMessage[], chatId?: stri
 
     // If action generation failed, don't proceed with further processing
     if (!action || !action.action) {
-      state.logger.log('No valid action returned. Skipping further processing.')
+      state.logger.withField('action', action).log('No valid action returned. Skipping further processing.')
       return
     }
 
     switch (action.action) {
       case 'readMessages':
         if (Object.keys(state.unreadMessages).length === 0) {
-          state.logger.log('No unread messages - deleting all unread messages')
+          state.logger.withField('action', action).log('No unread messages - deleting all unread messages')
           state.unreadMessages = {}
           break
         }
         if (action.chatId == null) {
-          state.logger.log('No group ID - deleting all unread messages')
-          state.unreadMessages = {}
+          state.logger.withField('action', action).warn('No group ID - deleting all unread messages')
           break
         }
 
@@ -95,19 +94,19 @@ async function handleLoopStep(state: BotSelf, msgs?: LLMMessage[], chatId?: stri
           })
 
           if (shouldInterrupt) {
-            state.logger.log(`Interrupting message processing for chat ${action.chatId} - new messages deemed more important`)
+            state.logger.withField('action', action).log(`Interrupting message processing for chat - new messages deemed more important`)
             return () => handleLoopStep(state)
           }
           else {
-            state.logger.log(`Continuing current processing despite new messages in chat ${action.chatId}`)
+            state.logger.withField('action', action).log(`Continuing current processing despite new messages in chat`)
           }
         }
         if (!Array.isArray(unreadMessagesForThisChat)) {
-          state.logger.log(`Unread messages for group ${action.chatId} is not an array - converting to array`)
+          state.logger.withField('action', action).log(`Unread messages for group is not an array - converting to array`)
           unreadMessagesForThisChat = []
         }
         if (unreadMessagesForThisChat.length === 0) {
-          state.logger.log(`No unread messages for group ${action.chatId} - deleting`)
+          state.logger.withField('action', action).log(`No unread messages for group - deleting`)
           delete state.unreadMessages[action.chatId]
           break
         }

services/telegram-bot/src/models/chat-message.ts

@@ -5,7 +5,7 @@ import type { Message, UserFromGetMe } from 'grammy/types'
 import { env } from 'node:process'
 import { useLogg } from '@guiiai/logg'
 import { embed } from '@xsai/embed'
-import { and, cosineDistance, desc, eq, gt, inArray, lt, notInArray, sql } from 'drizzle-orm'
+import { and, cosineDistance, desc, eq, gt, inArray, lt, ne, notInArray, sql } from 'drizzle-orm'
 
 import { useDrizzle } from '../db'
 import { chatMessagesTable } from '../db/schema'
@@ -55,7 +55,7 @@ export async function recordMessage(botInfo: UserFromGetMe, message: Message) {
 
   switch (env.EMBEDDING_DIMENSION) {
     case '1536':
-      values.content_vector_1536 = embedding?.embedding
+      values.content_vector_1536 = embedding.embedding
       break
     case '1024':
       values.content_vector_1024 = embedding.embedding
@@ -250,6 +250,10 @@ export async function findMessagesByIDs(messageIds: string[]) {
     .select()
     .from(chatMessagesTable)
     .where(
-      inArray(chatMessagesTable.platform_message_id, messageIds),
+      and(
+        inArray(chatMessagesTable.platform_message_id, messageIds),
+        eq(chatMessagesTable.platform, 'telegram'),
+        ne(chatMessagesTable.platform_message_id, ''),
+      ),
     )
 }