poimen-memory/crates/mem-cli/src/http_server.rs

use actix_web::{web, App, HttpServer, HttpResponse, HttpRequest, middleware::Logger};
use anyhow::Result;
use chrono;
use mem_llm::{EmbeddingsClient, RerankClient};
use mem_store::{init_schema, VectorStore};
use serde_json::json;
use sqlx::PgPool;
use std::sync::Arc;
use std::time::Instant;
use crate::ingest_worker::IngestWorker;
use serde::Deserialize;

/// JWT claims structure (extracted from deleted jwt_validator module)
/// Will be replaced by riotpiao-rust-sdk claims (issue #56)
#[derive(Debug, Clone, serde::Serialize, Deserialize)]
pub struct JwtClaims {
    pub sub: String,
    pub iss: String,
    pub aud: String,
    pub exp: i64,
    pub iat: i64,
    pub nbf: Option<i64>,
    pub permissions: Option<Vec<String>>,
    pub groups: Option<Vec<String>>,
    pub roles: Option<Vec<String>>,
}

/// Ingest request body
#[derive(Debug, Clone, Deserialize)]
pub struct IngestRequest {
    pub project: String,
    pub source: String,
    pub ingest_id: String,
    pub records: Vec<IngestRecord>,
}

#[derive(Debug, Clone, Deserialize)]
pub struct IngestRecord {
    pub text: String,
    #[serde(default)]
    pub role: Option<String>,
    #[serde(default)]
    pub timestamp: Option<String>,
    #[serde(default)]
    pub source_position: Option<i32>,
}
// RBAC removed for MVP - will add after core ingest/query working
use crate::handlers::{
    QueryParams, 
    LearnParams, build_learn_response,
    visualize_handler, visualize_stream_handler, compact_handler
};

/// Server state with database and workers
pub struct AppState {
    pub api_key: String,
    pub start_time: Instant,
    pub pool: PgPool,
    pub vector_store: Arc<VectorStore>,
    pub embeddings: Arc<EmbeddingsClient>,
    pub ingest_worker: Arc<IngestWorker>,
    pub auth_mode: AuthMode,
    /// M3.8 Query Optimizer (optional, from environment)
    pub optimizer_service: Option<Arc<mem_core::optimizer::OptimizerService>>,
}

/// Authentication mode
#[derive(Debug, Clone, Copy)]
pub enum AuthMode {
    Jwt,      // Validate JWT from Authentik
    ApiKey,   // Fallback to static API key
    None,     // No auth (testing only)
}

/// Auth extractor — validates JWT, apikey, or disabled
async fn validate_auth(req: &HttpRequest, state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    match state.auth_mode {
        AuthMode::Jwt => validate_jwt_token(req, state).await,
        AuthMode::ApiKey => validate_apikey(req, state),
        AuthMode::None => {
            tracing::warn!("Auth disabled - returning synthetic claims");
            let claims = JwtClaims {
                sub: "test-user".to_string(),
                iss: "test".to_string(),
                aud: "memory".to_string(),
                exp: i64::MAX,
                iat: chrono::Utc::now().timestamp(),
                nbf: None,
                permissions: Some(vec!["memory:write".to_string(), "memory:read".to_string()]),
                groups: Some(vec!["test".to_string()]),
                roles: None,
            };
            Ok((claims, "synthetic-token".to_string()))
        }
    }
}

/// Validate JWT token from Authorization header
/// NOTE: Full JWT validation deferred to riotpiao-rust-sdk migration (issue #56).
/// For now, extracts Bearer token and creates synthetic claims.
async fn validate_jwt_token(req: &HttpRequest, _state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    let auth_header = req
        .headers()
        .get("Authorization")
        .and_then(|h| h.to_str().ok())
        .ok_or_else(|| {
            HttpResponse::Unauthorized().json(json!({
                "error": "unauthorized",
                "reason": "missing Authorization header"
            }))
        })?
        .to_string();

    let token = auth_header
        .strip_prefix("Bearer ")
        .ok_or_else(|| {
            HttpResponse::Unauthorized().json(json!({
                "error": "unauthorized",
                "reason": "invalid Authorization header format, expected 'Bearer <token>'"
            }))
        })?
        .to_string();

    // Synthetic claims — real JWT validation will come with riotpiao-rust-sdk
    let claims = JwtClaims {
        sub: "jwt-user".to_string(),
        iss: "authentik".to_string(),
        aud: "memory".to_string(),
        exp: i64::MAX,
        iat: chrono::Utc::now().timestamp(),
        nbf: None,
        permissions: Some(vec!["*".to_string()]),
        groups: None,
        roles: Some(vec!["admin".to_string()]),
    };

    Ok((claims, token))
}

/// Fallback: validate apikey header
fn validate_apikey(req: &HttpRequest, state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    let api_key = req
        .headers()
        .get("apikey")
        .and_then(|h| h.to_str().ok())
        .map(|s| s.to_string());

    if api_key.as_ref() != Some(&state.api_key) {
        return Err(HttpResponse::Unauthorized().json(json!({
            "error": "unauthorized",
            "reason": "missing or invalid apikey header"
        })));
    }

    // Create a synthetic JWT claims for apikey mode (all permissions)
    let claims = JwtClaims {
        sub: "apikey-user".to_string(),
        iss: "internal".to_string(),
        aud: "memory".to_string(),
        exp: i64::MAX,
        iat: chrono::Utc::now().timestamp(),
        nbf: None,
        permissions: Some(vec!["*".to_string()]),
        groups: None,
        roles: Some(vec!["admin".to_string()]),  // API key gets admin role
    };

    Ok((claims, "apikey".to_string()))
}

/// Check if claims have required capability
fn has_capability(claims: &JwtClaims, required_capability: &str) -> bool {
    if let Some(perms) = &claims.permissions {
        // Wildcard permission grants everything
        if perms.contains(&"*".to_string()) {
            return true;
        }
        perms.contains(&required_capability.to_string())
    } else {
        false
    }
}

/// Extract client identifier from claims for rate limiting
#[allow(dead_code)]
fn extract_rate_limit_key(claims: &JwtClaims) -> String {
    // Use subject (user/service ID) as rate limit key
    claims.sub.clone()
}

/// Rate limit guard — stub until riotpiao-rust-sdk (issue #56)
fn check_rate_limit(_claims: &JwtClaims, _state: &AppState, _endpoint: &str) -> Result<(), HttpResponse> {
    // Rate limiting deferred to API gateway / riotpiao-rust-sdk
    Ok(())
}

/// Start HTTP server with database initialization
pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Result<()> {
    // Create connection pool
    let pool = PgPool::connect(database_url).await?;
    tracing::info!("Connected to database");

    // Initialize schema
    match init_schema(&pool).await {
        Ok(_) => tracing::info!("Schema initialized"),
        Err(e) => {
            tracing::warn!("Schema init error (may be non-fatal): {}", e);
            // Continue anyway - tables might exist
        }
    }

    // Create workers
    let vector_store = Arc::new(VectorStore::new(pool.clone()));
    let embeddings = Arc::new(EmbeddingsClient::from_env()?);
    let ingest_worker = Arc::new(IngestWorker::new(pool.clone(), (*embeddings).clone()));
    let _reranker = RerankClient::from_env()?;

    // Determine auth mode
    let auth_mode = std::env::var("MEM_AUTH_MODE")
        .unwrap_or_else(|_| "apikey".to_string())
        .to_lowercase();
    let auth_mode = match auth_mode.as_str() {
        "jwt" => AuthMode::Jwt,
        "apikey" => AuthMode::ApiKey,
        "none" => AuthMode::None,
        _ => {
            tracing::warn!("Unknown auth mode: {}, defaulting to apikey", auth_mode);
            AuthMode::ApiKey
        }
    };

    // JWT auth will be handled by riotpiao-rust-sdk (issue #56)
    if matches!(auth_mode, AuthMode::Jwt) {
        tracing::warn!("JWT auth mode selected but JwtValidator removed. Use riotpiao-rust-sdk (issue #56).");
    }

    // Initialize M3.8 Query Optimizer if enabled
    let optimizer_service = match mem_core::optimizer::OptimizerServiceBuilder::new().build() {
        Ok(service) => {
            tracing::info!("M3.8 Query Optimizer enabled");
            Some(Arc::new(service))
        }
        Err(e) => {
            tracing::debug!("M3.8 Query Optimizer not available: {}", e);
            None
        }
    };

    // Queue adapter + dual-write will use riotpiao-rust-sdk (issue #56)

    let state = web::Data::new(AppState {
        api_key,
        start_time: Instant::now(),
        pool,
        vector_store,
        embeddings,
        ingest_worker,
        auth_mode,
        optimizer_service,
    });

    tracing::info!("Starting HTTP server on port {}", port);

    // O5/O7/O9: Background stats collector (every 60s)
    {
        let stats_pool = state.get_ref().pool.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
            loop {
                interval.tick().await;
                // O5: Table row counts
                if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM memory_entity")
                    .fetch_one(&stats_pool).await {
                    crate::metrics::DB_TABLE_ENTITY_ROWS.set(row.0 as u64);
                }
                if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM memory_edge")
                    .fetch_one(&stats_pool).await {
                    crate::metrics::DB_TABLE_EDGE_ROWS.set(row.0 as u64);
                }
                // O9: Pool stats
                crate::metrics::DB_POOL_SIZE.set(stats_pool.size() as u64);
                crate::metrics::DB_POOL_IDLE.set(stats_pool.num_idle() as u64);
            }
        });
    }

    tracing::info!("Creating HttpServer instance...");

    let server = HttpServer::new(move || {
        tracing::debug!("HttpServer::new() closure executing");
        App::new()
            .app_data(state.clone())
            .wrap(Logger::default())
            .route("/health", web::get().to(health_check))
        .route("/ready", web::get().to(readiness_check))
            .route("/metrics", web::get().to(crate::metrics::metrics_handler))
            .route("/memory/ingest", web::post().to(ingest_handler))
            .route("/memory/ingest/{ingest_id}", web::get().to(ingest_status))
            .route("/memory/query", web::get().to(query_handler))
            .route("/memory/query", web::post().to(crate::handlers::unified_query::unified_query_handler))
            .route("/memory/query/semantic/entities", web::post().to(crate::handlers::semantic::search_entities_handler))
            .route("/memory/query/semantic/edges", web::post().to(crate::handlers::semantic::search_edges_handler))
            .route("/memory/query/hybrid", web::post().to(crate::handlers::semantic::hybrid_search_handler))
            // context_handler removed — will be reimplemented with riotpiao-rust-sdk (issue #56)
            .route("/memory/projects", web::get().to(projects_handler))
            .route("/memory/skills", web::get().to(skills_handler))
            .route("/memory/learn", web::post().to(learn_handler))
            .route("/memory/vault/generate", web::post().to(vault_generate_handler))
            .route("/memory/vault", web::get().to(vault_browser_handler))
            .route("/memory/vault/{project}", web::get().to(vault_project_handler))
            .route("/memory/vault/{project}/{file}", web::get().to(vault_file_handler))
            .route("/memory/visualize", web::post().to(visualize_handler))
            .route("/memory/visualize/stream", web::post().to(visualize_stream_handler))
            .route("/memory/compact", web::post().to(compact_handler))
            .route("/memory/synthesis/link-entities", web::post().to(crate::handlers::synthesis::link_entities_handler))
            .route("/memory/synthesis/detect-aliases", web::post().to(crate::handlers::synthesis::detect_aliases_handler))
            .route("/memory/synthesis/suggest-merges", web::post().to(crate::handlers::synthesis::suggest_merges_handler))
            .route("/memory/synthesis/detect-coreferences", web::post().to(crate::handlers::synthesis::detect_coreferences_handler))
            .route("/memory/synthesis/infer", web::post().to(crate::handlers::synthesis::infer_facts_handler))
            .route("/memory/synthesis/transitive-closure", web::post().to(crate::handlers::synthesis::transitive_closure_handler))
            .route("/memory/synthesis/reasoning-paths", web::post().to(crate::handlers::synthesis::reasoning_paths_handler))
            .route("/memory/synthesis/reason", web::post().to(crate::handlers::synthesis::reason_query_handler))
            .route("/memory/synthesis/summarize", web::post().to(crate::handlers::synthesis::summarize_handler))
            .route("/memory/synthesis", web::post().to(crate::handlers::unified_synthesis::unified_synthesis_handler))
            .route("/agents", web::post().to(crate::handlers::agent_handler::register_agent_handler))
            .route("/agents/{id}", web::get().to(crate::handlers::agent_handler::get_agent_handler))
            .route("/agents/{id}", web::put().to(crate::handlers::agent_handler::update_agent_handler))
            .route("/agents/{id}", web::delete().to(crate::handlers::agent_handler::delete_agent_handler))
            .route("/agents/{id}/metrics", web::get().to(crate::handlers::agent_handler::get_agent_metrics_handler))
            .route("/agents/{id}/prompts", web::post().to(crate::handlers::agent_handler::create_prompt_handler))
            .route("/agents/{id}/roles", web::post().to(crate::handlers::agent_handler::map_role_to_prompt_handler))
            .route("/agents/{id}/roles/{role_name}/prompts", web::get().to(crate::handlers::agent_handler::get_role_prompts_handler))
    });
    
    tracing::info!("HttpServer instance created, binding to 0.0.0.0:{}", port);
    let server = server.bind(("0.0.0.0", port))?;
    tracing::info!("Successfully bound to port {}, about to run", port);
    
    server.run().await?;

    Ok(())
}

/// Health check (no auth)
pub async fn health_check(state: web::Data<AppState>) -> HttpResponse {
    use crate::metrics::*;
    HEALTH_CHECKS_TOTAL.inc();
    let uptime = state.start_time.elapsed().as_secs();
    APP_UPTIME_SECONDS.set(uptime);

    // O7: Check DB dependency
    let db_start = std::time::Instant::now();
    match sqlx::query("SELECT 1").execute(&state.pool).await {
        Ok(_) => {
            DEP_DB_UP.set(1);
            DEP_DB_LATENCY.observe(db_start.elapsed().as_secs_f64());
        }
        Err(_) => {
            DEP_DB_UP.set(0);
            HEALTH_CHECK_FAILURES.inc();
        }
    }

    HttpResponse::Ok().json(json!({"status": "ok", "uptime_seconds": uptime}))
}

/// GET /ready — readiness probe (checks DB)
pub async fn readiness_check(state: web::Data<AppState>) -> HttpResponse {
    let uptime = state.start_time.elapsed().as_secs();
    let db_start = std::time::Instant::now();
    match sqlx::query("SELECT 1").execute(&state.pool).await {
        Ok(_) => {
            crate::metrics::DEP_DB_UP.set(1);
            crate::metrics::DEP_DB_LATENCY.observe(db_start.elapsed().as_secs_f64());
            HttpResponse::Ok().json(json!({"status": "ready", "uptime_seconds": uptime, "db": "ok"}))
        }
        Err(e) => {
            crate::metrics::DEP_DB_UP.set(0);
            crate::metrics::HEALTH_CHECK_FAILURES.inc();
            HttpResponse::ServiceUnavailable().json(json!({"status": "not_ready", "uptime_seconds": uptime, "db": format!("error: {}", e)}))
        }
    }
}

/// POST /memory/ingest — queue an ingest job
pub async fn ingest_handler(
    req: HttpRequest,
    body: web::Json<IngestRequest>,
    state: web::Data<AppState>,
) -> HttpResponse {
    use crate::metrics::*;
    INGEST_REQUESTS_TOTAL.inc();
    INGEST_IN_FLIGHT.inc();
    let _timer = Timer::new(&INGEST_DURATION);

    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => {
            INGEST_AUTH_FAILURES.inc();
            INGEST_ERRORS_TOTAL.inc();
            ERROR_AUTH_FAILURE_INGEST.inc();
            INGEST_IN_FLIGHT.dec();
            return e;
        }
    };

    let _user_id = &claims.sub;
    if !has_capability(&claims, "memory:write") {
        INGEST_AUTH_FAILURES.inc();
        INGEST_ERRORS_TOTAL.inc();
        ERROR_FORBIDDEN_INGEST.inc();
        INGEST_IN_FLIGHT.dec();
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
        INGEST_RATE_LIMITED.inc();
        ERROR_RATE_LIMITED_INGEST.inc();
        INGEST_IN_FLIGHT.dec();
        return e;
    }

    // Idempotency check via DB (ingest_id is UNIQUE)
    // In-memory idempotency store removed; DB ON CONFLICT handles dedup

    let byte_count: usize = body.records.iter().map(|r| r.text.len()).sum();
    INGEST_BYTES_TOTAL.inc_by(byte_count as u64);
    INGEST_RECORDS_TOTAL.inc_by(body.records.len() as u64);

    // Extract X-Forward-User header for LLM auth (API Gateway pattern)
    let x_forward_user = req
        .headers()
        .get("X-Forward-User")
        .and_then(|h| h.to_str().ok())
        .map(|s| s.to_string());
    
    if let Some(ref user) = x_forward_user {
        tracing::info!("Ingest request with X-Forward-User: {}", user);
    }

    // Execute ingest
    let resp = execute_ingest(&state, &body, x_forward_user).await;
    INGEST_IN_FLIGHT.dec();
    resp
}

/// Execute ingest job creation and spawn worker
async fn execute_ingest(
    state: &web::Data<AppState>,
    body: &IngestRequest,
    x_forward_user: Option<String>,
) -> HttpResponse {
    let records: Vec<(String, String)> = body.records
        .iter()
        .map(|r| (r.text.clone(), body.source.clone()))
        .collect();

    let job_result = sqlx::query(
        "INSERT INTO ingest_jobs (id, project, ingest_id, status, created_at)
         VALUES ($1, $2, $3, 'pending', NOW())
         ON CONFLICT (ingest_id) DO NOTHING
         RETURNING id",
    )
    .bind(uuid::Uuid::new_v4())
    .bind(&body.project)
    .bind(&body.ingest_id)
    .fetch_optional(&state.pool)
    .await;

    let response = json!({
        "ingest_id": body.ingest_id,
        "status": "pending",
        "status_url": format!("/memory/ingest/{}", body.ingest_id)
    });

    match job_result {
        Ok(Some(_)) => {
            // Spawn async ingest task
            let worker = state.ingest_worker.clone();
            let project = body.project.clone();
            let ingest_id = body.ingest_id.clone();
            let x_fwd = x_forward_user.clone();
            tokio::spawn(async move {
                if let Err(e) = worker.process_ingest_with_auth(&project, &ingest_id, records, x_fwd).await {
                    tracing::error!("Ingest failed: {}", e);
                }
            });
            HttpResponse::Accepted().json(response)
        }
        Ok(None) => {
            // Already exists (concurrent insert — DB UNIQUE constraint)
            HttpResponse::Accepted().json(response)
        }
        Err(e) => {
            crate::metrics::ERROR_UNEXPECTED_INGEST.inc();
            crate::metrics::ERROR_UNEXPECTED_TOTAL.inc();
            tracing::error!(user_id = body.project.as_str(), "Unexpected DB error during ingest: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/ingest/{ingest_id} — check ingest status
pub async fn ingest_status(
    req: HttpRequest,
    ingest_id: web::Path<String>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let id = ingest_id.into_inner();
    let result = sqlx::query_as::<_, (String, String, Option<String>)>(
        "SELECT ingest_id, status, error FROM ingest_jobs WHERE ingest_id = $1",
    )
    .bind(&id)
    .fetch_optional(&state.pool)
    .await;

    match result {
        Ok(Some((ingest_id, status, error))) => {
            HttpResponse::Ok().json(json!({
                "ingest_id": ingest_id,
                "status": status,
                "error": error
            }))
        }
        Ok(None) => {
            HttpResponse::NotFound().json(json!({"error": "not_found"}))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}


/// POST /memory/learn — Ingest knowledge via gated loop (LLM evaluates + compacts)
///
/// OpenAI-compatible style endpoint. Accepts markdown text, chunks it,
/// runs each chunk through the gated loop where the LLM decides whether
/// to accept/reject and rewrites memory to stay compact.
///
/// Request:
///   POST /memory/learn
///   { "project": "knowledge", "text": "## Rust\n- ownership...", "query": "What are key Rust patterns?" }
///
/// Response:
///   { "project": "knowledge", "chunks_seen": 5, "chunks_used": 3,
///     "memory": "compacted memory text...", "status": "completed" }
pub async fn learn_handler(
    req: HttpRequest,
    body: web::Json<serde_json::Value>,
    state: web::Data<AppState>,
) -> HttpResponse {
    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };
    if !has_capability(&claims, "memory:write") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
        return e;
    }

    // Parse + validate params
    let params = match LearnParams::from_body(&body) {
        Ok(p) => p,
        Err(e) => return e.to_response(),
    };

    // Chunk the markdown
    let chunks = chunk_markdown_text(&params.text, params.chunk_size);
    if chunks.is_empty() {
        return HttpResponse::BadRequest().json(json!({
            "error": "bad_request",
            "reason": "text produced no chunks"
        }));
    }

    // Build domain chunks
    let domain_chunks: Vec<mem_core::domain::Chunk> = chunks
        .iter()
        .enumerate()
        .map(|(i, text)| {
            let record = mem_core::Record {
                role: mem_core::Role::User,
                text: text.clone(),
                timestamp: time::OffsetDateTime::now_utc(),
                provenance: mem_core::Provenance {
                    source_id: format!("learn://{}:{}", params.project, i),
                    offset: i as u64,
                },
            };
            mem_core::domain::Chunk::new((i + 1) as u32, vec![record], text.len() / 4)
        })
        .collect();

    // Build query for gated loop
    let query = mem_core::query::Query {
        id: format!("learn-{}", uuid::Uuid::new_v4()),
        question: params.question.clone(),
        exit_gate: false,
    };

    let config = mem_core::gated_loop::LoopConfig {
        level: mem_core::Level::L1,
        query,
        memory_budget: params.memory_budget,
        use_exit_gate: false,
    };

    // Create LLM client
    let llm_base = std::env::var("LLM_API_BASE")
        .unwrap_or_else(|_| "https://api.riotpiao.com".to_string());
    let llm_key = std::env::var("LLM_API_KEY")
        .or_else(|_| std::env::var("MEM_API_KEY"))
        .unwrap_or_default();
    let llm = match mem_llm::ChatClient::new(&llm_base, &llm_key, &params.model) {
        Ok(c) => c,
        Err(e) => return HttpResponse::InternalServerError().json(json!({
            "error": "llm_init_failed",
            "reason": e.to_string()
        })),
    };

    // Run gated loop
    let outcome = match mem_core::gated_loop::run_loop(config, domain_chunks, &llm) {
        Ok(o) => o,
        Err(e) => return HttpResponse::InternalServerError().json(json!({
            "error": "gated_loop_failed",
            "reason": e.to_string()
        })),
    };

    // Store compacted memory
    let stored = store_compacted_memory(&state, &params.project, &outcome.final_memory).await;

    build_learn_response(
        &params.project,
        &params.model,
        outcome.chunks_seen,
        outcome.chunks_used,
        &outcome.final_memory,
        stored,
    )
}

/// Store compacted memory in pgvector
async fn store_compacted_memory(
    state: &web::Data<AppState>,
    project: &str,
    memory: &str,
) -> bool {
    if memory.is_empty() {
        return false;
    }

    let embedding = match state.embeddings.embed_one(memory).await {
        Ok(e) => e,
        Err(e) => {
            tracing::error!("Failed to embed compacted memory: {}", e);
            return false;
        }
    };

    let sha = {
        use sha2::{Digest, Sha256};
        let mut h = Sha256::new();
        h.update(memory.as_bytes());
        format!("{:x}", h.finalize())
    };

    let result = sqlx::query(
        "INSERT INTO memory_chunks (id, project, level, text, embedding, sha256, source, created_at)
         VALUES ($1, $2, 'L1', $3, $4, $5, $6, NOW())
         ON CONFLICT (sha256) DO UPDATE SET text = $3, embedding = $4",
    )
    .bind(uuid::Uuid::new_v4())
    .bind(project)
    .bind(memory)
    .bind(embedding.to_vec())
    .bind(&sha)
    .bind(format!("learn://{}", project))
    .fetch_optional(&state.pool)
    .await;

    match result {
        Ok(_) => {
            crate::metrics::WRITE_CHUNKS_TOTAL.inc();
            crate::metrics::WRITE_BYTES_TOTAL.inc_by(memory.len() as u64);
            true
        }
        Err(e) => {
            crate::metrics::WRITE_ERRORS_TOTAL.inc();
            tracing::error!("Failed to store compacted memory: {}", e);
            false
        }
    }
}

/// Split markdown on ## headings for learn endpoint.
fn chunk_markdown_text(content: &str, max_chunk: usize) -> Vec<String> {
    let mut chunks = Vec::new();
    let mut current = String::new();
    for line in content.lines() {
        if line.starts_with("## ") && !current.is_empty() {
            let trimmed = current.trim().to_string();
            if !trimmed.is_empty() { chunks.push(trimmed); }
            current = String::new();
        }
        current.push_str(line);
        current.push('\n');
        if current.len() > max_chunk {
            let trimmed = current.trim().to_string();
            if !trimmed.is_empty() { chunks.push(trimmed); }
            current = String::new();
        }
    }
    let trimmed = current.trim().to_string();
    if !trimmed.is_empty() { chunks.push(trimmed); }
    chunks
}

/// GET /memory/query — semantic search across memories
pub async fn query_handler(
    req: HttpRequest,
    query: web::Query<std::collections::HashMap<String, String>>,
    state: web::Data<AppState>,
) -> HttpResponse {
    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/query") {
        return e;
    }

    // Parse + validate params
    let params = match QueryParams::from_query(&query) {
        Ok(p) => p,
        Err(e) => return e.to_response(),
    };

    // Execute temporal graph query
    match query_temporal_graph(&state, &params).await {
        Ok(response) => HttpResponse::Ok().json(response),
        Err(e) => {
            crate::metrics::ERROR_UNEXPECTED_QUERY.inc();
            crate::metrics::ERROR_UNEXPECTED_TOTAL.inc();
            tracing::error!(user_id = claims.sub.as_str(), "Unexpected error: temporal graph query failed: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "query_failed", "reason": e.to_string()}))
        }
    }
}


/// GET /memory/projects — list projects with memory
pub async fn projects_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    if let Err(e) = check_rate_limit(&claims, &state, "/memory/projects") {
        return e;
    }

    let result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l2 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let projects: Vec<String> = rows.into_iter().map(|(p,)| p).collect();
            HttpResponse::Ok().json(json!({
                "projects": projects,
                "count": projects.len()
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/skills — list extracted skills
pub async fn skills_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let result = sqlx::query_as::<_, (String, String, String)>(
        "SELECT name, description, when_to_use FROM skills ORDER BY created_at DESC LIMIT 50",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let skills: Vec<serde_json::Value> = rows
                .into_iter()
                .map(|(name, desc, when_to_use)| {
                    json!({
                        "name": name,
                        "description": desc,
                        "when_to_use": when_to_use
                    })
                })
                .collect();
            HttpResponse::Ok().json(json!({
                "skills": skills,
                "count": skills.len()
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}


/// POST /memory/vault/generate — generate Obsidian vault from memories
pub async fn vault_generate_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check write capability
    if !has_capability(&claims, "memory:write") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }

    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    
    // Get all projects from database
    let projects_result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l1 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match projects_result {
        Ok(projects) => {
            let mut generated = 0;
            let mut errors = Vec::new();

            for (project,) in projects {
                let project_vault_dir = format!("{}/vault/{}", vault_dir, project);
                
                // Create project directory
                if let Err(e) = std::fs::create_dir_all(&project_vault_dir) {
                    errors.push(format!("Failed to create {}: {}", project_vault_dir, e));
                    continue;
                }

                // Get all L1 memories for this project
                let l1s_result = sqlx::query_as::<_, (String, String, String)>(
                    "SELECT id, query_id, content FROM memories_l1 WHERE project = $1 ORDER BY updated_at DESC",
                )
                .bind(&project)
                .fetch_all(&state.pool)
                .await;

                match l1s_result {
                    Ok(l1s) => {
                        for (id, query_id, content) in l1s {
                            let filename = format!("{}/{}.md", project_vault_dir, query_id);
                            let note = format!(
                                "---\nproject: {}\nlevel: L1\nquery_id: {}\nid: {}\nupdated: {}\n---\n\n{}",
                                project,
                                query_id,
                                id,
                                chrono::Utc::now().to_rfc3339(),
                                content
                            );
                            if std::fs::write(&filename, note).is_ok() {
                                generated += 1;
                            }
                        }
                    }
                    Err(e) => {
                        errors.push(format!("Failed to fetch L1s for {}: {}", project, e));
                    }
                }
                
                // Get L2 synthesis
                let l2_result = sqlx::query_as::<_, (String,)>(
                    "SELECT content FROM memories_l2 WHERE project = $1",
                )
                .bind(&project)
                .fetch_optional(&state.pool)
                .await;

                if let Ok(Some((content,))) = l2_result {
                    let filename = format!("{}/index.md", project_vault_dir);
                    let note = format!(
                        "---\nproject: {}\nlevel: L2\ntitle: {} Synthesis\nupdated: {}\n---\n\n{}",
                        project,
                        project,
                        chrono::Utc::now().to_rfc3339(),
                        content
                    );
                    if std::fs::write(&filename, note).is_ok() {
                        generated += 1;
                    }
                }
            }

            HttpResponse::Ok().json(json!({
                "status": "generated",
                "vault_dir": format!("{}/vault", vault_dir),
                "notes_created": generated,
                "errors": errors
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/vault — list all projects with vault browser UI
pub async fn vault_browser_handler(
    req: HttpRequest,
    query: web::Query<std::collections::HashMap<String, String>>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    // If project parameter provided, return file tree for that project
    if let Some(project) = query.get("project") {
        return vault_project_tree(project, &state).await;
    }

    // Otherwise return all projects
    let result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l1 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let projects: Vec<serde_json::Value> = rows
                .into_iter()
                .map(|(p,)| json!({"name": p}))
                .collect();
            HttpResponse::Ok().json(json!({
                "projects": projects
            }))
        }
        Err(e) => {
            tracing::error!("Error fetching projects: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// Helper: Build file tree for a project
async fn vault_project_tree(
    project: &str,
    _state: &web::Data<AppState>,
) -> HttpResponse {
    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    let project_path = format!("{}/vault/{}", vault_dir, project);

    match std::fs::read_dir(&project_path) {
        Ok(entries) => {
            let mut files = Vec::new();
            
            for entry in entries {
                if let Ok(e) = entry {
                    if let Ok(metadata) = e.metadata() {
                        if let Some(name) = e.file_name().to_str() {
                            if name.ends_with(".md") && !name.starts_with(".") {
                                let path = format!("{}/{}", project, name);
                                let title = name.replace(".md", "").replace("-", " ");
                                let updated = metadata
                                    .modified()
                                    .ok()
                                    .and_then(|t| {
                                        let elapsed = t.elapsed().ok()?;
                                        Some(chrono::Local::now() - chrono::Duration::from_std(elapsed).ok()?)
                                    })
                                    .map(|t| t.to_rfc3339())
                                    .unwrap_or_else(|| "unknown".to_string());
                                
                                files.push(json!({
                                    "path": path,
                                    "name": name,
                                    "title": title,
                                    "updated_at": updated
                                }));
                            }
                        }
                    }
                }
            }
            
            files.sort_by(|a, b| {
                let a_name = a["name"].as_str().unwrap_or("");
                let b_name = b["name"].as_str().unwrap_or("");
                a_name.cmp(b_name)
            });

            HttpResponse::Ok().json(json!({
                "project": project,
                "files": files
            }))
        }
        Err(_) => {
            HttpResponse::NotFound().json(json!({
                "error": "project_not_found",
                "project": project
            }))
        }
    }
}

/// GET /memory/vault/{project} — list files in project vault
pub async fn vault_project_handler(
    req: HttpRequest,
    project: web::Path<String>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let proj = project.into_inner();
    vault_project_tree(&proj, &state).await
}

/// GET /memory/vault/{project}/{file} — view markdown file
pub async fn vault_file_handler(
    req: HttpRequest,
    path: web::Path<(String, String)>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let (project, file) = path.into_inner();
    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    let file_path = format!("{}/vault/{}/{}.md", vault_dir, project, file);

    // Security: prevent path traversal
    if file.contains("..") || file.contains("/") {
        return HttpResponse::BadRequest().json(json!({
            "error": "invalid_filename"
        }));
    }

    match std::fs::read_to_string(&file_path) {
        Ok(content) => {
            // Parse frontmatter and body
            let (frontmatter, body) = if content.starts_with("---") {
                let parts: Vec<&str> = content.split("---").collect();
                if parts.len() >= 3 {
                    (parts[1].to_string(), parts[2..].join("---"))
                } else {
                    ("".to_string(), content.clone())
                }
            } else {
                ("".to_string(), content.clone())
            };

            // Parse YAML frontmatter into JSON
            let metadata = if !frontmatter.is_empty() {
                frontmatter
                    .lines()
                    .filter_map(|line| {
                        let parts: Vec<&str> = line.splitn(2, ':').collect();
                        if parts.len() == 2 {
                            Some((parts[0].trim().to_string(), parts[1].trim().to_string()))
                        } else {
                            None
                        }
                    })
                    .collect::<std::collections::HashMap<_, _>>()
            } else {
                std::collections::HashMap::new()
            };

            let mut meta_json = json!({});
            for (k, v) in metadata {
                meta_json[&k] = json!(v);
            }

            HttpResponse::Ok().json(json!({
                "project": project,
                "file": format!("{}.md", file),
                "path": format!("{}/{}.md", project, file),
                "title": file.replace("-", " "),
                "metadata": meta_json,
                "content": body.trim()
            }))
        }
        Err(_) => {
            HttpResponse::NotFound().json(json!({
                "error": "file_not_found",
                "project": project,
                "file": file
            }))
        }
    }
}

/// Query temporal knowledge graph
/// 1. Find entities via semantic search
/// 2. Traverse edges from entities
/// 3. Apply temporal filtering (t_valid/t_invalid)
/// 4. Return graph with confidence scores
async fn query_temporal_graph(
    state: &web::Data<AppState>,
    params: &QueryParams,
) -> anyhow::Result<serde_json::Value> {
    // Step 1: Find entities matching the question
    // Use keyword search (ILIKE) on name + description for GET endpoint.
    // POST /memory/query uses the full semantic retriever with embeddings.
    let search_pattern = format!("%{}%", params.question);
    let entities_rows: Vec<(String, String, String)> = sqlx::query_as(
        "SELECT id::TEXT, name, entity_type FROM memory_entity \
         WHERE project_id = $1 AND t_expired IS NULL \
           AND (name ILIKE $3 OR COALESCE(description, '') ILIKE $3 OR COALESCE(summary, '') ILIKE $3) \
         ORDER BY confidence DESC \
         LIMIT $2"
    )
    .bind(&params.project)
    .bind(params.limit as i32)
    .bind(&search_pattern)
    .fetch_all(&state.pool)
    .await
    .unwrap_or_default();
    
    // Step 2: Traverse edges from found entities
    // NOTE: Edges will be empty until temporal schema is migrated
    let mut edges_data: Vec<(String, String, String, String, String, f32)> = Vec::new();
    
    // Try to fetch edges (will be empty if schema not migrated yet)
    for (entity_id, _name, _type_str) in &entities_rows {
        let entity_edges: Vec<(String, String, String, String, f32, Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>)> = 
            sqlx::query_as(
                "SELECT id::TEXT, target_id::TEXT, relation_type, fact, confidence, t_valid, t_invalid FROM memory_edge WHERE project_id = $1 AND source_id = $2::UUID"
            )
            .bind(&params.project)
            .bind(entity_id)
            .fetch_all(&state.pool)
            .await
            .unwrap_or_default();  // Returns empty vec if table schema doesn't match
        
        for (id, target, rel, fact, conf, t_valid, t_invalid) in entity_edges {
            // Apply temporal filtering
            let now = chrono::Utc::now();
            let valid = t_valid.as_ref().map(|t| *t <= now).unwrap_or(true);
            let not_invalid = t_invalid.as_ref().map(|t| *t > now).unwrap_or(true);
            
            if valid && not_invalid {
                edges_data.push((id, entity_id.clone(), target, rel, fact, conf));
            }
        }
    }
    
    // Step 4: Build response
    let response = json!({
        "query": params.question,
        "project": params.project,
        "entities": entities_rows.iter().map(|(id, name, etype)| json!({
            "id": id,
            "name": name,
            "type": etype
        })).collect::<Vec<_>>(),
        "edges": edges_data.iter().map(|(id, src, tgt, rel, fact, conf)| json!({
            "id": id,
            "source": src,
            "target": tgt,
            "relation": rel,
            "fact": fact,
            "confidence": conf
        })).collect::<Vec<_>>(),
        "count": json!({
            "entities": entities_rows.len(),
            "edges": edges_data.len()
        })
    });
    
    Ok(response)
}