feat(search_processing_service): chats kafka (#5304)

Author: whutchinson98

.github/workspace-dep-closures.json

@@ -1304,6 +1304,7 @@
       "crates/authentication_service_client",
       "crates/bot_id",
       "crates/channel_sender",
+      "crates/chat",
       "crates/cowlike",
       "crates/crm",
       "crates/document_sub_type",
@@ -1359,6 +1360,7 @@
       "crates/properties",
       "crates/rate_limit",
       "crates/remote_env_var",
+      "crates/roles_and_permissions",
       "crates/s3_key",
       "crates/sqs_client",
       "crates/system_properties",
@@ -4924,6 +4926,7 @@
       "crates/call",
       "crates/channel_sender",
       "crates/channels",
+      "crates/chat",
       "crates/comms_db_client",
       "crates/connection",
       "crates/connection_gateway_client",
@@ -4996,6 +4999,7 @@
       "crates/properties",
       "crates/rate_limit",
       "crates/remote_env_var",
+      "crates/roles_and_permissions",
       "crates/s3_client",
       "crates/s3_key",
       "crates/share_permission_db_utils",

…b9ae11c6625c948798e19ca19da4a30f3c1.json …243051bd3bde6e517014cd5768334fd7757.json.sqlx/query-92e2916a16d165600f457c778ac3db9ae11c6625c948798e19ca19da4a30f3c1.json renamed to .sqlx/query-eb360ba6aad69fe2c501baa8b61b4243051bd3bde6e517014cd5768334fd7757.json .sqlx/query-92e2916a16d165600f457c778ac3db9ae11c6625c948798e19ca19da4a30f3c1.json renamed to .sqlx/query-eb360ba6aad69fe2c501baa8b61b4243051bd3bde6e517014cd5768334fd7757.json

@@ -4,12 +4,14 @@ VALUES ('a1111111-1111-1111-1111-111111111111', 'user', 'user@user.com', 'stripe
 INSERT INTO public."User" ("id","email","stripeCustomerId","macro_user_id")
 VALUES ('macro|user@user.com', 'user@user.com','stripe_id','a1111111-1111-1111-1111-111111111111');
 
-INSERT INTO public."Chat" ("id","name","userId","model","createdAt","updatedAt","isPersistent")
+INSERT INTO public."Chat" ("id","name","userId","model","createdAt","updatedAt","deletedAt","isPersistent")
 VALUES
-  ('chat-persistent', 'persistent chat', 'macro|user@user.com', 'gpt-4o', '2024-01-01 00:00:00', '2024-01-01 00:00:00', true),
-  ('chat-ephemeral', 'ephemeral chat', 'macro|user@user.com', 'gpt-4o', '2024-01-01 00:00:00', '2024-01-01 00:00:00', false);
+  ('chat-persistent', 'persistent chat', 'macro|user@user.com', 'gpt-4o', '2024-01-01 00:00:00', '2024-01-01 00:00:00', NULL, true),
+  ('chat-ephemeral', 'ephemeral chat', 'macro|user@user.com', 'gpt-4o', '2024-02-01 00:00:00', '2024-02-01 00:00:00', NULL, false),
+  ('chat-deleted', 'deleted chat', 'macro|user@user.com', 'gpt-4o', '2024-03-01 00:00:00', '2024-03-01 00:00:00', '2024-03-04 05:06:07.890', true);
 
-INSERT INTO public."ChatMessage" ("id","content","role","chatId")
+INSERT INTO public."ChatMessage" ("id","content","role","chatId","createdAt","updatedAt")
 VALUES
-  ('msg-persistent', '"codebase brighter"', 'user', 'chat-persistent'),
-  ('msg-ephemeral', '"another message"', 'user', 'chat-ephemeral');
+  ('msg-persistent', '"codebase brighter"', 'user', 'chat-persistent', '2024-01-02 03:04:05.123', '2024-01-03 04:05:06.789'),
+  ('msg-ephemeral', '"another message"', 'assistant', 'chat-ephemeral', '2024-02-02 03:04:05.123', '2024-02-03 04:05:06.789'),
+  ('msg-deleted', '"remove from search"', 'assistant', 'chat-deleted', '2024-03-02 03:04:05.123', '2024-03-03 04:05:06.789');

Cargo.lock

@@ -258,6 +258,9 @@ pub struct ChatMessageInfo {
     pub name: String,
     pub content: String,
     pub role: String,
+    pub owner_user_id: String,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
     pub deleted_at: Option<DateTime<Utc>>,
 }
 
@@ -277,7 +280,10 @@ pub async fn get_chat_message_info(
             m.content as "content",
             c.name as "name",
             m.role as "role",
-            c."deletedAt"::timestamptz as "deleted_at"
+            c."deletedAt"::timestamptz as "deleted_at",
+            c."userId" as "owner_user_id",
+            m."createdAt" as "created_at",
+            m."updatedAt" as "updated_at"
         FROM
             "ChatMessage" m
         JOIN
@@ -300,6 +306,9 @@ pub async fn get_chat_message_info(
         name: row.name,
         content,
         role: row.role,
+        owner_user_id: row.owner_user_id,
+        created_at: DateTime::<Utc>::from_naive_utc_and_offset(row.created_at, Utc),
+        updated_at: DateTime::<Utc>::from_naive_utc_and_offset(row.updated_at, Utc),
         deleted_at: row.deleted_at,
     }))
 }

crates/macro_db_client/fixtures/chat_message_info.sql

@@ -2,20 +2,86 @@ use super::*;
 use sqlx::{Pool, Postgres};
 
 #[sqlx::test(fixtures(path = "../../../fixtures", scripts("chat_message_info")))]
-async fn persistent_chat_messages_are_indexed_for_search(
+async fn returns_persistent_and_ephemeral_chat_message_metadata(
     pool: Pool<Postgres>,
 ) -> anyhow::Result<()> {
     let persistent = get_chat_message_info(&pool, "chat-persistent", "msg-persistent")
         .await?
         .expect("persistent chat message should be returned for indexing");
+    assert_eq!(persistent.name, "persistent chat");
     assert_eq!(persistent.content, "codebase brighter");
+    assert_eq!(persistent.role, "user");
+    assert_eq!(persistent.owner_user_id, "macro|user@user.com");
+    assert_eq!(
+        persistent.created_at,
+        "2024-01-02T03:04:05.123Z".parse::<DateTime<Utc>>()?
+    );
+    assert_eq!(
+        persistent.updated_at,
+        "2024-01-03T04:05:06.789Z".parse::<DateTime<Utc>>()?
+    );
     assert!(persistent.deleted_at.is_none());
 
     let ephemeral = get_chat_message_info(&pool, "chat-ephemeral", "msg-ephemeral")
         .await?
         .expect("ephemeral chat message should be returned for indexing");
+    assert_eq!(ephemeral.name, "ephemeral chat");
     assert_eq!(ephemeral.content, "another message");
+    assert_eq!(ephemeral.role, "assistant");
+    assert_eq!(ephemeral.owner_user_id, "macro|user@user.com");
+    assert_eq!(
+        ephemeral.created_at,
+        "2024-02-02T03:04:05.123Z".parse::<DateTime<Utc>>()?
+    );
+    assert_eq!(
+        ephemeral.updated_at,
+        "2024-02-03T04:05:06.789Z".parse::<DateTime<Utc>>()?
+    );
     assert!(ephemeral.deleted_at.is_none());
 
     Ok(())
 }
+
+#[sqlx::test(fixtures(path = "../../../fixtures", scripts("chat_message_info")))]
+async fn returns_soft_deleted_chat_message_metadata(pool: Pool<Postgres>) -> anyhow::Result<()> {
+    let message = get_chat_message_info(&pool, "chat-deleted", "msg-deleted")
+        .await?
+        .expect("soft-deleted chat message should be returned for removal");
+
+    assert_eq!(message.name, "deleted chat");
+    assert_eq!(message.content, "remove from search");
+    assert_eq!(message.role, "assistant");
+    assert_eq!(message.owner_user_id, "macro|user@user.com");
+    assert_eq!(
+        message.created_at,
+        "2024-03-02T03:04:05.123Z".parse::<DateTime<Utc>>()?
+    );
+    assert_eq!(
+        message.updated_at,
+        "2024-03-03T04:05:06.789Z".parse::<DateTime<Utc>>()?
+    );
+    assert_eq!(
+        message.deleted_at,
+        Some("2024-03-04T05:06:07.890Z".parse::<DateTime<Utc>>()?)
+    );
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures(path = "../../../fixtures", scripts("chat_message_info")))]
+async fn returns_none_for_missing_or_mismatched_message_id(
+    pool: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    assert!(
+        get_chat_message_info(&pool, "chat-persistent", "msg-missing")
+            .await?
+            .is_none()
+    );
+    assert!(
+        get_chat_message_info(&pool, "chat-ephemeral", "msg-persistent")
+            .await?
+            .is_none()
+    );
+
+    Ok(())
+}

crates/macro_db_client/src/chat/get.rs

@@ -206,15 +206,9 @@ pub trait ShaCounterPort: Send + Sync + 'static {
     ) -> impl Future<Output = anyhow::Result<()>> + Send;
 }
 
-/// Port for publishing cross-family search and document-deletion work.
-// TODO: Remove this port and its SQS adapter after the chats and documents migrations.
+/// Port for publishing document search-removal and deletion work.
+// TODO: Remove this port and its SQS adapter after the documents migration.
 pub trait ProjectSearchIndexer: Send + Sync + 'static {
-    /// Remove chats from the search index.
-    fn remove_chats(
-        &self,
-        chat_ids: Vec<String>,
-    ) -> impl Future<Output = anyhow::Result<()>> + Send;
-
     /// Remove documents from the search index.
     fn remove_documents(
         &self,

crates/macro_db_client/src/chat/get/test.rs

@@ -77,7 +77,7 @@ where
     pub sha_counter: Sha,
     /// Entity-access inheritance manager.
     pub entity_access_management_service: Eam,
-    /// Search and deletion queue publisher.
+    /// Document search and deletion queue publisher.
     pub search_indexer: Idx,
     /// Project lifecycle event broker.
     pub macro_event_broker: B,
@@ -534,13 +534,6 @@ where
             .map(|(document_id, _)| document_id.clone())
             .collect::<Vec<_>>();
 
-        if !purged.chat_ids.is_empty() {
-            let _ = self
-                .search_indexer
-                .remove_chats(purged.chat_ids)
-                .await
-                .inspect_err(|error| tracing::error!(error = ?error, "unable to enqueue purged chats for search"));
-        }
         if !purged.documents.is_empty() {
             let _ = self
                 .search_indexer

crates/projects/src/domain/ports.rs

@@ -153,10 +153,6 @@ impl EntityAccessManagementService for NullPort {
 }
 
 impl ProjectSearchIndexer for NullPort {
-    async fn remove_chats(&self, _chat_ids: Vec<String>) -> anyhow::Result<()> {
-        unreachable!()
-    }
-
     async fn remove_documents(&self, _document_ids: Vec<String>) -> anyhow::Result<()> {
         unreachable!()
     }
@@ -224,10 +220,6 @@ struct RecordingIndexer {
 }
 
 impl ProjectSearchIndexer for RecordingIndexer {
-    async fn remove_chats(&self, _chat_ids: Vec<String>) -> anyhow::Result<()> {
-        unreachable!()
-    }
-
     async fn remove_documents(&self, _document_ids: Vec<String>) -> anyhow::Result<()> {
         unreachable!()
     }
@@ -1574,10 +1566,6 @@ impl OrderedIndexer {
 }
 
 impl ProjectSearchIndexer for OrderedIndexer {
-    async fn remove_chats(&self, _chat_ids: Vec<String>) -> anyhow::Result<()> {
-        self.record("chats")
-    }
-
     async fn remove_documents(&self, _document_ids: Vec<String>) -> anyhow::Result<()> {
         self.record("documents")
     }
@@ -1740,7 +1728,7 @@ async fn permanent_delete_runs_external_work_after_committed_purge() {
 
     assert_eq!(
         *events.lock().unwrap(),
-        vec!["repo", "sha", "chats", "documents", "document_deletes"]
+        vec!["repo", "sha", "documents", "document_deletes"]
     );
     let published = published.lock().unwrap();
     assert_eq!(published.len(), 1);

crates/projects/src/domain/service.rs

@@ -1,12 +1,12 @@
-//! SQS adapter for cross-family search cleanup and document deletion.
+//! SQS adapter for document search cleanup and deletion.
 
 use std::sync::Arc;
 
-use sqs_client::search::{SearchQueueMessage, chat::RemoveChatMessage, document::DocumentId};
+use sqs_client::search::{SearchQueueMessage, document::DocumentId};
 
 use crate::domain::ports::ProjectSearchIndexer;
 
-/// SQS-backed cross-family search cleanup and document-deletion adapter.
+/// SQS-backed document search cleanup and deletion adapter.
 #[derive(Clone)]
 pub struct SqsProjectSearchIndexer {
     sqs: Arc<sqs_client::SQS>,
@@ -20,24 +20,6 @@ impl SqsProjectSearchIndexer {
 }
 
 impl ProjectSearchIndexer for SqsProjectSearchIndexer {
-    #[tracing::instrument(skip(self), err)]
-    async fn remove_chats(&self, chat_ids: Vec<String>) -> anyhow::Result<()> {
-        let messages = chat_ids
-            .into_iter()
-            .map(|chat_id| {
-                SearchQueueMessage::RemoveChatMessage(RemoveChatMessage {
-                    chat_id,
-                    message_id: None,
-                    index_override: None,
-                })
-            })
-            .collect();
-
-        self.sqs
-            .bulk_send_message_to_search_event_queue(messages)
-            .await
-    }
-
     #[tracing::instrument(skip(self), err)]
     async fn remove_documents(&self, document_ids: Vec<String>) -> anyhow::Result<()> {
         let messages = document_ids