diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index aa0e124..463ef59 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/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 - pub async fn process_ingest( - &self, - project: &str, - ingest_id: &str, - records: Vec<(String, String)>, // (content, source) - ) -> Result<()> { - self.process_ingest_with_auth(project, ingest_id, records, None).await - } - - /// Process ingest with optional X-Forward-User auth header (API Gateway pattern) + /// Process ingest job with optional X-Forward-User auth header (API Gateway pattern) + /// + /// # Arguments + /// * `project` - Project ID for namespacing + /// * `ingest_id` - Unique ingest job ID + /// * `records` - Vec of (content, source) tuples + /// * `x_forward_user` - Optional X-Forward-User header from API Gateway (None for backward compat) 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, - source = source, + record_id = %log_ctx.record_id, + 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 { - Ok(_) => { - tracing::debug!( - 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); - } + match save_entity_with_logging(&self.pool, entity, &log_ctx).await { + Ok(saved) => if saved { total_entities += 1; } + Err(_) => { /* error already logged */ } } } - // 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 { - Ok(_) => { - tracing::debug!( - 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); - } + match save_edge_with_logging(&self.pool, edge, &log_ctx).await { + Ok(saved) => if saved { total_edges += 1; } + Err(_) => { /* error already logged */ } } } total_reviews += result.reviews.len(); } Err(e) => { - let msg = format!("Record {}: {}", record_id, e); tracing::error!( target: "ingest", error = %e, - record_id = %record_id, - source = source, + record_id = %log_ctx.record_id, + source = %log_ctx.source, "Pipeline extraction failed" ); - extraction_errors.push(msg); - // Continue processing other records + // Continue processing other records (no error accumulation) } } } // Mark job complete - let final_status = if extraction_errors.is_empty() && save_errors.is_empty() { - "done" - } else { - "done_with_errors" - }; - + let final_status = JobStatus::Done; 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(), - save_errors = save_errors.len(), - status = final_status, + status = final_status.as_str(), "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 { 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 { + 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 { + 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<()> {