diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index bc4a5bb..8d8e7e8 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/crates/mem-cli/src/ingest_worker.rs @@ -10,6 +10,7 @@ use uuid::Uuid; use std::sync::Arc; use pgvector::Vector; + /// Ingest worker — processes queued records through entity/fact extraction pipeline pub struct IngestWorker { pool: PgPool, @@ -135,9 +136,15 @@ impl IngestWorker { .await?; tracing::info!( - "Ingest completed: {} (entities={}, edges={}, reviews={})", - ingest_id, total_entities, total_edges, total_reviews + target: "observability", + event = "ingest_complete", + ingest_id = ingest_id, + entities = total_entities, + edges = total_edges, + reviews = total_reviews, + "Ingest completed" ); + Ok(()) } @@ -187,7 +194,12 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) -> sqlx::query( "INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence) VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8) - ON CONFLICT (id) DO NOTHING" + ON CONFLICT (project_id, name) DO UPDATE SET + entity_type = EXCLUDED.entity_type, + description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description), + t_updated = NOW(), + confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence), + source_count = memory_entity.source_count + 1" ) .bind(&entity.id) .bind(&entity.project_id)