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::endpoints::IngestRequest;
use crate::ingest_worker::IngestWorker;
use crate::query_worker::QueryWorker;
use crate::rate_limiter::{RateLimiter, LimitConfig};
use crate::idempotency::IdempotencyStore;

/// 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 query_worker: Arc<QueryWorker>,
    pub rate_limiter: Arc<RateLimiter>,
    pub idempotency_store: Arc<IdempotencyStore>,
}

/// Auth extractor — validates apikey header
fn check_auth(req: &HttpRequest, state: &AppState) -> Result<(), 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 apikey header"})));
    }
    Ok(())
}

/// Extract apikey from request
fn extract_apikey(req: &HttpRequest) -> Option<String> {
    req.headers()
        .get("apikey")
        .and_then(|h| h.to_str().ok())
        .map(|s| s.to_string())
}

/// Rate limit guard — call this in handlers to check rate limit
fn check_rate_limit(req: &HttpRequest, state: &AppState, endpoint: &str) -> Result<(), HttpResponse> {
    let apikey = extract_apikey(req).unwrap_or_else(|| "unknown".to_string());
    
    match state.rate_limiter.check(&apikey, endpoint) {
        Ok(_) => Ok(()),
        Err(rate_limit_err) => {
            let retry_after = rate_limit_err.retry_after_seconds.to_string();
            Err(HttpResponse::TooManyRequests()
                .insert_header(("Retry-After", retry_after))
                .json(json!({
                    "error": "rate_limit_exceeded",
                    "reason": rate_limit_err.reason.clone(),
                    "retry_after_seconds": rate_limit_err.retry_after_seconds,
                    "limit_window": format!("{}s", rate_limit_err.limit_window_secs),
                })))
        }
    }
}

/// 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
    init_schema(&pool).await?;
    tracing::info!("Schema initialized");

    // 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()?;
    let query_worker = Arc::new(QueryWorker::new(VectorStore::new(pool.clone()), (*embeddings).clone(), reranker));

    // Initialize rate limiter and idempotency store
    let limit_config = LimitConfig {
        ingest_per_hour: std::env::var("MEM_RATE_LIMIT_INGEST")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(100.0),
        query_per_hour: std::env::var("MEM_RATE_LIMIT_QUERY")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(1000.0),
        projects_per_hour: std::env::var("MEM_RATE_LIMIT_PROJECTS")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(100.0),
        burst_per_second: std::env::var("MEM_RATE_LIMIT_BURST")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(10.0),
    };
    let rate_limiter = Arc::new(RateLimiter::new(limit_config));

    let idempotency_ttl = std::env::var("MEM_IDEMPOTENCY_TTL_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(86400); // 24 hours default
    let idempotency_store = Arc::new(IdempotencyStore::new(idempotency_ttl));

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

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

    HttpServer::new(move || {
        App::new()
            .app_data(state.clone())
            .wrap(Logger::default())
            .route("/health", web::get().to(health_check))
            .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/projects", web::get().to(projects_handler))
            .route("/memory/skills", web::get().to(skills_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))
    })
    .bind(("0.0.0.0", port))?
    .run()
    .await?;

    Ok(())
}

/// Health check (no auth)
pub async fn health_check(state: web::Data<AppState>) -> HttpResponse {
    let uptime = state.start_time.elapsed().as_secs();
    HttpResponse::Ok().json(json!({"status": "ok", "uptime_seconds": uptime}))
}

/// POST /memory/ingest — queue an ingest job
pub async fn ingest_handler(
    req: HttpRequest,
    body: web::Json<IngestRequest>,
    state: web::Data<AppState>,
) -> HttpResponse {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

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

    let project = body.project.clone();
    let ingest_id = body.ingest_id.clone();
    let records: Vec<(String, String)> = body
        .records
        .iter()
        .map(|r| (r.text.clone(), body.source.clone()))
        .collect();

    // Check idempotency cache first
    if let Some(cached_response) = state.idempotency_store.get(&ingest_id) {
        tracing::info!("Returning cached response for ingest_id: {}", ingest_id);
        return HttpResponse::Accepted().json(cached_response);
    }

    // Create ingest job in DB
    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(&project)
    .bind(&ingest_id)
    .fetch_optional(&state.pool)
    .await;

    match job_result {
        Ok(Some(_)) => {
            // Spawn async ingest task
            let worker = state.ingest_worker.clone();
            let proj = project.clone();
            let id = ingest_id.clone();
            tokio::spawn(async move {
                if let Err(e) = worker.process_ingest(&proj, &id, records).await {
                    tracing::error!("Ingest failed: {}", e);
                }
            });

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

            // Cache the response for idempotency
            state.idempotency_store.set(ingest_id.clone(), response.clone());

            HttpResponse::Accepted().json(response)
        }
        Ok(None) => {
            // Already exists in DB (was inserted concurrently)
            let response = json!({
                "ingest_id": ingest_id,
                "status": "pending",
                "status_url": format!("/memory/ingest/{}", ingest_id)
            });
            state.idempotency_store.set(ingest_id.clone(), response.clone());
            HttpResponse::Accepted().json(response)
        }
        Err(e) => {
            tracing::error!("DB error: {}", 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 {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

    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"}))
        }
    }
}

/// 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 {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

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

    let project = match query.get("project") {
        Some(p) => p.clone(),
        None => {
            return HttpResponse::BadRequest().json(json!({"error": "missing project parameter"}))
        }
    };

    let question = match query.get("query") {
        Some(q) => q.clone(),
        None => {
            return HttpResponse::BadRequest().json(json!({"error": "missing query parameter"}))
        }
    };

    let limit = query
        .get("limit")
        .and_then(|l| l.parse::<i64>().ok())
        .unwrap_or(5);

    match state.query_worker.query(&project, &question, Some(limit)).await {
        Ok(results) => {
            HttpResponse::Ok().json(json!({
                "query": question,
                "project": project,
                "results": results
            }))
        }
        Err(e) => {
            tracing::error!("Query failed: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "query_failed"}))
        }
    }
}

/// GET /memory/projects — list projects with memory
pub async fn projects_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

    if let Err(e) = check_rate_limit(&req, &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 {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

    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 {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

    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,
    state: web::Data<AppState>,
) -> HttpResponse {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

    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<String> = rows.into_iter().map(|(p,)| p).collect();
            let html = format!(
                r#"<!DOCTYPE html>
<html>
<head>
    <title>Poimen Memory Vault</title>
    <style>
        body {{ font-family: sans-serif; margin: 20px; background: #f5f5f5; }}
        h1 {{ color: #333; }}
        .project {{ background: white; padding: 10px; margin: 10px 0; border-radius: 5px; }}
        .project a {{ color: #0066cc; text-decoration: none; }}
        .project a:hover {{ text-decoration: underline; }}
    </style>
</head>
<body>
    <h1>📚 Poimen Memory Vault</h1>
    <p>Projects with stored memories:</p>
    {}
</body>
</html>"#,
                projects
                    .iter()
                    .map(|p| format!(r#"<div class="project"><a href="/memory/vault/{}">{}</a></div>"#, p, p))
                    .collect::<Vec<_>>()
                    .join("\n")
            );
            HttpResponse::Ok()
                .content_type("text/html; charset=utf-8")
                .body(html)
        }
        Err(_) => {
            HttpResponse::InternalServerError().body("Database error")
        }
    }
}

/// 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 {
    if let Err(e) = check_auth(&req, &state) {
        return e;
    }

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

    match std::fs::read_dir(&project_path) {
        Ok(entries) => {
            let files: Vec<String> = entries
                .filter_map(|e| e.ok())
                .filter_map(|e| {
                    e.file_name()
                        .to_str()
                        .filter(|n| n.ends_with(".md"))
                        .map(|n| n.to_string())
                })
                .collect();

            let html = format!(
                r#"<!DOCTYPE html>
<html>
<head>
    <title>Vault: {}</title>
    <style>
        body {{ font-family: sans-serif; margin: 20px; background: #f5f5f5; }}
        h1 {{ color: #333; }}
        .file {{ background: white; padding: 10px; margin: 10px 0; border-radius: 5px; }}
        .file a {{ color: #0066cc; text-decoration: none; font-weight: bold; }}
        .file a:hover {{ text-decoration: underline; }}
        .back {{ margin: 10px 0; }}
    </style>
</head>
<body>
    <div class="back"><a href="/memory/vault">← Back to projects</a></div>
    <h1>📖 {}</h1>
    <p>Memories in this project:</p>
    {}
</body>
</html>"#,
                proj,
                proj,
                files
                    .iter()
                    .map(|f| format!(r#"<div class="file"><a href="/memory/vault/{}/{}">{}</a></div>"#, proj, f, f.replace(".md", "")))
                    .collect::<Vec<_>>()
                    .join("\n")
            );
            HttpResponse::Ok()
                .content_type("text/html; charset=utf-8")
                .body(html)
        }
        Err(_) => {
            HttpResponse::NotFound().body("Project not found")
        }
    }
}

/// 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 {
    // Note: We don't check auth here to allow embedding in browsers
    // but in production you might want to add auth
    
    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().body("Invalid filename");
    }

    match std::fs::read_to_string(&file_path) {
        Ok(content) => {
            // Simple markdown to HTML conversion (frontmatter + code highlight)
            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())
            };

            let html = format!(
                r#"<!DOCTYPE html>
<html>
<head>
    <title>{} - {}</title>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width">
    <style>
        body {{ font-family: 'Segoe UI', sans-serif; margin: 20px; max-width: 900px; background: #f5f5f5; }}
        .container {{ background: white; padding: 20px; border-radius: 5px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        h1, h2, h3 {{ color: #333; }}
        code {{ background: #f0f0f0; padding: 2px 6px; border-radius: 3px; }}
        pre {{ background: #2d2d2d; color: #f8f8f2; padding: 15px; border-radius: 5px; overflow-x: auto; }}
        blockquote {{ border-left: 4px solid #0066cc; margin: 10px 0; padding-left: 10px; }}
        .meta {{ color: #666; font-size: 0.9em; margin-bottom: 20px; }}
        .back {{ margin-bottom: 20px; }}
        a {{ color: #0066cc; text-decoration: none; }}
        a:hover {{ text-decoration: underline; }}
    </style>
</head>
<body>
    <div class="container">
        <div class="back"><a href="/memory/vault/{}">← Back to {}</a></div>
        {}
        <div class="meta">Stored in Obsidian vault</div>
        <div class="content">{}</div>
    </div>
</body>
</html>"#,
                file,
                project,
                project,
                project,
                if !frontmatter.is_empty() {
                    format!("<pre>{}</pre>", frontmatter)
                } else {
                    String::new()
                },
                body.clone().replace("&", "&amp;")
                    .replace("<", "&lt;")
                    .replace(">", "&gt;")
                    .lines()
                    .map(|line| {
                        if line.starts_with("# ") {
                            format!("<h1>{}</h1>", &line[2..])
                        } else if line.starts_with("## ") {
                            format!("<h2>{}</h2>", &line[3..])
                        } else if line.starts_with("### ") {
                            format!("<h3>{}</h3>", &line[4..])
                        } else if line.starts_with("- ") {
                            format!("<li>{}</li>", &line[2..])
                        } else if !line.is_empty() {
                            format!("<p>{}</p>", line)
                        } else {
                            "<br/>".to_string()
                        }
                    })
                    .collect::<Vec<_>>()
                    .join("\n")
            );
            HttpResponse::Ok()
                .content_type("text/html; charset=utf-8")
                .body(html)
        }
        Err(_) => {
            HttpResponse::NotFound().body("File not found")
        }
    }
}