Conversations migrate firebase and typesense collection (#2150)

part of #1973 

- [x] Use conversations collection for typesense
- [x] Use conversations collection for db
- [x] Add backwards compatibility for plugins_results (replaced by
apps_results in conversations collection)
- [x] Test with old endpoints
- [x] Update models in app
- [x] Deprecate unused processing memories

Deployment Plan:
- Deploy backend

Typesense
<img width="656" alt="Screenshot 2025-04-04 at 6 08 44 PM"
src="https://github.com/user-attachments/assets/77062a72-56cd-4b4f-b552-d2ed550ea4ac"
/>

Firestore conversations
<img width="540" alt="Screenshot 2025-04-05 at 1 12 33 AM"
src="https://github.com/user-attachments/assets/8d9fc0bc-639b-49cd-8c37-874a4354accc"
/>

Author: beastoin

app/lib/backend/schema/conversation.dart

@@ -132,7 +132,7 @@ class ServerConversation {
           .map((segment) => TranscriptSegment.fromJson(segment))
           .toList(),
       appResults:
-          ((json['plugins_results'] ?? []) as List<dynamic>).map((result) => AppResponse.fromJson(result)).toList(),
+          ((json['apps_results'] ?? []) as List<dynamic>).map((result) => AppResponse.fromJson(result)).toList(),
       geolocation: json['geolocation'] != null ? Geolocation.fromJson(json['geolocation']) : null,
       photos: json['photos'] != null
           ? ((json['photos'] ?? []) as List<dynamic>).map((photo) => ConversationPhoto.fromJson(photo)).toList()

app/lib/backend/schema/structured.dart

@@ -116,10 +116,10 @@ class AppResponse {
 
   AppResponse(this.content, {this.id = 0, this.appId});
 
-  toJson() => {'pluginId': appId, 'content': content};
+  toJson() => {'appId': appId, 'content': content};
 
   factory AppResponse.fromJson(Map<String, dynamic> json) {
-    return AppResponse(json['content'], appId: json['pluginId'] ?? json['plugin_id']);
+    return AppResponse(json['content'], appId: json['appId'] ?? json['app_id']);
   }
 }
 

backend/database/chat.py

@@ -76,7 +76,7 @@ def get_plugin_messages(uid: str, plugin_id: str, limit: int = 20, offset: int =
 
     # Fetch all conversations at once
     conversations = {}
-    conversations_ref = user_ref.collection('memories')
+    conversations_ref = user_ref.collection('conversations')
     doc_refs = [conversations_ref.document(str(conversation_id)) for conversation_id in conversations_id]
     docs = db.get_all(doc_refs)
     for doc in docs:
@@ -129,7 +129,7 @@ def get_messages(
 
     # Fetch all conversations at once
     conversations = {}
-    conversations_ref = user_ref.collection('memories')
+    conversations_ref = user_ref.collection('conversations')
     doc_refs = [conversations_ref.document(str(conversation_id)) for conversation_id in conversations_id]
     docs = db.get_all(doc_refs)
     for doc in docs:

backend/database/conversations.py

@@ -0,0 +1,99 @@
+# DEPRECATED: This file has been deprecated long ago
+#
+# This file is deprecated and should be removed. The code is not used anymore and is not referenced in any other file.
+# The only files that references this file are routers/processing_memories.py and utils/processing_conversations.py, which are also deprecated.
+
+from datetime import datetime
+from typing import List
+
+from google.cloud import firestore
+from google.cloud.firestore_v1 import FieldFilter
+
+from ._client import db
+
+
+def upsert_processing_conversation(uid: str, processing_conversation_data: dict):
+    user_ref = db.collection('users').document(uid)
+    processing_conversation_ref = user_ref.collection('processing_memories').document(processing_conversation_data['id'])
+    processing_conversation_ref.set(processing_conversation_data)
+
+
+def update_processing_conversation(uid: str, processing_conversation_id: str, memoy_data: dict):
+    user_ref = db.collection('users').document(uid)
+    processing_conversation_ref = user_ref.collection('processing_memories').document(processing_conversation_id)
+    processing_conversation_ref.update(memoy_data)
+
+
+def delete_processing_conversation(uid, processing_conversation_id):
+    user_ref = db.collection('users').document(uid)
+    processing_conversation_ref = user_ref.collection('processing_memories').document(processing_conversation_id)
+    processing_conversation_ref.update({'deleted': True})
+
+
+def get_processing_conversations_by_id(uid, processing_conversation_ids):
+    user_ref = db.collection('users').document(uid)
+    conversations_ref = user_ref.collection('processing_memories')
+
+    doc_refs = [conversations_ref.document(str(processing_conversation_id)) for processing_conversation_id in processing_conversation_ids]
+    docs = db.get_all(doc_refs)
+
+    conversations = []
+    for doc in docs:
+        if doc.exists:
+            conversations.append(doc.to_dict())
+    return conversations
+
+
+def get_processing_conversation_by_id(uid, processing_conversation_id):
+    conversation_ref = db.collection('users').document(uid).collection('processing_memories').document(processing_conversation_id)
+    return conversation_ref.get().to_dict()
+
+
+def get_processing_conversations(uid: str, statuses: [str] = [], filter_ids: [str] = [], limit: int = 5):
+    processing_conversations_ref = (
+        db.collection('users').document(uid).collection('processing_memories')
+    )
+    if len(statuses) > 0:
+        processing_conversations_ref = processing_conversations_ref.where(filter=FieldFilter('status', 'in', statuses))
+    if len(filter_ids) > 0:
+        processing_conversations_ref = processing_conversations_ref.where(filter=FieldFilter('id', 'in', filter_ids))
+    processing_conversations_ref = processing_conversations_ref.order_by('created_at', direction=firestore.Query.DESCENDING)
+    processing_conversations_ref = processing_conversations_ref.limit(limit)
+    return [doc.to_dict() for doc in processing_conversations_ref.stream()]
+
+
+def update_processing_conversation_segments(uid: str, id: str, segments: List[dict], capturing_to: datetime):
+    user_ref = db.collection('users').document(uid)
+    conversation_ref = user_ref.collection('processing_memories').document(id)
+    conversation_ref.update({
+        'transcript_segments': segments,
+        'capturing_to': capturing_to,
+    })
+
+
+def update_processing_conversation_status(uid: str, id: str, status: str):
+    user_ref = db.collection('users').document(uid)
+    conversation_ref = user_ref.collection('processing_memories').document(id)
+    conversation_ref.update({
+        'status': status,
+    })
+
+
+def update_audio_url(uid: str, id: str, audio_url: str):
+    user_ref = db.collection('users').document(uid)
+    conversation_ref = user_ref.collection('processing_memories').document(id)
+    conversation_ref.update({
+        'audio_url': audio_url,
+    })
+
+
+def get_last(uid: str):
+    processing_conversations_ref = (
+        db.collection('users').document(uid).collection('processing_memories')
+    )
+    processing_conversations_ref = processing_conversations_ref.order_by('created_at', direction=firestore.Query.DESCENDING)
+    processing_conversations_ref = processing_conversations_ref.limit(1)
+    docs = [doc.to_dict() for doc in processing_conversations_ref.stream()]
+    if len(docs) > 0:
+        return docs[0]
+    return None

backend/database/processing_conversations.py

@@ -6,7 +6,7 @@
 
 from modal import Image, App, asgi_app, Secret
 from routers import workflow, chat, firmware, plugins, memories_deprecated, transcribe, notifications, \
-    speech_profile, agents, facts_deprecated, users, processing_memories, trends, sdcard, sync, apps, custom_auth, \
+    speech_profile, agents, facts_deprecated, users, processing_conversations, trends, sdcard, sync, apps, custom_auth, \
     payment, integration, conversations, memories
 
 from utils.other.timeout import TimeoutMiddleware
@@ -33,7 +33,7 @@
 app.include_router(integration.router)
 app.include_router(agents.router)
 app.include_router(users.router)
-app.include_router(processing_memories.router)
+app.include_router(processing_conversations.router)
 app.include_router(trends.router)
 
 app.include_router(firmware.router)

backend/database/processing_memories.py

@@ -54,12 +54,17 @@ class ConversationPhoto(BaseModel):
     base64: str
     description: str
 
-
+# TODO: remove this class when the app is updated to use apps_results
 class PluginResult(BaseModel):
     plugin_id: Optional[str]
     content: str
 
 
+class AppResult(BaseModel):
+    app_id: Optional[str]
+    content: str
+
+
 class ActionItem(BaseModel):
     description: str = Field(description="The action item to be completed")
     completed: bool = False  # IGNORE ME from the model parser
@@ -164,14 +169,17 @@ class Conversation(BaseModel):
     started_at: Optional[datetime]
     finished_at: Optional[datetime]
 
-    source: Optional[ConversationSource] = ConversationSource.omi  # TODO: once released migrate db to include this field
+    source: Optional[ConversationSource] = ConversationSource.omi
     language: Optional[str] = None  # applies only to Friend # TODO: once released migrate db to default 'en'
 
     structured: Structured
     transcript_segments: List[TranscriptSegment] = []
     geolocation: Optional[Geolocation] = None
     photos: List[ConversationPhoto] = []
 
+    apps_results: List[AppResult] = []
+
+    # TODO: plugins_results for backward compatibility with the old memories routes and app
     plugins_results: List[PluginResult] = []
 
     external_data: Optional[Dict] = None
@@ -181,10 +189,19 @@ class Conversation(BaseModel):
     deleted: bool = False
     visibility: ConversationVisibility = ConversationVisibility.private
 
-    # processing_memory_id should be migrated to processing_conversation_id once the memories routes are deprecated
+    # TODO: processing_memory_id for backward compatibility with the old memories routes and app
     processing_memory_id: Optional[str] = None
+
+    processing_conversation_id: Optional[str] = None
+    
     status: Optional[ConversationStatus] = ConversationStatus.completed
 
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Update plugins_results based on apps_results
+        self.plugins_results = [PluginResult(plugin_id=app.app_id, content=app.content) for app in self.apps_results]
+        self.processing_memory_id = self.processing_conversation_id
+
     @staticmethod
     def conversations_to_string(conversations: List['Conversation'], use_transcript: bool = False) -> str:
         result = []

backend/main.py

@@ -77,10 +77,10 @@ def list_processing_conversation(uid: str = Depends(auth.get_current_user_uid),
     :param uid: user id.
     :return: The list of processing_memories.
     """
-    processing_memories = processing_conversation_utils.get_processing_memories(
+    processing_conversations = processing_conversation_utils.get_processing_memories(
         uid, filter_ids=filter_ids.split(",") if filter_ids else [], limit=5
     )
-    if not processing_memories or len(processing_memories) == 0:
+    if not processing_conversations or len(processing_conversations) == 0:
         return DetailProcessingConversationsResponse(result=[])
 
-    return DetailProcessingConversationsResponse(result=list(processing_memories))
+    return DetailProcessingConversationsResponse(result=list(processing_conversations))

backend/models/conversation.py

@@ -195,16 +195,16 @@ async def _create_conversation(conversation: dict):
 
         _send_message_event(ConversationEvent(event_type="memory_created", memory=conversation, messages=messages))
 
-    async def finalize_processing_memories(processing: List[dict]):
+    async def finalize_processing_conversations(processing: List[dict]):
         # handle edge case of conversation was actually processing? maybe later, doesn't hurt really anyway.
         # also fix from getMemories endpoint?
-        print('finalize_processing_memories len(processing):', len(processing), uid)
+        print('finalize_processing_conversations len(processing):', len(processing), uid)
         for conversation in processing:
             await _create_conversation(conversation)
 
     # Process processing conversations
     processing = conversations_db.get_processing_conversations(uid)
-    asyncio.create_task(finalize_processing_memories(processing))
+    asyncio.create_task(finalize_processing_conversations(processing))
 
     # Send last completed conversation to client
     async def send_last_conversation():