user restricted to own objects

src/ingress/analysis/ingress_analyser.rs

@@ -37,9 +37,10 @@ impl<'a> IngressAnalyzer<'a> {
         category: &str,
         instructions: &str,
         text: &str,
+        user_id: &str,
     ) -> Result<LLMGraphAnalysisResult, ProcessingError> {
         let similar_entities = self
-            .find_similar_entities(category, instructions, text)
+            .find_similar_entities(category, instructions, text, user_id)
             .await?;
         let llm_request =
             self.prepare_llm_request(category, instructions, text, &similar_entities)?;
@@ -51,13 +52,20 @@ impl<'a> IngressAnalyzer<'a> {
         category: &str,
         instructions: &str,
         text: &str,
+        user_id: &str,
     ) -> Result<Vec<KnowledgeEntity>, ProcessingError> {
         let input_text = format!(
             "content: {}, category: {}, user_instructions: {}",
             text, category, instructions
         );
 
-        combined_knowledge_entity_retrieval(self.db_client, self.openai_client, &input_text).await
+        combined_knowledge_entity_retrieval(
+            self.db_client,
+            self.openai_client,
+            &input_text,
+            user_id,
+        )
+        .await
     }
 
     fn prepare_llm_request(

src/ingress/analysis/types/llm_analysis_result.rs

@@ -53,6 +53,7 @@ impl LLMGraphAnalysisResult {
     pub async fn to_database_entities(
         &self,
         source_id: &str,
+        user_id: &str,
         openai_client: &async_openai::Client<async_openai::config::OpenAIConfig>,
     ) -> Result<(Vec<KnowledgeEntity>, Vec<KnowledgeRelationship>), ProcessingError> {
         // Create mapper and pre-assign IDs
@@ -60,7 +61,7 @@ impl LLMGraphAnalysisResult {
 
         // Process entities
         let entities = self
-            .process_entities(source_id, Arc::clone(&mapper), openai_client)
+            .process_entities(source_id, user_id, Arc::clone(&mapper), openai_client)
             .await?;
 
         // Process relationships
@@ -83,6 +84,7 @@ impl LLMGraphAnalysisResult {
     async fn process_entities(
         &self,
         source_id: &str,
+        user_id: &str,
         mapper: Arc<Mutex<GraphMapper>>,
         openai_client: &async_openai::Client<async_openai::config::OpenAIConfig>,
     ) -> Result<Vec<KnowledgeEntity>, ProcessingError> {
@@ -93,10 +95,12 @@ impl LLMGraphAnalysisResult {
                 let mapper = Arc::clone(&mapper);
                 let openai_client = openai_client.clone();
                 let source_id = source_id.to_string();
+                let user_id = user_id.to_string();
                 let entity = entity.clone();
 
                 task::spawn(async move {
-                    create_single_entity(&entity, &source_id, mapper, &openai_client).await
+                    create_single_entity(&entity, &source_id, &user_id, mapper, &openai_client)
+                        .await
                 })
             })
             .collect();
@@ -135,6 +139,7 @@ impl LLMGraphAnalysisResult {
 async fn create_single_entity(
     llm_entity: &LLMKnowledgeEntity,
     source_id: &str,
+    user_id: &str,
     mapper: Arc<Mutex<GraphMapper>>,
     openai_client: &async_openai::Client<async_openai::config::OpenAIConfig>,
 ) -> Result<KnowledgeEntity, ProcessingError> {
@@ -168,5 +173,6 @@ async fn create_single_entity(
         source_id: source_id.to_string(),
         metadata: None,
         embedding,
+        user_id: user_id.into(),
     })
 }

src/ingress/content_processor.rs

@@ -49,7 +49,7 @@ impl ContentProcessor {
 
         // Convert analysis to objects
         let (entities, relationships) = analysis
-            .to_database_entities(&content.id, &self.openai_client)
+            .to_database_entities(&content.id, &content.user_id, &self.openai_client)
             .await?;
 
         // Store everything
@@ -68,7 +68,12 @@ impl ContentProcessor {
     ) -> Result<LLMGraphAnalysisResult, ProcessingError> {
         let analyser = IngressAnalyzer::new(&self.db_client, &self.openai_client);
         analyser
-            .analyze_content(&content.category, &content.instructions, &content.text)
+            .analyze_content(
+                &content.category,
+                &content.instructions,
+                &content.text,
+                &content.user_id,
+            )
             .await
     }
 
@@ -102,7 +107,12 @@ impl ContentProcessor {
         // Could potentially process chunks in parallel with a bounded concurrent limit
         for chunk in chunks {
             let embedding = generate_embedding(&self.openai_client, chunk).await?;
-            let text_chunk = TextChunk::new(content.id.to_string(), chunk.to_string(), embedding);
+            let text_chunk = TextChunk::new(
+                content.id.to_string(),
+                chunk.to_string(),
+                embedding,
+                content.user_id.to_string(),
+            );
             store_item(&self.db_client, text_chunk).await?;
         }
 

src/ingress/types/ingress_input.rs

@@ -56,6 +56,7 @@ pub enum IngressContentError {
 pub async fn create_ingress_objects(
     input: IngressInput,
     db_client: &SurrealDbClient,
+    user_id: &str,
 ) -> Result<Vec<IngressObject>, IngressContentError> {
     // Initialize list
     let mut object_list = Vec::new();
@@ -69,6 +70,7 @@ pub async fn create_ingress_objects(
                     url: url.to_string(),
                     instructions: input.instructions.clone(),
                     category: input.category.clone(),
+                    user_id: user_id.into(),
                 });
             }
             Err(_) => {
@@ -77,6 +79,7 @@ pub async fn create_ingress_objects(
                     text: input_content.to_string(),
                     instructions: input.instructions.clone(),
                     category: input.category.clone(),
+                    user_id: user_id.into(),
                 });
             }
         }
@@ -90,6 +93,7 @@ pub async fn create_ingress_objects(
                     file_info,
                     instructions: input.instructions.clone(),
                     category: input.category.clone(),
+                    user_id: user_id.into(),
                 });
             } else {
                 info!("No file with id: {}", id);

src/ingress/types/ingress_object.rs

@@ -10,16 +10,19 @@ pub enum IngressObject {
         url: String,
         instructions: String,
         category: String,
+        user_id: String,
     },
     Text {
         text: String,
         instructions: String,
         category: String,
+        user_id: String,
     },
     File {
         file_info: FileInfo,
         instructions: String,
         category: String,
+        user_id: String,
     },
 }
 
@@ -37,6 +40,7 @@ impl IngressObject {
                 url,
                 instructions,
                 category,
+                user_id,
             } => {
                 let text = Self::fetch_text_from_url(url).await?;
                 Ok(TextContent::new(
@@ -44,22 +48,26 @@ impl IngressObject {
                     instructions.into(),
                     category.into(),
                     None,
+                    user_id.into(),
                 ))
             }
             IngressObject::Text {
                 text,
                 instructions,
                 category,
+                user_id,
             } => Ok(TextContent::new(
                 text.into(),
                 instructions.into(),
                 category.into(),
                 None,
+                user_id.into(),
             )),
             IngressObject::File {
                 file_info,
                 instructions,
                 category,
+                user_id,
             } => {
                 let text = Self::extract_text_from_file(file_info).await?;
                 Ok(TextContent::new(
@@ -67,6 +75,7 @@ impl IngressObject {
                     instructions.into(),
                     category.into(),
                     Some(file_info.to_owned()),
+                    user_id.into(),
                 ))
             }
         }