// SPDX-License-Identifier: MIT
use crate::budgets::BudgetEndpoint;
use crate::handlers;
use crate::handlers::ensure_auth;
use crate::handlers::mcp::handle_mcp_message_with_caller;
use crate::state::RuntimeState;
use axum::body::Bytes;
use axum::extract::connect_info::ConnectInfo;
use axum::extract::{Request, State};
use axum::http::{HeaderMap, HeaderValue, StatusCode};
use axum::middleware::Next;
use axum::response::Response;
use axum::routing::{get, post};
use axum::Json;
use axum::Router;
use serde_json::Value;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tower::Service;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::cors::CorsLayer;

pub fn build_router(state: RuntimeState, port: u16) -> Router {
    // SEC-001: restrict CORS to localhost origins only.
    let allowed_origins = vec![
        format!("http://127.0.0.1:{port}"),
        format!("http://localhost:{port}"),
        "http://127.0.0.1:1420".to_string(),
        "http://localhost:1420".to_string(),
        "http://127.0.0.1:3000".to_string(),
        "http://localhost:3000".to_string(),
        "http://127.0.0.1:5173".to_string(),
        "http://localhost:5173".to_string(),
        "tauri://localhost".to_string(),
        "https://tauri.localhost".to_string(),
    ]
    .into_iter()
    .filter_map(|origin| parse_allowed_origin(&origin))
    .collect::<Vec<_>>();

    let cors = CorsLayer::new()
        .allow_origin(allowed_origins)
        .allow_methods(tower_http::cors::Any)
        .allow_headers(tower_http::cors::Any);

    Router::new()
        // ── Public endpoints (no auth) ─────────────────────────────
        .route("/health", get(handlers::health::handle_health))
        .route("/readiness", get(handlers::health::handle_readiness))
        // ── Core endpoints ─────────────────────────────────────────
        // boot and recall already accept HeaderMap and call ensure_auth.
        // digest, savings, peek, budget_recall now have auth added to
        // their handler bodies directly.
        .route("/digest", get(handlers::health::handle_digest))
        .route("/savings", get(handlers::health::handle_savings))
        .route("/stats", get(handlers::health::handle_stats))
        .route("/dump", get(handlers::health::handle_dump))
        .route("/store", post(handlers::store::handle_store))
        .route(
            "/recall",
            get(handlers::recall::handle_recall).post(handlers::recall::handle_recall_post),
        )
        .route(
            "/recall/explain",
            get(handlers::recall::handle_recall_explain),
        )
        .route(
            "/recall/semantic",
            get(handlers::recall::handle_semantic_recall),
        )
        .route("/peek", get(handlers::recall::handle_peek))
        .route("/unfold", get(handlers::recall::handle_unfold))
        .route("/boot", get(handlers::boot::handle_boot))
        .route("/boot/audit", get(handlers::boot::handle_boot_audit))
        .route("/diary", post(handlers::diary::handle_diary))
        .route(
            "/recall/budget",
            get(handlers::recall::handle_budget_recall),
        )
        .route("/feedback", post(handlers::feedback::handle_feedback))
        .route(
            "/feedback/stats",
            get(handlers::feedback::handle_feedback_stats),
        )
        .route(
            "/agent-feedback",
            post(handlers::feedback::handle_agent_feedback_record),
        )
        .route(
            "/agent-feedback/stats",
            get(handlers::feedback::handle_agent_feedback_stats),
        )
        .route("/crystals", get(handle_crystals))
        .route("/crystallize", post(handle_crystallize))
        .route("/compact", post(handle_compact))
        .route("/compact/benchmark", post(handle_compact_benchmark))
        .route("/storage", get(handle_storage))
        .route("/forget", post(handlers::mutate::handle_forget))
        .route("/resolve", post(handlers::mutate::handle_resolve))
        .route("/conflicts/resolve", post(handlers::mutate::handle_resolve))
        .route("/conflicts", get(handlers::mutate::handle_conflicts))
        .route(
            "/permissions",
            get(handlers::mutate::handle_permissions_list),
        )
        .route(
            "/permissions/grant",
            post(handlers::mutate::handle_permissions_grant),
        )
        .route(
            "/permissions/revoke",
            post(handlers::mutate::handle_permissions_revoke),
        )
        .route("/archive", post(handlers::mutate::handle_archive))
        .route("/focus/start", post(handle_focus_start))
        .route("/focus/end", post(handle_focus_end))
        .route("/shutdown", post(handlers::mutate::handle_shutdown))
        // ── Conductor (locks, activity, messages, sessions, tasks) ──
        .route("/lock", post(handlers::conductor::handle_lock))
        .route("/unlock", post(handlers::conductor::handle_unlock))
        .route("/locks", get(handlers::conductor::handle_locks))
        .route(
            "/activity",
            post(handlers::conductor::handle_post_activity)
                .get(handlers::conductor::handle_get_activity),
        )
        .route("/message", post(handlers::conductor::handle_post_message))
        .route("/messages", get(handlers::conductor::handle_get_messages))
        .route(
            "/session/start",
            post(handlers::conductor::handle_session_start),
        )
        .route(
            "/session/heartbeat",
            post(handlers::conductor::handle_session_heartbeat),
        )
        .route(
            "/session/end",
            post(handlers::conductor::handle_session_end),
        )
        .route("/sessions", get(handlers::conductor::handle_sessions))
        .route(
            "/tasks",
            post(handlers::conductor::handle_create_task)
                .get(handlers::conductor::handle_get_tasks),
        )
        .route("/tasks/next", get(handlers::conductor::handle_next_task))
        .route("/tasks/claim", post(handlers::conductor::handle_claim_task))
        .route(
            "/tasks/complete",
            post(handlers::conductor::handle_complete_task),
        )
        .route(
            "/tasks/abandon",
            post(handlers::conductor::handle_abandon_task),
        )
        .route(
            "/tasks/delete",
            post(handlers::conductor::handle_delete_task),
        )
        // ── Feed ────────────────────────────────────────────────────
        .route(
            "/feed",
            post(handlers::feed::handle_post_feed).get(handlers::feed::handle_get_feed),
        )
        .route("/feed/ack", post(handlers::feed::handle_feed_ack))
        .route("/feed/{id}", get(handlers::feed::handle_get_feed_by_id))
        // ── Export / Import ────────────────────────────────────────
        .route("/export", get(handlers::export::handle_export))
        .route("/import", post(handlers::export::handle_import))
        // ── Admin (team-mode only, owner/admin role required) ──
        .route("/admin/user/add", post(handlers::admin::handle_user_add))
        .route(
            "/admin/user/rotate-key",
            post(handlers::admin::handle_user_rotate_key),
        )
        .route(
            "/admin/user/remove",
            post(handlers::admin::handle_user_remove),
        )
        .route("/admin/users", get(handlers::admin::handle_user_list))
        .route(
            "/admin/team/create",
            post(handlers::admin::handle_team_create),
        )
        .route(
            "/admin/team/add-member",
            post(handlers::admin::handle_team_add_member),
        )
        .route(
            "/admin/team/remove-member",
            post(handlers::admin::handle_team_remove_member),
        )
        .route("/admin/teams", get(handlers::admin::handle_team_list))
        .route("/admin/unowned", get(handlers::admin::handle_unowned))
        .route(
            "/admin/assign-owner",
            post(handlers::admin::handle_assign_owner),
        )
        .route(
            "/admin/set-visibility",
            post(handlers::admin::handle_set_visibility),
        )
        .route("/admin/archive", post(handlers::admin::handle_archive))
        .route("/admin/stats", get(handlers::admin::handle_stats))
        // ── SSE events ──────────────────────────────────────────────
        .route(
            "/events/stream",
            get(handlers::events::handle_events_stream),
        )
        // ── Brain firing telemetry (owner-scoped, full payload) ─────
        .route(
            "/brain/firing",
            get(handlers::events::handle_brain_firing_stream),
        )
        // ── MCP-RPC proxy endpoint ─────────────────────────────────
        // Accepts raw JSON-RPC messages (same format as MCP stdio).
        // This lets `cortex mcp` run as a thin proxy -- no separate
        // ONNX engine, no separate caches, zero duplication.
        .route("/mcp-rpc", post(handle_mcp_rpc))
        .layer(axum::middleware::from_fn_with_state(
            state.clone(),
            activity_tracking_middleware,
        ))
        // Convert handler panics into HTTP 500 responses instead of letting
        // them tear down the request task and surface as "MCP server exited"
        // / "daemon went down" to clients. The daemon stays up; the bad call
        // gets a structured error.
        .layer(CatchPanicLayer::custom(handle_handler_panic))
        .layer(cors)
        .with_state(state)
}

fn handle_handler_panic(err: Box<dyn std::any::Any + Send + 'static>) -> Response {
    let message = if let Some(s) = err.downcast_ref::<&str>() {
        (*s).to_string()
    } else if let Some(s) = err.downcast_ref::<String>() {
        s.clone()
    } else {
        "<non-string panic payload>".to_string()
    };
    eprintln!("[cortex] HTTP handler panic: {message}");
    handlers::json_response(
        StatusCode::INTERNAL_SERVER_ERROR,
        serde_json::json!({
            "error": "internal server error",
        }),
    )
}

async fn activity_tracking_middleware(
    State(state): State<RuntimeState>,
    mut request: Request,
    next: Next,
) -> Response {
    state.mark_activity_now();
    request
        .headers_mut()
        .remove(handlers::CORTEX_PEER_IP_HEADER);
    let peer_ip = request
        .extensions()
        .get::<ConnectInfo<std::net::SocketAddr>>()
        .map(|ConnectInfo(addr)| addr.ip());
    if let Some(ip) = peer_ip {
        if let Ok(value) = HeaderValue::from_str(&ip.to_string()) {
            request
                .headers_mut()
                .insert(handlers::CORTEX_PEER_IP_HEADER, value);
        }
    }
    next.run(request).await
}

/// HTTP endpoint for MCP proxy -- accepts JSON-RPC, returns JSON-RPC.
/// SEC-001 fix: requires Bearer auth like all other POST mutation endpoints.
async fn handle_mcp_rpc(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
    body: Bytes,
) -> axum::response::Response {
    let caller_id = match handlers::ensure_auth_with_caller(&headers, &state) {
        Ok(caller_id) => caller_id,
        Err(resp) => {
            let (message, hint) = match resp.status() {
                StatusCode::FORBIDDEN => (
                    "Missing X-Cortex-Request header",
                    Some("Include header X-Cortex-Request: true"),
                ),
                StatusCode::UNAUTHORIZED => ("Unauthorized", None),
                _ => ("Auth failed", None),
            };
            let status = resp.status();
            return handlers::json_response(
                status,
                serde_json::json!({
                    "jsonrpc": "2.0",
                    "error": {
                        "code": -32600,
                        "message": message,
                        "hint": hint
                    },
                    "id": serde_json::Value::Null
                }),
            );
        }
    };

    let msg: Value = match serde_json::from_slice(&body) {
        Ok(msg) => msg,
        Err(_) => {
            return handlers::json_response(
                StatusCode::BAD_REQUEST,
                serde_json::json!({
                    "jsonrpc": "2.0",
                    "error": {
                        "code": -32700,
                        "message": "Parse error"
                    },
                    "id": serde_json::Value::Null
                }),
            );
        }
    };
    let source = handlers::resolve_source_identity(&headers, "mcp");
    let ip = handlers::client_ip(&headers);
    if let Some(decision) = state
        .rate_limiter
        .check_budget_for_endpoint(ip, BudgetEndpoint::Mcp)
        .await
    {
        if !decision.allowed {
            handlers::log_budget_rejection(&state, &decision, &source.agent, &ip).await;
            let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
            return handlers::json_response(
                StatusCode::OK,
                serde_json::json!({
                    "jsonrpc": "2.0",
                    "error": {
                        "code": -32029,
                        "message": "budget_exceeded",
                        "data": decision.http_body_json()
                    },
                    "id": id
                }),
            );
        }
    }
    handlers::register_agent_presence_from_headers(&state, &headers, caller_id).await;

    match handle_mcp_message_with_caller(&state, &msg, caller_id, Some(&source)).await {
        Some(resp) => handlers::json_response(StatusCode::OK, resp),
        None => handlers::json_response(StatusCode::OK, serde_json::json!({})),
    }
}

// ─── Compaction handlers ────────────────────────────────────────────────

async fn handle_compact(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let conn = state.db.lock().await;
    let result = crate::compaction::run_compaction(&conn);
    handlers::json_response(
        axum::http::StatusCode::OK,
        serde_json::json!({
            "eventsPruned": result.events_pruned,
            "benchmarkPruned": result.benchmark_pruned,
            "archivedTextStripped": result.archived_text_stripped,
            "expiredPruned": result.expired_pruned,
            "crystalEmbeddingsPruned": result.crystal_embeddings_pruned,
            "clusterMembersPruned": result.cluster_members_pruned,
            "feedbackAggregated": result.feedback_aggregated,
            "staleEmbeddingsPruned": result.stale_embeddings_pruned,
            "coOccurrencePruned": result.co_occurrence_pruned,
            "legacyEmbeddingsMigrated": result.legacy_embeddings_migrated,
            "ftsOptimized": result.fts_optimized,
            "bytesBefore": result.bytes_before,
            "bytesAfter": result.bytes_after,
            "savedKB": (result.bytes_before - result.bytes_after) / 1024,
        }),
    )
}

async fn handle_compact_benchmark(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let conn = state.db.lock().await;
    let result = crate::compaction::purge_benchmark_artifacts(&conn);
    handlers::json_response(
        axum::http::StatusCode::OK,
        serde_json::json!({
            "decisionsDeleted": result.decisions_deleted,
            "embeddingsDeleted": result.embeddings_deleted,
            "clusterMembersDeleted": result.cluster_members_deleted,
            "decisionConflictsDeleted": result.decision_conflicts_deleted,
            "recallFeedbackDeleted": result.recall_feedback_deleted,
            "coOccurrenceDeleted": result.co_occurrence_deleted,
            "eventsDeleted": result.events_deleted,
            "bytesBefore": result.bytes_before,
            "bytesAfter": result.bytes_after,
            "savedKB": (result.bytes_before - result.bytes_after) / 1024,
        }),
    )
}

async fn handle_storage(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let conn = state.db_read.lock().await;
    let breakdown = crate::compaction::storage_breakdown(&conn);
    let total_bytes: i64 = conn
        .query_row("PRAGMA page_count", [], |r| r.get::<_, i64>(0))
        .unwrap_or(0)
        * conn
            .query_row("PRAGMA page_size", [], |r| r.get::<_, i64>(0))
            .unwrap_or(4096);

    let tables: Vec<serde_json::Value> = breakdown
        .iter()
        .map(|(name, count)| serde_json::json!({"table": name, "rows": count}))
        .collect();

    handlers::json_response(
        axum::http::StatusCode::OK,
        serde_json::json!({
            "totalBytes": total_bytes,
            "totalMB": format!("{:.1}", total_bytes as f64 / 1_048_576.0),
            "tables": tables,
        }),
    )
}

// ─── Crystal handlers ───────────────────────────────────────────────────

async fn handle_crystals(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let conn = state.db_read.lock().await;
    let crystals = crate::crystallize::list_crystals(&conn);
    handlers::json_response(
        axum::http::StatusCode::OK,
        serde_json::json!({ "crystals": crystals, "count": crystals.len() }),
    )
}

async fn handle_crystallize(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let conn = state.db.lock().await;
    let brain_sender = Some(state.brain_firing.clone());
    let result = crate::crystallize::run_crystallize_pass_with_brain(
        &conn,
        state.embedding_engine.as_deref(),
        state.default_owner_id,
        &brain_sender,
    );
    handlers::json_response(
        axum::http::StatusCode::OK,
        serde_json::json!({
            "clusters": result.clusters_found,
            "created": result.crystals_created,
            "updated": result.crystals_updated,
            "consolidated": result.entries_consolidated,
        }),
    )
}

// ─── Focus handlers (thin wrappers around focus.rs) ──────────────────────

#[derive(serde::Deserialize)]
struct FocusRequest {
    label: Option<String>,
    agent: Option<String>,
}

async fn handle_focus_start(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
    Json(body): Json<FocusRequest>,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let label = match &body.label {
        Some(l) if !l.is_empty() => l.as_str(),
        _ => {
            return handlers::json_error(
                axum::http::StatusCode::BAD_REQUEST,
                "Missing field: label",
            );
        }
    };
    let agent = body.agent.as_deref().unwrap_or("http");
    let conn = state.db.lock().await;
    match crate::focus::focus_start(&conn, label, agent) {
        Ok(v) => handlers::json_response(axum::http::StatusCode::OK, v),
        Err(e) => handlers::json_error(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &e),
    }
}

async fn handle_focus_end(
    State(state): State<RuntimeState>,
    headers: HeaderMap,
    Json(body): Json<FocusRequest>,
) -> axum::response::Response {
    if let Err(resp) = ensure_auth(&headers, &state) {
        return resp;
    }
    let label = match &body.label {
        Some(l) if !l.is_empty() => l.as_str(),
        _ => {
            return handlers::json_error(
                axum::http::StatusCode::BAD_REQUEST,
                "Missing field: label",
            );
        }
    };
    let agent = body.agent.as_deref().unwrap_or("http");
    let conn = state.db.lock().await;
    match crate::focus::focus_end(&conn, label, agent, state.default_owner_id) {
        Ok(v) => handlers::json_response(axum::http::StatusCode::OK, v),
        Err(e) => handlers::json_error(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &e),
    }
}

pub async fn run(
    router: Router,
    bind_addr: &str,
    port: u16,
    ipc_endpoint: Option<String>,
    db_path: &Path,
    readiness_signal: Option<Arc<AtomicBool>>,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) {
    if let Some(endpoint) = ipc_endpoint {
        spawn_ipc_listener(router.clone(), endpoint);
    }
    let mut activated_listener = match resolve_socket_activation_listener(port) {
        Ok(listener) => listener,
        Err(err) => {
            eprintln!("[cortex] FATAL: socket activation listener error: {err}");
            std::process::exit(1);
        }
    };

    match crate::tls::try_load_tls() {
        Ok(Some(acceptor)) => {
            run_tls(
                router,
                bind_addr,
                port,
                acceptor,
                activated_listener.take(),
                readiness_signal,
                shutdown,
            )
            .await;
        }
        Ok(None) => {
            run_plain(
                router,
                bind_addr,
                port,
                activated_listener.take(),
                readiness_signal,
                shutdown,
            )
            .await;
        }
        Err(e) => {
            // Team mode: refuse to start with broken TLS (auth integrity requires it)
            // Solo mode: allow plain fallback only for localhost binds (or explicit override).
            let team_mode = detect_team_mode_for_tls(db_path);
            let local_bind = is_local_bind_addr(bind_addr);
            let allow_insecure_remote = std::env::var("CORTEX_ALLOW_INSECURE_REMOTE")
                .ok()
                .is_some_and(|value| matches!(value.trim(), "1" | "true" | "TRUE" | "yes"));
            if team_mode {
                eprintln!("[cortex] TLS configuration error: {e}");
                eprintln!(
                    "[cortex] Team mode requires valid TLS -- fix certs at ~/.cortex/tls/ or set CORTEX_TLS_CERT/CORTEX_TLS_KEY"
                );
                std::process::exit(1);
            } else if !local_bind && !allow_insecure_remote {
                eprintln!("[cortex] TLS configuration error: {e}");
                eprintln!(
                    "[cortex] Refusing insecure HTTP fallback for non-local bind '{bind_addr}'."
                );
                eprintln!(
                    "[cortex] Fix TLS certs, bind to localhost, or set CORTEX_ALLOW_INSECURE_REMOTE=1 for explicit temporary override."
                );
                std::process::exit(1);
            } else {
                eprintln!("[cortex] TLS certificate error: {e}");
                if local_bind {
                    eprintln!("[cortex] Starting without TLS (solo mode -- localhost bind)");
                } else {
                    eprintln!(
                        "[cortex] Starting without TLS on non-local bind due to CORTEX_ALLOW_INSECURE_REMOTE=1"
                    );
                }
                run_plain(
                    router,
                    bind_addr,
                    port,
                    activated_listener.take(),
                    readiness_signal,
                    shutdown,
                )
                .await;
            }
        }
    }
}

#[cfg(unix)]
fn resolve_socket_activation_listener(
    expected_port: u16,
) -> Result<Option<tokio::net::TcpListener>, String> {
    use std::os::fd::FromRawFd;

    let listen_fds = std::env::var("LISTEN_FDS")
        .ok()
        .and_then(|raw| raw.trim().parse::<usize>().ok())
        .unwrap_or(0);
    if listen_fds == 0 {
        return Ok(None);
    }

    let Some(listen_pid) = std::env::var("LISTEN_PID")
        .ok()
        .and_then(|raw| raw.trim().parse::<u32>().ok())
    else {
        return Err("LISTEN_FDS is set but LISTEN_PID is missing or invalid".to_string());
    };
    if listen_pid != std::process::id() {
        return Ok(None);
    }
    if listen_fds != 1 {
        return Err(format!(
            "Expected exactly one activated socket (LISTEN_FDS=1), got {listen_fds}"
        ));
    }

    let std_listener = unsafe { std::net::TcpListener::from_raw_fd(3) };
    std_listener
        .set_nonblocking(true)
        .map_err(|e| format!("configure activated socket nonblocking: {e}"))?;
    if let Ok(addr) = std_listener.local_addr() {
        if expected_port > 0 && addr.port() != expected_port {
            eprintln!(
                "[cortex] Warning: activated socket port {} does not match configured port {}",
                addr.port(),
                expected_port
            );
        }
        eprintln!("[cortex] Using socket-activated listener on {addr}");
    } else {
        eprintln!("[cortex] Using socket-activated listener (LISTEN_FDS=1)");
    }
    std::env::remove_var("LISTEN_FDS");
    std::env::remove_var("LISTEN_PID");
    tokio::net::TcpListener::from_std(std_listener)
        .map(Some)
        .map_err(|e| format!("adopt socket-activated listener: {e}"))
}

#[cfg(not(unix))]
fn resolve_socket_activation_listener(
    _expected_port: u16,
) -> Result<Option<tokio::net::TcpListener>, String> {
    Ok(None)
}

fn mark_runtime_ready(readiness_signal: Option<&Arc<AtomicBool>>) {
    if let Some(readiness) = readiness_signal {
        let was_ready = readiness.swap(true, Ordering::SeqCst);
        if !was_ready {
            eprintln!("[cortex] Runtime readiness gate is open");
        }
    }
}

fn spawn_ipc_listener(router: Router, endpoint: String) {
    tokio::spawn(async move {
        if let Err(err) = run_ipc_listener(router, endpoint.clone()).await {
            eprintln!("[cortex] IPC listener disabled for '{endpoint}': {err}");
        }
    });
}

#[cfg(unix)]
async fn run_ipc_listener(router: Router, endpoint: String) -> Result<(), String> {
    use std::os::unix::fs::PermissionsExt;
    use tokio::net::UnixListener;

    let path = std::path::PathBuf::from(&endpoint);
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| format!("create IPC dir: {e}"))?;
    }
    if path.exists() {
        std::fs::remove_file(&path).map_err(|e| format!("remove stale IPC socket: {e}"))?;
    }

    let listener = UnixListener::bind(&path).map_err(|e| format!("bind IPC socket: {e}"))?;
    let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
    eprintln!("[cortex] Listening on unix://{}", path.display());

    loop {
        match listener.accept().await {
            Ok((stream, _)) => {
                let hyper_svc = hyper_util::service::TowerToHyperService::new(router.clone());
                tokio::spawn(async move {
                    let io = hyper_util::rt::TokioIo::new(stream);
                    if let Err(err) = hyper_util::server::conn::auto::Builder::new(
                        hyper_util::rt::TokioExecutor::new(),
                    )
                    .serve_connection(io, hyper_svc)
                    .await
                    {
                        eprintln!("[cortex] IPC unix connection error: {err}");
                    }
                });
            }
            Err(err) => eprintln!("[cortex] IPC unix accept error: {err}"),
        }
    }
}

#[cfg(windows)]
async fn run_ipc_listener(router: Router, endpoint: String) -> Result<(), String> {
    use tokio::net::windows::named_pipe::ServerOptions;

    let mut first_instance = true;
    eprintln!("[cortex] Listening on pipe://{endpoint}");
    loop {
        let mut options = ServerOptions::new();
        if first_instance {
            options.first_pipe_instance(true);
        }
        let server = match options.create(&endpoint) {
            Ok(server) => server,
            Err(err) => {
                if first_instance {
                    return Err(format!("create named pipe: {err}"));
                }
                eprintln!("[cortex] IPC pipe create error: {err}");
                tokio::time::sleep(std::time::Duration::from_millis(150)).await;
                continue;
            }
        };
        first_instance = false;

        if let Err(err) = server.connect().await {
            eprintln!("[cortex] IPC pipe connect error: {err}");
            continue;
        }

        let hyper_svc = hyper_util::service::TowerToHyperService::new(router.clone());
        tokio::spawn(async move {
            let io = hyper_util::rt::TokioIo::new(server);
            if let Err(err) =
                hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
                    .serve_connection(io, hyper_svc)
                    .await
            {
                eprintln!("[cortex] IPC pipe connection error: {err}");
            }
        });
    }
}

fn is_local_bind_addr(bind_addr: &str) -> bool {
    matches!(
        bind_addr.trim().to_ascii_lowercase().as_str(),
        "127.0.0.1" | "localhost" | "::1" | "[::1]"
    )
}

/// Lightweight team-mode detection for TLS decisions (before full state init).
/// Opens the DB briefly to read the config table.
fn detect_team_mode_for_tls(db_path: &Path) -> bool {
    if let Ok(conn) = crate::db::open(db_path) {
        crate::db::is_team_mode(&conn)
    } else {
        false
    }
}

async fn run_plain(
    router: Router,
    bind_addr: &str,
    port: u16,
    activated_listener: Option<tokio::net::TcpListener>,
    readiness_signal: Option<Arc<AtomicBool>>,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) {
    let listener = match activated_listener {
        Some(listener) => listener,
        None => match tokio::net::TcpListener::bind((bind_addr, port)).await {
            Ok(l) => l,
            Err(e) => {
                eprintln!("[cortex] FATAL: Cannot bind to {bind_addr}:{port} -- {e}");
                eprintln!("[cortex] Is another Cortex instance running? Try: cortex paths --json");
                std::process::exit(1);
            }
        },
    };
    mark_runtime_ready(readiness_signal.as_ref());
    if let Ok(addr) = listener.local_addr() {
        eprintln!("[cortex] Listening on http://{addr}");
    } else {
        eprintln!("[cortex] Listening on http://{bind_addr}:{port}");
    }
    if let Err(e) = axum::serve(
        listener,
        router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
    )
    .with_graceful_shutdown(shutdown)
    .await
    {
        eprintln!("[cortex] HTTP server exited with error: {e}");
    }
}

async fn run_tls(
    router: Router,
    bind_addr: &str,
    port: u16,
    acceptor: tokio_rustls::TlsAcceptor,
    activated_listener: Option<tokio::net::TcpListener>,
    readiness_signal: Option<Arc<AtomicBool>>,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) {
    let listener = match activated_listener {
        Some(listener) => listener,
        None => match tokio::net::TcpListener::bind((bind_addr, port)).await {
            Ok(l) => l,
            Err(e) => {
                eprintln!("[cortex] FATAL: Cannot bind to {bind_addr}:{port} -- {e}");
                eprintln!("[cortex] Is another Cortex instance running? Try: cortex paths --json");
                std::process::exit(1);
            }
        },
    };
    mark_runtime_ready(readiness_signal.as_ref());
    if let Ok(addr) = listener.local_addr() {
        eprintln!("[cortex] Listening on https://{addr} (TLS via rustls)");
    } else {
        eprintln!("[cortex] Listening on https://{bind_addr}:{port} (TLS via rustls)");
    }

    let mut make_svc = router.into_make_service_with_connect_info::<std::net::SocketAddr>();

    tokio::pin!(shutdown);

    loop {
        tokio::select! {
            _ = &mut shutdown => {
                eprintln!("[cortex] TLS server shutting down");
                break;
            }
            accept = listener.accept() => {
                match accept {
                    Ok((stream, _addr)) => {
                        let acceptor = acceptor.clone();
                        let tower_svc = match make_svc.call(_addr).await {
                            Ok(tower_svc) => tower_svc,
                            Err(e) => {
                                eprintln!("[cortex] Failed to build TLS service for {_addr}: {e}");
                                continue;
                            }
                        };
                        tokio::spawn(async move {
                            match acceptor.accept(stream).await {
                                Ok(tls_stream) => {
                                    let hyper_svc = hyper_util::service::TowerToHyperService::new(tower_svc);
                                    let io = hyper_util::rt::TokioIo::new(tls_stream);
                                    if let Err(e) = hyper_util::server::conn::auto::Builder::new(
                                        hyper_util::rt::TokioExecutor::new(),
                                    )
                                    .serve_connection(io, hyper_svc)
                                    .await
                                    {
                                        eprintln!("[cortex] TLS connection error for {_addr}: {e}");
                                    }
                                }
                                Err(e) => {
                                    eprintln!("[cortex] TLS handshake failed: {e}");
                                }
                            }
                        });
                    }
                    Err(e) => {
                        eprintln!("[cortex] TCP accept error: {e}");
                    }
                }
            }
        }
    }
}

fn parse_allowed_origin(origin: &str) -> Option<HeaderValue> {
    match origin.parse::<HeaderValue>() {
        Ok(value) => Some(value),
        Err(e) => {
            eprintln!("[cortex] Invalid CORS origin '{origin}': {e}");
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::{to_bytes, Body};
    use axum::http::{HeaderMap, Method, Request, StatusCode};
    use tower::ServiceExt;

    async fn build_state(team_mode: bool) -> RuntimeState {
        build_state_with_budgets(team_mode, None).await
    }

    async fn build_state_with_budgets(team_mode: bool, budgets_toml: Option<&str>) -> RuntimeState {
        let mut home_dir = std::env::temp_dir();
        let mut db_path = std::env::temp_dir();
        let suffix = if team_mode { "team" } else { "solo" };
        home_dir.push(format!(
            "cortex-api-parity-home-{suffix}-{}",
            uuid::Uuid::new_v4()
        ));
        db_path.push(format!(
            "cortex-api-parity-{suffix}-{}.db",
            uuid::Uuid::new_v4()
        ));
        std::fs::create_dir_all(&home_dir).unwrap();
        if let Some(contents) = budgets_toml {
            std::fs::write(home_dir.join("budgets.toml"), contents).unwrap();
        }

        let conn = crate::db::open(&db_path).unwrap();
        crate::db::configure(&conn).unwrap();
        crate::db::initialize_schema(&conn).unwrap();
        crate::db::migrate_focus_table(&conn);
        crate::crystallize::migrate_crystal_tables(&conn);
        if team_mode {
            crate::db::create_team_mode_tables(&conn).unwrap();
            let owner_id =
                crate::db::upsert_owner_user(&conn, "owner", Some("Owner"), "argon2id-placeholder")
                    .unwrap();
            crate::db::migrate_to_team_mode(&conn, owner_id).unwrap();
        }
        drop(conn);

        let home_str = home_dir.to_string_lossy().to_string();
        let db_str = db_path.to_string_lossy().to_string();
        let paths = crate::auth::CortexPaths::resolve_with_overrides(
            Some(&home_str),
            Some(&db_str),
            None,
            None,
        );
        let (state, _shutdown_rx) = crate::state::initialize(&paths, false).unwrap();
        let _ = std::fs::remove_file(db_path);
        state
    }

    async fn route_status(
        router: &Router,
        method: Method,
        path: &str,
        body: Option<&str>,
    ) -> StatusCode {
        let mut req = Request::builder().method(method).uri(path);
        if body.is_some() {
            req = req.header("content-type", "application/json");
        }
        let req = req
            .body(Body::from(body.unwrap_or_default().to_string()))
            .unwrap();
        router.clone().oneshot(req).await.unwrap().status()
    }

    fn budget_toml(endpoint: &str, limit: usize) -> String {
        format!(
            r#"
[defaults]
enabled = true

[endpoints.{endpoint}]
limit = {limit}
window_seconds = 60
"#
        )
    }

    fn authed_json_request(token: &str, method: Method, uri: &str, body: &str) -> Request<Body> {
        Request::builder()
            .method(method)
            .uri(uri)
            .header("content-type", "application/json")
            .header("authorization", format!("Bearer {token}"))
            .header("x-cortex-request", "true")
            .body(Body::from(body.to_string()))
            .unwrap()
    }

    async fn send_json(
        router: &Router,
        token: &str,
        method: Method,
        uri: &str,
        body: &str,
    ) -> (StatusCode, HeaderMap, Value) {
        let resp = router
            .clone()
            .oneshot(authed_json_request(token, method, uri, body))
            .await
            .unwrap();
        let status = resp.status();
        let headers = resp.headers().clone();
        let body = to_bytes(resp.into_body(), 1024 * 1024).await.unwrap();
        let payload = serde_json::from_slice(&body).unwrap();
        (status, headers, payload)
    }

    #[tokio::test]
    async fn test_non_admin_routes_preserved_across_team_migration() {
        let solo_router = build_router(build_state(false).await, 7437);
        let team_router = build_router(build_state(true).await, 7437);

        let cases: Vec<(Method, &str, Option<&str>)> = vec![
            (Method::GET, "/health", None),
            (Method::GET, "/readiness", None),
            (Method::GET, "/digest", None),
            (Method::GET, "/savings", None),
            (Method::GET, "/stats", None),
            (Method::GET, "/dump", None),
            (Method::POST, "/store", Some("{}")),
            (Method::GET, "/recall", None),
            (Method::POST, "/recall", Some("{}")),
            (Method::GET, "/recall/explain", None),
            (Method::GET, "/peek", None),
            (Method::GET, "/unfold", None),
            (Method::GET, "/boot", None),
            (Method::POST, "/diary", Some("{}")),
            (Method::GET, "/recall/budget", None),
            (Method::POST, "/feedback", Some("{}")),
            (Method::GET, "/feedback/stats", None),
            (Method::POST, "/agent-feedback", Some("{}")),
            (Method::GET, "/agent-feedback/stats", None),
            (Method::GET, "/crystals", None),
            (Method::POST, "/crystallize", Some("{}")),
            (Method::POST, "/compact", Some("{}")),
            (Method::POST, "/compact/benchmark", Some("{}")),
            (Method::GET, "/storage", None),
            (Method::POST, "/forget", Some("{}")),
            (Method::POST, "/resolve", Some("{}")),
            (Method::POST, "/conflicts/resolve", Some("{}")),
            (Method::GET, "/conflicts", None),
            (Method::POST, "/archive", Some("{}")),
            (Method::POST, "/focus/start", Some("{}")),
            (Method::POST, "/focus/end", Some("{}")),
            (Method::POST, "/shutdown", Some("{}")),
            (Method::POST, "/lock", Some("{}")),
            (Method::POST, "/unlock", Some("{}")),
            (Method::GET, "/locks", None),
            (Method::POST, "/activity", Some("{}")),
            (Method::GET, "/activity", None),
            (Method::POST, "/message", Some("{}")),
            (Method::GET, "/messages", None),
            (Method::POST, "/session/start", Some("{}")),
            (Method::POST, "/session/heartbeat", Some("{}")),
            (Method::POST, "/session/end", Some("{}")),
            (Method::GET, "/sessions", None),
            (Method::POST, "/tasks", Some("{}")),
            (Method::GET, "/tasks", None),
            (Method::GET, "/tasks/next", None),
            (Method::POST, "/tasks/claim", Some("{}")),
            (Method::POST, "/tasks/complete", Some("{}")),
            (Method::POST, "/tasks/abandon", Some("{}")),
            (Method::POST, "/tasks/delete", Some("{}")),
            (Method::POST, "/feed", Some("{}")),
            (Method::GET, "/feed", None),
            (Method::POST, "/feed/ack", Some("{}")),
            (Method::GET, "/feed/demo", None),
            (Method::GET, "/export", None),
            (Method::POST, "/import", Some("{}")),
            (Method::GET, "/events/stream", None),
            (Method::GET, "/brain/firing", None),
            (Method::POST, "/mcp-rpc", Some("{}")),
        ];

        for (method, path, body) in cases {
            let solo_status = route_status(&solo_router, method.clone(), path, body).await;
            let team_status = route_status(&team_router, method, path, body).await;

            assert_ne!(solo_status, StatusCode::NOT_FOUND, "solo missing {path}");
            assert_ne!(team_status, StatusCode::NOT_FOUND, "team missing {path}");
            assert_eq!(
                solo_status, team_status,
                "status drift for route {path}: solo={solo_status} team={team_status}"
            );
        }
    }

    #[tokio::test]
    async fn admin_permission_routes_are_registered() {
        let solo_router = build_router(build_state(false).await, 7437);
        let team_router = build_router(build_state(true).await, 7437);

        let cases: Vec<(Method, &str, Option<&str>)> = vec![
            (Method::GET, "/permissions", None),
            (Method::POST, "/permissions/grant", Some("{}")),
            (Method::POST, "/permissions/revoke", Some("{}")),
        ];

        for (method, path, body) in cases {
            let solo_status = route_status(&solo_router, method.clone(), path, body).await;
            let team_status = route_status(&team_router, method, path, body).await;

            assert_ne!(solo_status, StatusCode::NOT_FOUND, "solo missing {path}");
            assert_ne!(team_status, StatusCode::NOT_FOUND, "team missing {path}");
        }
    }

    #[tokio::test]
    async fn mcp_rpc_malformed_json_returns_jsonrpc_parse_error() {
        let state = build_state(false).await;
        let token = state.token.as_ref().clone();
        let router = build_router(state, 7437);

        let req = Request::builder()
            .method(Method::POST)
            .uri("/mcp-rpc")
            .header("content-type", "application/json")
            .header("authorization", format!("Bearer {token}"))
            .header("x-cortex-request", "true")
            .body(Body::from("{\"jsonrpc\":"))
            .unwrap();

        let resp = router.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
        let payload: Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(payload["jsonrpc"], "2.0");
        assert_eq!(payload["error"]["code"], -32700);
        assert_eq!(payload["error"]["message"], "Parse error");
        assert_eq!(payload["id"], Value::Null);
    }

    #[tokio::test]
    async fn mcp_rpc_malformed_json_without_auth_returns_unauthorized() {
        let state = build_state(false).await;
        let router = build_router(state, 7437);

        let req = Request::builder()
            .method(Method::POST)
            .uri("/mcp-rpc")
            .header("content-type", "application/json")
            .header("x-cortex-request", "true")
            .body(Body::from("{\"jsonrpc\":"))
            .unwrap();

        let resp = router.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
        let payload: Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(payload["jsonrpc"], "2.0");
        assert_eq!(payload["error"]["message"], "Unauthorized");
        assert_eq!(payload["id"], Value::Null);
    }

    #[tokio::test]
    async fn handler_panic_response_does_not_expose_panic_payload() {
        let response = handle_handler_panic(Box::new("secret internal detail"));

        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
        let body = to_bytes(response.into_body(), 64 * 1024).await.unwrap();
        let payload: Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(payload["error"], "internal server error");
        assert!(!String::from_utf8_lossy(&body).contains("secret internal detail"));
    }

    #[tokio::test]
    async fn store_budget_exhaustion_returns_429_and_skips_decision_write() {
        let state = build_state_with_budgets(false, Some(&budget_toml("store", 1))).await;
        let token = state.token.as_ref().clone();
        let router = build_router(state.clone(), 7437);

        let first = r#"{
            "decision": "Budget governance regression test allows the first store mutation in the configured window.",
            "context": "budget integration test",
            "confidence": 0.95
        }"#;
        let (status, _, _) = send_json(&router, &token, Method::POST, "/store", first).await;
        assert_eq!(status, StatusCode::OK);

        let second = r#"{
            "decision": "Budget governance regression test rejects the second store mutation before writing a row.",
            "context": "budget integration test",
            "confidence": 0.95
        }"#;
        let (status, headers, payload) =
            send_json(&router, &token, Method::POST, "/store", second).await;
        assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
        assert_eq!(payload["error"], "budget_exceeded");
        assert_eq!(payload["endpoint"], "store");
        assert_eq!(payload["limit"], 1);
        assert_eq!(payload["window_seconds"], 60);
        assert_eq!(payload["source"], "budgets.toml");
        assert!(headers.get("Retry-After").is_some());

        let conn = state.db.lock().await;
        let decisions: i64 = conn
            .query_row("SELECT COUNT(*) FROM decisions", [], |row| row.get(0))
            .unwrap();
        assert_eq!(decisions, 1);
    }

    #[tokio::test]
    async fn recall_budget_exhaustion_returns_429_and_health_stays_reachable() {
        let state = build_state_with_budgets(false, Some(&budget_toml("recall", 1))).await;
        let token = state.token.as_ref().clone();
        let router = build_router(state.clone(), 7437);

        let (status, _, _) = send_json(&router, &token, Method::GET, "/recall?q=budget", "").await;
        assert_eq!(status, StatusCode::OK);

        let (status, _, payload) =
            send_json(&router, &token, Method::GET, "/recall?q=budget", "").await;
        assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
        assert_eq!(payload["error"], "budget_exceeded");
        assert_eq!(payload["endpoint"], "recall");

        let resp = router
            .clone()
            .oneshot(
                Request::builder()
                    .method(Method::GET)
                    .uri("/health")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = to_bytes(resp.into_body(), 1024 * 1024).await.unwrap();
        let health: Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(health["budgets"]["enabled"], true);
        assert_eq!(health["budgets"]["recentDenials"], 1);
    }

    #[tokio::test]
    async fn boot_budget_exhaustion_returns_429() {
        let state = build_state_with_budgets(false, Some(&budget_toml("boot", 1))).await;
        let token = state.token.as_ref().clone();
        let router = build_router(state, 7437);

        let (status, _, _) = send_json(&router, &token, Method::GET, "/boot?agent=test", "").await;
        assert_eq!(status, StatusCode::OK);

        let (status, _, payload) =
            send_json(&router, &token, Method::GET, "/boot?agent=test", "").await;
        assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
        assert_eq!(payload["error"], "budget_exceeded");
        assert_eq!(payload["endpoint"], "boot");
    }

    #[tokio::test]
    async fn mcp_budget_exhaustion_returns_jsonrpc_error_data() {
        let state = build_state_with_budgets(false, Some(&budget_toml("mcp", 1))).await;
        let token = state.token.as_ref().clone();
        let router = build_router(state, 7437);
        let body = r#"{
            "jsonrpc": "2.0",
            "id": 7,
            "method": "tools/call",
            "params": { "name": "cortex_health", "arguments": {} }
        }"#;

        let (status, _, first) = send_json(&router, &token, Method::POST, "/mcp-rpc", body).await;
        assert_eq!(status, StatusCode::OK);
        assert!(first.get("result").is_some());

        let (status, _, denied) = send_json(&router, &token, Method::POST, "/mcp-rpc", body).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(denied["jsonrpc"], "2.0");
        assert_eq!(denied["id"], 7);
        assert_eq!(denied["error"]["code"], -32029);
        assert_eq!(denied["error"]["message"], "budget_exceeded");
        assert_eq!(denied["error"]["data"]["endpoint"], "mcp");
        assert_eq!(denied["error"]["data"]["limit"], 1);
        assert_eq!(denied["error"]["data"]["window_seconds"], 60);
        assert!(
            denied["error"]["data"]["retry_after_seconds"]
                .as_u64()
                .unwrap_or_default()
                > 0
        );
    }

    #[test]
    fn local_bind_detection_is_strict() {
        assert!(is_local_bind_addr("127.0.0.1"));
        assert!(is_local_bind_addr("localhost"));
        assert!(is_local_bind_addr("::1"));
        assert!(!is_local_bind_addr("0.0.0.0"));
        assert!(!is_local_bind_addr("100.84.247.96"));
    }
}
