fix racing issues when sending message

app/chat_channel.py

@@ -25,10 +25,9 @@ class ChatChannel:
             base_url=settings.gemini_base_url,
             model=settings.gemini_models_list[0] if settings.gemini_models_list else "gemini-2.5-flash"
         )
-        self.facebook = FacebookClient(settings.facebook_app_secret, page_id=None, page_token=None, sender_id=None)
-        self.reranker = Reranker(facebook_client=self.facebook)
+        self.reranker = Reranker()
         self.embedder = EmbeddingClient()
-        self.message_processor = MessageProcessor(self)
+        self.conversations = {}  # sender_id -> MessageProcessor
 
     def get_page_token(self, force_refresh=False):
         if self.page_token is None or force_refresh:
@@ -36,4 +35,9 @@ class ChatChannel:
         return self.page_token
 
     def invalidate_page_token(self):
-        self.page_token = None 
+        self.page_token = None
+
+    def get_or_create_conversation(self, sender_id: str):
+        if sender_id not in self.conversations:
+            self.conversations[sender_id] = MessageProcessor(self, sender_id)
+        return self.conversations[sender_id] 

app/main.py

@@ -181,13 +181,12 @@ async def webhook(request: Request):
         if not message_data:
             return {"status": "ok"}
 
-        # --- Refactor: Lấy page_id, page_token, channel, gọi message_processor ---
+        # --- Refactor: Lấy page_id, sender_id, channel, conversation, gọi message_processor ---
         page_id = message_data.get("page_id")
-        # Không lấy page_token ở đây nữa
-        # Lấy hoặc tạo ChatChannel
+        sender_id = message_data.get("sender_id")
         channel = channel_manager.get_or_create_channel("facebook", page_id)
-        # Gọi message_processor xử lý message
-        await channel.message_processor.process_message(message_data)
+        conversation = channel.get_or_create_conversation(sender_id)
+        await conversation.process_message(message_data)
         return {"status": "ok"}
     except Exception as e:
         logger.error(f"Error processing webhook: {e}\nTraceback: {traceback.format_exc()}")

app/message_processor.py

@@ -2,12 +2,21 @@ from typing import Dict, Any, List
 import asyncio
 import traceback
 from loguru import logger
-from .constants import SUMMARY_STATUS_MESSAGES, PROCESSING_STATUS_MESSAGES, FOUND_REGULATIONS_MESSAGES
+from .constants import SUMMARY_STATUS_MESSAGES, PROCESSING_STATUS_MESSAGES, FOUND_REGULATIONS_MESSAGES, BATCH_STATUS_MESSAGES
 from .utils import get_random_message
+from .facebook import FacebookClient
 
 class MessageProcessor:
-    def __init__(self, channel):
+    def __init__(self, channel, sender_id):
         self.channel = channel
+        self.sender_id = sender_id
+        # FacebookClient riêng cho từng conversation
+        self.facebook = FacebookClient(
+            app_secret=channel.llm.gemini_client.limit_manager.get_settings().facebook_app_secret,
+            page_id=channel.page_id,
+            page_token=channel.get_page_token(),
+            sender_id=sender_id
+        )
 
     async def process_message(self, message_data: Dict[str, Any]):
         # Refactor logic từ main.py vào đây
@@ -120,7 +129,7 @@ class MessageProcessor:
             return
         # Gửi message Facebook, nếu lỗi token expired thì invalidate và thử lại một lần
         try:
-            await self.channel.facebook.send_message(message=get_random_message(PROCESSING_STATUS_MESSAGES))
+            await self.facebook.send_message(message=get_random_message(PROCESSING_STATUS_MESSAGES))
         except Exception as e:
             if "expired" in str(e).lower():
                 logger.warning("[FACEBOOK] Token expired, invalidate and refresh")
@@ -181,7 +190,7 @@ class MessageProcessor:
         # 6. Gửi response và cập nhật final state
         # Replace all occurrences of '**' with '*' before sending
         response_to_send = response.replace('**', '*') if isinstance(response, str) else response
-        await self.channel.facebook.send_message(message=response_to_send)
+        await self.facebook.send_message(message=response_to_send)
         if hasattr(self.channel, 'sheets'):
             await loop.run_in_executor(None, lambda: self.channel.sheets.log_conversation(**conv))
         return
@@ -256,10 +265,17 @@ class MessageProcessor:
     async def format_search_results(self, question: str, matches: List[Dict[str, Any]], page_token: str, sender_id: str) -> str:
         if not matches:
             return "Không tìm thấy kết quả phù hợp."
-        await self.channel.facebook.send_message(message=get_random_message(FOUND_REGULATIONS_MESSAGES))
+        await self.facebook.send_message(message=get_random_message(FOUND_REGULATIONS_MESSAGES))
         try:
             reranked = await self.channel.reranker.rerank(question, matches, top_k=5)
             if reranked:
+                # Gửi Facebook message sau khi hoàn thành
+                if self.facebook:
+                    try:
+                        message = get_random_message(BATCH_STATUS_MESSAGES)
+                        await self.facebook.send_message(message=f"... {message} ...")
+                    except Exception as e:
+                        logger.error(f"[RERANK][FACEBOOK] Error sending batch message: {e}")
                 matches = reranked
         except Exception as e:
             logger.error(f"[RERANK] Lỗi khi rerank: {e}")
@@ -328,7 +344,7 @@ class MessageProcessor:
             "\n\nHãy trả lời ngắn gọn, dễ hiểu, trích dẫn rõ ràng thông tin từ các đoạn luật nếu cần."
             f"\n\nCâu hỏi của người dùng: {question}\n"
         )
-        await self.channel.facebook.send_message(message=f"... {get_random_message(SUMMARY_STATUS_MESSAGES)} ...")
+        await self.facebook.send_message(message=f"... {get_random_message(SUMMARY_STATUS_MESSAGES)} ...")
         try:
             answer = await self.channel.llm.generate_text(prompt)
             if answer and answer.strip():

app/reranker.py

@@ -5,11 +5,11 @@ from loguru import logger
 import asyncio
 import hashlib
 import time
-from .constants import BATCH_STATUS_MESSAGES
-from .utils import get_random_message
+# from .constants import BATCH_STATUS_MESSAGES
+# from .utils import get_random_message
 
 class Reranker:
-    def __init__(self, facebook_client=None):
+    def __init__(self):
         settings = get_settings()
         self.provider = getattr(settings, 'rerank_provider', settings.llm_provider)
         self.model = getattr(settings, 'rerank_model', settings.llm_model)
@@ -21,14 +21,11 @@ class Reranker:
         #     self.client = CohereClient(settings.cohere_api_key, model=self.model)
         else:
             raise NotImplementedError(f"Rerank provider {self.provider} not supported yet.")
-        self.facebook_client = facebook_client
-        
         # Cải thiện cache với TTL và quản lý memory
         self._rerank_cache = {}
         self._cache_ttl = 3600  # 1 giờ
         self._max_cache_size = 200  # Tăng cache size
         self._cache_timestamps = {}
-        
         # Sử dụng max_docs_to_rerank từ config
         self.max_docs_to_rerank = settings.max_docs_to_rerank
 
@@ -244,14 +241,6 @@ class Reranker:
                     doc['rerank_score'] = 0
                     scored.append(doc)
         
-        # Gửi Facebook message sau khi hoàn thành
-        if self.facebook_client:
-            try:
-                message = get_random_message(BATCH_STATUS_MESSAGES)
-                await self.facebook_client.send_message(message=f"... {message} ...")
-            except Exception as e:
-                logger.error(f"[RERANK][FACEBOOK] Error sending batch message: {e}")
-        
         # Sort theo score và trả về top_k
         scored = sorted(scored, key=lambda x: x['rerank_score'], reverse=True)
         result = scored[:top_k]