test: production ingest E2E test suite with enhanced logging (#55)

## Summary
Production testing of ingest + embedding pipeline with api-gw integration.

## Root Cause
9 SQL migrations in `crates/mem-store/migrations/` not applied to production database.

Missing tables:
- `memory_entity`
- `memory_edge`
- `memory_edge_temporal`
- Vector embeddings tables
- And 15+ more schema objects

Evidence from logs:
```
WARN: Failed to save entity Docker:
      error returned from database: relation "memory_entity" does not exist
```

## Deliverables
- `test_prod_ingest_real.sh` - Full E2E test against K8s + api-gw
- `apply_migrations.sh` - Manual schema migration (backup)
- `collect_prod_logs.sh` - Pod log collection before/after
- `run_production_test.sh` - Test orchestrator
- `tests/integration_ingest_with_gw.rs` - Integration test
- `tests/unit_ingest_logging.rs` - Unit tests for extraction
- Enhanced logging in `ingest_worker.rs` - Per-record event tracking

## Next Steps
1. Trigger "DB Migration" workflow in Forgejo Actions
2. This applies all 9 migrations from `crates/mem-store/migrations/`
3. Pod restart (automatic)
4. Re-run E2E test - should pass completely

**ETA:** ~15 minutes (3-5 min migrations + 2 min restart + verification)

## How to Test Locally
```bash
./test_prod_ingest_real.sh --verbose
```

Requires:
- kubectl access to poimen namespace
- Port-forwarding to memory-service

---------

Co-authored-by: rock <[email protected]>
Reviewed-on: #55
Co-authored-by: poimen <[email protected]>

crates/mem-cli/src/ingest_worker.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use mem_store::{MemoryL1, VectorStore, ChunkL0, EntityRepoOps, EdgeRepoOps};
+use mem_store::{VectorStore, ChunkL0};
 use mem_llm::EmbeddingsClient;
 use mem_ingest::ingest_pipeline::{IngestPipeline, Episode};
 use mem_ingest::entity_extractor::{WikiLinkFallbackExtractor, LlmEntityExtractor};
@@ -8,22 +8,145 @@ use mem_ingest::contradiction_detector::ContradictionHandler;
 use sqlx::PgPool;
 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)]
+#[allow(dead_code)]
+pub enum JobStatus {
+    Processing,
+    Done,
+    DoneWithErrors,
+}
+
+#[allow(dead_code)]
+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())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Mock JobStatusStore for testing
+    pub struct MockJobStatusStore {
+        updates: std::sync::Arc<std::sync::Mutex<Vec<(String, JobStatus)>>>,
+    }
+
+    impl MockJobStatusStore {
+        pub fn new() -> Self {
+            Self {
+                updates: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
+            }
+        }
+
+        pub fn updates(&self) -> Vec<(String, JobStatus)> {
+            self.updates.lock().unwrap().clone()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl JobStatusStore for MockJobStatusStore {
+        async fn update_status(&self, ingest_id: &str, status: JobStatus) -> Result<()> {
+            self.updates.lock().unwrap().push((ingest_id.to_string(), status));
+            Ok(())
+        }
+    }
+}
+
+/// Structured logging context for ingest operations — ensures consistent field names across all logs
+#[derive(Debug, Clone)]
+#[allow(dead_code)]
+pub struct IngestLogContext {
+    pub ingest_id: String,
+    pub project: String,
+    pub record_id: String,
+    pub source: String,
+}
+
+#[allow(dead_code)]
+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(),
+        }
+    }
+}
+
+/// Job status store trait — abstracts database persistence of job status (enables mocking)
+#[async_trait::async_trait]
+#[allow(dead_code)]
+pub trait JobStatusStore: Send + Sync {
+    /// Update job status in storage
+    async fn update_status(&self, ingest_id: &str, status: JobStatus) -> Result<()>;
+}
+
+/// PostgreSQL implementation of JobStatusStore
+#[allow(dead_code)]
+pub struct PgJobStatusStore {
+    pool: PgPool,
+}
+
+#[allow(dead_code)]
+impl PgJobStatusStore {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait::async_trait]
+impl JobStatusStore for PgJobStatusStore {
+    async fn update_status(&self, ingest_id: &str, status: JobStatus) -> Result<()> {
+        sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
+            .bind(status.as_str())
+            .bind(ingest_id)
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
+}
 
 
 /// Ingest worker — processes queued records through entity/fact extraction pipeline
+#[allow(dead_code)]
 pub struct IngestWorker {
     pool: PgPool,
     vector_store: Arc<VectorStore>,
     embeddings: Arc<EmbeddingsClient>,
     pipeline: Arc<IngestPipeline>,
+    job_status_store: Arc<dyn JobStatusStore>,
 }
 
+#[allow(dead_code)]
 impl IngestWorker {
     /// Create worker with full ingest pipeline
     pub fn new(
         pool: PgPool,
         embeddings: EmbeddingsClient,
+    ) -> Self {
+        let job_status_store = Arc::new(PgJobStatusStore::new(pool.clone()));
+        Self::with_job_store(pool, embeddings, job_status_store)
+    }
+
+    /// Create worker with custom job status store (for testing)
+    pub fn with_job_store(
+        pool: PgPool,
+        embeddings: EmbeddingsClient,
+        job_status_store: Arc<dyn JobStatusStore>,
     ) -> Self {
         let vector_store = Arc::new(VectorStore::new(pool.clone()));
         
@@ -58,24 +181,43 @@ impl IngestWorker {
             vector_store,
             embeddings: Arc::new(embeddings),
             pipeline,
+            job_status_store,
         }
     }
 
-    /// Process ingest job: records -> entities/facts/edges via pipeline -> temporal storage
-    pub async fn process_ingest(
+    /// 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,
         ingest_id: &str,
         records: Vec<(String, String)>,  // (content, source)
+        x_forward_user: Option<String>,
     ) -> Result<()> {
-        tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len());
+        tracing::info!(
+            target: "ingest",
+            event = "ingest_start",
+            ingest_id = ingest_id,
+            project = project,
+            record_count = records.len(),
+            "Starting ingest job"
+        );
 
-        // Update job status to processing
-        sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
-            .bind("processing")
-            .bind(ingest_id)
-            .execute(&self.pool)
-            .await?;
+        // Update job status to processing (via trait, testable)
+        if let Err(e) = self.job_status_store.update_status(ingest_id, JobStatus::Processing).await {
+            tracing::error!(
+                target: "ingest",
+                error = %e,
+                ingest_id = ingest_id,
+                "Failed to update job status to processing"
+            );
+            return Err(e.into());
+        }
 
         let mut total_entities = 0;
         let mut total_edges = 0;
@@ -83,66 +225,90 @@ impl IngestWorker {
 
         // 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 = %log_ctx.record_id,
+                source = %log_ctx.source,
+                content_len = content.len(),
+                "Processing record"
+            );
+
             // Create episode from record
             let episode = Episode {
-                id: format!("{}-{}", ingest_id, idx),
+                id: record_id.clone(),
                 project_id: project.to_string(),
                 text: content.clone(),
                 wiki_links: extract_wiki_links(content),
             };
 
             // Run extraction pipeline (entity + fact extraction + contradiction detection)
-            match self.pipeline.ingest(&episode).await {
+            let x_forward_user_ref = x_forward_user.as_deref();
+            match self.pipeline.ingest_with_auth(&episode, x_forward_user_ref).await {
                 Ok(result) => {
                     tracing::debug!(
-                        "Pipeline extracted {} entities, {} edges for episode {}",
-                        result.entities.len(),
-                        result.edges.len(),
-                        episode.id
+                        target: "ingest",
+                        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 with embeddings (RAG-006)
                     for entity in &result.entities {
-                        if let Err(e) = save_entity_to_db(&self.pool, entity).await {
-                            tracing::warn!("Failed to save entity {}: {}", entity.name, e);
-                        } else {
-                            total_entities += 1;
+                        match save_entity_with_embedding(&self.pool, &self.embeddings, 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 with embeddings (RAG-006)
                     for edge in &result.edges {
-                        if let Err(e) = save_edge_to_db(&self.pool, edge).await {
-                            tracing::warn!("Failed to save edge: {}", e);
-                        } else {
-                            total_edges += 1;
+                        match save_edge_with_embedding(&self.pool, &self.embeddings, edge, &log_ctx).await {
+                            Ok(saved) => if saved { total_edges += 1; }
+                            Err(_) => { /* error already logged */ }
                         }
                     }
 
                     total_reviews += result.reviews.len();
                 }
                 Err(e) => {
-                    tracing::error!("Pipeline failed for episode {}: {}", episode.id, e);
-                    // Continue processing other records
+                    tracing::error!(
+                        target: "ingest",
+                        error = %e,
+                        record_id = %log_ctx.record_id,
+                        source = %log_ctx.source,
+                        "Pipeline extraction failed"
+                    );
+                    // Continue processing other records (no error accumulation)
                 }
             }
         }
 
-        // Mark job complete
-        sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
-            .bind("done")
-            .bind(ingest_id)
-            .execute(&self.pool)
-            .await?;
+        // Mark job complete (via trait, testable)
+        let final_status = JobStatus::Done;
+        if let Err(e) = self.job_status_store.update_status(ingest_id, final_status).await {
+            tracing::error!(
+                target: "ingest",
+                error = %e,
+                ingest_id = ingest_id,
+                "Failed to update job completion status"
+            );
+        }
 
         tracing::info!(
-            target: "observability",
+            target: "ingest",
             event = "ingest_complete",
             ingest_id = ingest_id,
+            project = project,
             entities = total_entities,
             edges = total_edges,
             reviews = total_reviews,
-            "Ingest completed"
+            status = final_status.as_str(),
+            "Ingest job completed"
         );
 
         Ok(())
@@ -150,7 +316,7 @@ impl IngestWorker {
 
     /// Process a single chunk
     pub async fn process_chunk(&self, project: &str, query_id: &str, content: &str, source: &str) -> Result<()> {
-        let embedding = self.embeddings.embed_one(content).await?;
+        let _embedding = self.embeddings.embed_one(content).await?;
         let chunk = ChunkL0 {
             id: Uuid::new_v4(),
             project: project.to_string(),
@@ -165,6 +331,7 @@ impl IngestWorker {
 }
 
 /// Extract wiki links from text (e.g., [[Kubernetes]] -> "Kubernetes")
+#[allow(dead_code)]
 fn extract_wiki_links(text: &str) -> Vec<String> {
     let mut links = Vec::new();
     let mut chars = text.chars().peekable();
@@ -186,41 +353,178 @@ fn extract_wiki_links(text: &str) -> Vec<String> {
     links
 }
 
-/// 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
+/// 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
+/// Save entity with embeddings (RAG-006)
+/// Embeds name + summary before persisting, so semantic search can find entities.
+#[allow(dead_code)]
+async fn save_entity_with_embedding(
+    pool: &PgPool,
+    embeddings: &EmbeddingsClient,
+    entity: &mem_core::entity::Entity,
+    log_ctx: &IngestLogContext,
+) -> Result<bool> {
+    // Embed entity name
+    let name_embedding = match embeddings.embed_one(&entity.name).await {
+        Ok(emb) => Some(emb.to_vec()),
+        Err(e) => {
+            tracing::warn!(
+                target: "ingest",
+                error = %e,
+                entity_name = &entity.name,
+                "Name embedding failed, saving entity without name_embedding"
+            );
+            None
+        }
+    };
+
+    // Embed summary if present
+    let summary_embedding = if let Some(ref summary) = entity.summary {
+        match embeddings.embed_one(summary).await {
+            Ok(emb) => Some(emb.to_vec()),
+            Err(e) => {
+                tracing::debug!(target: "ingest", error = %e, "Summary embedding failed");
+                None
+            }
+        }
+    } else {
+        None
+    };
+
     let t_created_str = entity.t_created.to_string();
-    
-    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 (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),
+
+    let result = sqlx::query(
+        "INSERT INTO memory_entity (id, project_id, name, entity_type, description, summary, \
+         name_embedding, summary_embedding, t_created, t_updated, confidence) \
+         VALUES ($1::UUID, $2, $3, $4, $5, $6, $7, $8, $9::TIMESTAMPTZ, $10::TIMESTAMPTZ, $11) \
+         ON CONFLICT (project_id, name) DO UPDATE SET \
+           entity_type = EXCLUDED.entity_type, \
+           description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description), \
+           summary = COALESCE(NULLIF(EXCLUDED.summary, ''), memory_entity.summary), \
+           name_embedding = COALESCE(EXCLUDED.name_embedding, memory_entity.name_embedding), \
+           summary_embedding = COALESCE(EXCLUDED.summary_embedding, memory_entity.summary_embedding), \
+           t_updated = NOW(), \
+           confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence), \
            source_count = memory_entity.source_count + 1"
     )
     .bind(&entity.id)
     .bind(&entity.project_id)
     .bind(&entity.name)
     .bind(entity.entity_type.as_str())
-    .bind(entity.summary.as_deref())
+    .bind(entity.summary.as_deref()) // description
+    .bind(entity.summary.as_deref()) // summary
+    .bind(name_embedding.as_deref())
+    .bind(summary_embedding.as_deref())
     .bind(&t_created_str)
     .bind(&t_created_str)
-    .bind(1.0_f32)  // default confidence
+    .bind(1.0_f32)
     .execute(pool)
-    .await?;
-    Ok(())
+    .await;
+
+    match result {
+        Ok(_) => {
+            tracing::debug!(
+                target: "ingest",
+                record_id = %log_ctx.record_id,
+                entity_name = &entity.name,
+                entity_type = entity.entity_type.as_str(),
+                has_name_emb = name_embedding.is_some(),
+                has_summary_emb = summary_embedding.is_some(),
+                "Saved entity with embeddings"
+            );
+            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"
+            );
+            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.
+/// Save edge with fact embedding (RAG-006)
+/// Embeds fact text before persisting, so semantic search can find edges.
+#[allow(dead_code)]
+async fn save_edge_with_embedding(
+    pool: &PgPool,
+    embeddings: &EmbeddingsClient,
+    edge: &mem_core::edge::Edge,
+    log_ctx: &IngestLogContext,
+) -> Result<bool> {
+    // Embed the fact text
+    let fact_embedding = match embeddings.embed_one(&edge.fact).await {
+        Ok(emb) => Some(emb.to_vec()),
+        Err(e) => {
+            tracing::warn!(
+                target: "ingest",
+                error = %e,
+                fact = &edge.fact,
+                "Fact embedding failed, saving edge without fact_embedding"
+            );
+            None
+        }
+    };
+
+    let result = sqlx::query(
+        "INSERT INTO memory_edge (id, project_id, source_id, target_id, relation_type, fact, \
+         fact_embedding, t_valid, t_invalid, t_created, confidence) \
+         VALUES ($1::UUID, $2, $3::UUID, $4::UUID, $5, $6, $7, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10::TIMESTAMPTZ, $11) \
+         ON CONFLICT (id) DO NOTHING"
+    )
+    .bind(&edge.id)
+    .bind(&edge.project_id)
+    .bind(&edge.source_entity_id)
+    .bind(&edge.target_entity_id)
+    .bind(&edge.relation_type)
+    .bind(&edge.fact)
+    .bind(fact_embedding.as_deref())
+    .bind(edge.t_valid.map(|t| t.to_string()))
+    .bind(edge.t_invalid.map(|t| t.to_string()))
+    .bind(edge.t_created.to_string())
+    .bind(edge.confidence)
+    .execute(pool)
+    .await;
+
+    match result {
+        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,
+                has_fact_emb = fact_embedding.is_some(),
+                "Saved edge with embedding"
+            );
+            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"
+            );
+            Ok(false)
+        }
+    }
+}
+
+// Legacy save functions kept for backward compatibility but unused
+#[allow(dead_code)]
+#[allow(dead_code)]
 async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> {
-    // Try temporal schema first (id, project_id, source_entity_id, etc)
     let result = sqlx::query(
         "INSERT INTO memory_edge (id, project_id, source_id, target_id, relation_type, fact, t_valid, t_invalid, t_created, confidence)
-         VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10)
+         VALUES ($1::UUID, $2, $3::UUID, $4::UUID, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10)
          ON CONFLICT (id) DO NOTHING"
     )
     .bind(&edge.id)
@@ -239,8 +543,7 @@ async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<(
     match result {
         Ok(_) => Ok(()),
         Err(e) => {
-            tracing::debug!("Temporal edge schema not available: {}. Skipping edge save (will be available after schema migration).", e);
-            // This is expected if production DB hasn't migrated to temporal schema yet
+            tracing::debug!("Temporal edge schema not available: {}. Skipping edge save.", e);
             Ok(())
         }
     }