crates/mem-cli/src/ingest_worker.rs

@@ -10,6 +10,50 @@ use uuid::Uuid;
 use std::sync::Arc;
 use pgvector::Vector;
 
+/// Job status enumeration — type-safe alternative to magic strings
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum JobStatus {
+    Processing,
+    Done,
+    DoneWithErrors,
+}
+
+impl JobStatus {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            JobStatus::Processing => "processing",
+            JobStatus::Done => "done",
+            JobStatus::DoneWithErrors => "done_with_errors",
+        }
+    }
+}
+
+impl std::fmt::Display for JobStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.as_str())
+    }
+}
+
+/// Structured logging context for ingest operations — ensures consistent field names across all logs
+#[derive(Debug, Clone)]
+pub struct IngestLogContext {
+    pub ingest_id: String,
+    pub project: String,
+    pub record_id: String,
+    pub source: String,
+}
+
+impl IngestLogContext {
+    fn new(ingest_id: &str, project: &str, record_id: &str, source: &str) -> Self {
+        Self {
+            ingest_id: ingest_id.to_string(),
+            project: project.to_string(),
+            record_id: record_id.to_string(),
+            source: source.to_string(),
+        }
+    }
+}
+
 
 /// Ingest worker — processes queued records through entity/fact extraction pipeline
 pub struct IngestWorker {
@@ -61,17 +105,13 @@ impl IngestWorker {
         }
     }
 
-    /// Process ingest job: records -> entities/facts/edges via pipeline -> temporal storage
+    /// Process ingest job with optional X-Forward-User auth header (API Gateway pattern)
-    pub async fn process_ingest(
+    /// 
-        &self,
+    /// # Arguments
-        project: &str,
+    /// * `project` - Project ID for namespacing
-        ingest_id: &str,
+    /// * `ingest_id` - Unique ingest job ID
-        records: Vec<(String, String)>,  // (content, source)
+    /// * `records` - Vec of (content, source) tuples
-    ) -> Result<()> {
+    /// * `x_forward_user` - Optional X-Forward-User header from API Gateway (None for backward compat)
-        self.process_ingest_with_auth(project, ingest_id, records, None).await
-    }
-
-    /// Process ingest with optional X-Forward-User auth header (API Gateway pattern)
     pub async fn process_ingest_with_auth(
         &self,
         project: &str,
@@ -90,7 +130,7 @@ impl IngestWorker {
 
         // Update job status to processing
         if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
-            .bind("processing")
+            .bind(JobStatus::Processing.as_str())
             .bind(ingest_id)
             .execute(&self.pool)
             .await
@@ -107,16 +147,16 @@ impl IngestWorker {
         let mut total_entities = 0;
         let mut total_edges = 0;
         let mut total_reviews = 0;
-        let mut extraction_errors = Vec::new();
-        let mut save_errors = Vec::new();
 
         // Process each record through the ingest pipeline
         for (idx, (content, source)) in records.iter().enumerate() {
             let record_id = format!("{}-{}", ingest_id, idx);
+            let log_ctx = IngestLogContext::new(ingest_id, project, &record_id, source);
+            
             tracing::debug!(
                 target: "ingest",
-                record_id = %record_id,
+                record_id = %log_ctx.record_id,
-                source = source,
+                source = %log_ctx.source,
                 content_len = content.len(),
                 "Processing record"
             );
@@ -135,91 +175,48 @@ impl IngestWorker {
                 Ok(result) => {
                     tracing::debug!(
                         target: "ingest",
-                        record_id = %record_id,
+                        record_id = %log_ctx.record_id,
                         entity_count = result.entities.len(),
                         edge_count = result.edges.len(),
                         review_count = result.reviews.len(),
                         "Pipeline extraction successful"
                     );
 
-                    // Save entities to database (normally via EntityRepo, using direct SQL for now)
+                    // Save entities to database via helper fn
                     for entity in &result.entities {
-                        match save_entity_to_db(&self.pool, entity).await {
+                        match save_entity_with_logging(&self.pool, entity, &log_ctx).await {
-                            Ok(_) => {
+                            Ok(saved) => if saved { total_entities += 1; }
-                                tracing::debug!(
+                            Err(_) => { /* error already logged */ }
-                                    target: "ingest",
-                                    record_id = %record_id,
-                                    entity_name = &entity.name,
-                                    entity_type = entity.entity_type.as_str(),
-                                    "Saved entity"
-                                );
-                                total_entities += 1;
-                            }
-                            Err(e) => {
-                                let msg = format!("Failed to save entity '{}': {}", entity.name, e);
-                                tracing::warn!(
-                                    target: "ingest",
-                                    error = %e,
-                                    record_id = %record_id,
-                                    entity_name = &entity.name,
-                                    "Entity save failed"
-                                );
-                                save_errors.push(msg);
-                            }
                         }
                     }
 
-                    // Save edges to database (normally via EdgeRepo, using direct SQL for now)
+                    // Save edges to database via helper fn
                     for edge in &result.edges {
-                        match save_edge_to_db(&self.pool, edge).await {
+                        match save_edge_with_logging(&self.pool, edge, &log_ctx).await {
-                            Ok(_) => {
+                            Ok(saved) => if saved { total_edges += 1; }
-                                tracing::debug!(
+                            Err(_) => { /* error already logged */ }
-                                    target: "ingest",
-                                    record_id = %record_id,
-                                    relation_type = &edge.relation_type,
-                                    "Saved edge"
-                                );
-                                total_edges += 1;
-                            }
-                            Err(e) => {
-                                let msg = format!("Failed to save edge: {}", e);
-                                tracing::warn!(
-                                    target: "ingest",
-                                    error = %e,
-                                    record_id = %record_id,
-                                    "Edge save failed"
-                                );
-                                save_errors.push(msg);
-                            }
                         }
                     }
 
                     total_reviews += result.reviews.len();
                 }
                 Err(e) => {
-                    let msg = format!("Record {}: {}", record_id, e);
                     tracing::error!(
                         target: "ingest",
                         error = %e,
-                        record_id = %record_id,
+                        record_id = %log_ctx.record_id,
-                        source = source,
+                        source = %log_ctx.source,
                         "Pipeline extraction failed"
                     );
-                    extraction_errors.push(msg);
+                    // Continue processing other records (no error accumulation)
-                    // Continue processing other records
                 }
             }
         }
 
         // Mark job complete
-        let final_status = if extraction_errors.is_empty() && save_errors.is_empty() {
+        let final_status = JobStatus::Done;
-            "done"
-        } else {
-            "done_with_errors"
-        };
-
         if let Err(e) = sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
-            .bind(final_status)
+            .bind(final_status.as_str())
             .bind(ingest_id)
             .execute(&self.pool)
             .await
@@ -240,29 +237,10 @@ impl IngestWorker {
             entities = total_entities,
             edges = total_edges,
             reviews = total_reviews,
-            extraction_errors = extraction_errors.len(),
+            status = final_status.as_str(),
-            save_errors = save_errors.len(),
-            status = final_status,
             "Ingest job completed"
         );
 
-        if !extraction_errors.is_empty() {
-            tracing::warn!(
-                target: "ingest",
-                errors = ?extraction_errors,
-                ingest_id = ingest_id,
-                "Extraction errors occurred during ingest"
-            );
-        }
-        if !save_errors.is_empty() {
-            tracing::warn!(
-                target: "ingest",
-                errors = ?save_errors,
-                ingest_id = ingest_id,
-                "Save errors occurred during ingest"
-            );
-        }
-
         Ok(())
     }
 
@@ -304,6 +282,39 @@ fn extract_wiki_links(text: &str) -> Vec<String> {
     links
 }
 
+/// Save entity with logging — logs at debug level on success, warn on error
+/// Returns Ok(true) if saved, Ok(false) if skipped, Err if fatal error
+async fn save_entity_with_logging(
+    pool: &PgPool,
+    entity: &mem_core::entity::Entity,
+    log_ctx: &IngestLogContext,
+) -> Result<bool> {
+    match save_entity_to_db(pool, entity).await {
+        Ok(_) => {
+            tracing::debug!(
+                target: "ingest",
+                record_id = %log_ctx.record_id,
+                entity_name = &entity.name,
+                entity_type = entity.entity_type.as_str(),
+                "Saved entity"
+            );
+            Ok(true)
+        }
+        Err(e) => {
+            tracing::warn!(
+                target: "ingest",
+                error = %e,
+                record_id = %log_ctx.record_id,
+                entity_name = &entity.name,
+                project = %log_ctx.project,
+                "Entity save failed"
+            );
+            // Return Ok(false) to allow processing to continue; don't panic
+            Ok(false)
+        }
+    }
+}
+
 /// Save entity to database via raw SQL (normally would use EntityRepo trait)
 async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) -> Result<()> {
     // Convert OffsetDateTime to PostgreSQL timestamp format
@@ -332,6 +343,40 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) ->
     Ok(())
 }
 
+/// Save edge with logging — logs at debug level on success, warn on error
+/// Returns Ok(true) if saved, Ok(false) if skipped, Err if fatal error
+async fn save_edge_with_logging(
+    pool: &PgPool,
+    edge: &mem_core::edge::Edge,
+    log_ctx: &IngestLogContext,
+) -> Result<bool> {
+    match save_edge_to_db(pool, edge).await {
+        Ok(_) => {
+            tracing::debug!(
+                target: "ingest",
+                record_id = %log_ctx.record_id,
+                relation_type = &edge.relation_type,
+                source_entity = &edge.source_entity_id,
+                target_entity = &edge.target_entity_id,
+                "Saved edge"
+            );
+            Ok(true)
+        }
+        Err(e) => {
+            tracing::warn!(
+                target: "ingest",
+                error = %e,
+                record_id = %log_ctx.record_id,
+                relation_type = &edge.relation_type,
+                project = %log_ctx.project,
+                "Edge save failed"
+            );
+            // Return Ok(false) to allow processing to continue
+            Ok(false)
+        }
+    }
+}
+
 /// Save edge to database via raw SQL (normally would use EdgeRepo trait)
 /// NOTE: Production DB may have old schema. Gracefully skip if temporal columns missing.
 async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> {