use std::sync::Arc;
use std::time::Duration;

use axum::body::{to_bytes, Body};
use axum::http::{header, HeaderMap, Method, Request, Uri};
use axum::response::Response;
use serde_json::{json, Value};

use crate::db::{get_node_docker_statuses, get_nodes, provider_supports};
use crate::http::{
    add_header, json_response, percent_decode, response_to_text, response_with_body,
};
use crate::json_util::{compact_json, millis_now, truncate_text};
use crate::state::{AppState, HttpTunnelResponse};
use crate::task_dispatcher::{create_and_send_task, wait_for_task_terminal};
use crate::types::{MicroserviceConfig, MicroserviceProxyCacheEntry};

const MICROSERVICE_PROXY_MAX_BODY_TEXT_LENGTH: usize = 8 * 1024 * 1024;
const MICROSERVICE_AVAILABILITY_TTL_MS: i64 = 30_000;
const CODE_QUEUE_OVERVIEW_PATH_FALLBACK_STALE_MS: i64 = 30_000;
const PROVIDER_HTTP_TUNNEL_MAX_ATTEMPTS: i32 = 3;
const FORWARD_REQUEST_HEADERS: &[&str] = &[
    "accept",
    "content-type",
    "cookie",
    "range",
    "x-auth",
    "x-requested-with",
    "destination",
    "overwrite",
    "tus-resumable",
    "upload-concat",
    "upload-defer-length",
    "upload-length",
    "upload-metadata",
    "upload-offset",
];
const FORWARD_RESPONSE_HEADERS: &[&str] = &[
    "cache-control",
    "content-disposition",
    "etag",
    "expires",
    "last-modified",
    "location",
    "set-cookie",
    "www-authenticate",
];

fn service_by_id(state: &Arc<AppState>, service_id: &str) -> Option<MicroserviceConfig> {
    state
        .config
        .microservices
        .iter()
        .find(|service| service.id == service_id)
        .cloned()
}

fn is_microservice_path_allowed(service: &MicroserviceConfig, path: &str) -> bool {
    service
        .backend
        .allowed_path_prefixes
        .iter()
        .any(|prefix| path == prefix || path.starts_with(prefix))
}

fn is_microservice_method_allowed(service: &MicroserviceConfig, method: &Method) -> bool {
    service
        .backend
        .allowed_methods
        .iter()
        .any(|allowed| allowed == method.as_str())
}

fn docker_status_record(status: Option<&Value>) -> serde_json::Map<String, Value> {
    status
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default()
}

fn find_container(status: Option<&Value>, container_name: &str) -> Value {
    status
        .and_then(|status| status.get("containers"))
        .and_then(Value::as_array)
        .and_then(|containers| {
            containers
                .iter()
                .find(|item| item.get("name").and_then(Value::as_str) == Some(container_name))
                .cloned()
        })
        .unwrap_or(Value::Null)
}

fn is_k3sctl_managed_microservice(service: &MicroserviceConfig) -> bool {
    service.deployment.mode == "k3sctl-managed"
        || service.backend.proxy_mode == "k3sctl-adapter-http"
}

fn can_direct_proxy_microservice(service: &MicroserviceConfig) -> bool {
    service.provider_id == "main-server"
}

fn code_queue_mgr_internal_url() -> String {
    std::env::var("CODE_QUEUE_MGR_INTERNAL_URL")
        .unwrap_or_else(|_| "http://code-queue-mgr:4278".to_string())
}

fn direct_code_queue_mgr_service(
    service: &MicroserviceConfig,
    timeout_ms: Option<u64>,
) -> MicroserviceConfig {
    let mut copy = service.clone();
    copy.provider_id = "main-server".to_string();
    copy.backend.node_base_url = code_queue_mgr_internal_url();
    copy.backend.proxy_mode = "unidesk-direct".to_string();
    copy.backend.timeout_ms = timeout_ms.unwrap_or_else(|| service.backend.timeout_ms.min(12_000));
    copy.deployment.mode = "unidesk-direct".to_string();
    copy
}

fn code_queue_k3s_service_id_for_request(method: &Method, target_path: &str) -> &'static str {
    let method = method.as_str();
    if matches!(target_path, "/" | "/health" | "/live" | "/api/dev-ready" | "/api/runtime-preflight") {
        return "code-queue-scheduler";
    }
    if matches!(
        target_path,
        "/api/oa/backfill"
            | "/api/notifications/claudeqq/drain"
            | "/api/notifications/claudeqq/backfill"
    ) {
        return "code-queue-scheduler";
    }
    if matches!(
        target_path,
        "/api/judge/probe"
            | "/api/judge/self-test"
            | "/api/queue-order/self-test"
            | "/api/reference-injection/self-test"
            | "/api/trace-port/self-test"
    ) {
        return "code-queue-scheduler";
    }
    if path_matches_task_action(target_path, "steer")
        || path_matches_task_action(target_path, "interrupt")
    {
        return "code-queue-scheduler";
    }
    if method == "DELETE" && path_matches_task_detail(target_path) {
        return "code-queue-scheduler";
    }
    if target_path == "/api/dev-containers" || target_path.starts_with("/api/dev-containers/") {
        return "code-queue-scheduler";
    }
    if method == "GET" || method == "HEAD" {
        "code-queue-read"
    } else {
        "code-queue-write"
    }
}

fn json_bool(value: &Value, path: &[&str]) -> Option<bool> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_bool()
}

fn json_i64(value: &Value, path: &[&str]) -> Option<i64> {
    let mut current = value;
    for key in path {
        current = current.get(*key)?;
    }
    current.as_i64()
}

fn code_queue_infra_service_names(service: &MicroserviceConfig) -> (String, String) {
    let provider_slug = service.provider_id.to_ascii_lowercase();
    (
        format!("{provider_slug}-provider-egress-proxy"),
        format!("{provider_slug}-tcp-egress-gateway"),
    )
}

fn code_queue_route_postgres_check(scheduler_body: &Value, tcp_egress_service: &str) -> Value {
    let storage = scheduler_body
        .get("queue")
        .and_then(|queue| queue.get("storage"))
        .unwrap_or(&Value::Null);
    let postgres_ready = storage
        .get("postgresReady")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let last_error = storage.get("lastError").cloned().unwrap_or(Value::Null);
    json!({
        "ok": postgres_ready && last_error.is_null(),
        "route": format!("{tcp_egress_service}.unidesk.svc.cluster.local:15432"),
        "postgresReady": postgres_ready,
        "lastError": last_error,
        "source": "scheduler.health.queue.storage"
    })
}

fn code_queue_reconcile_check(scheduler_body: &Value) -> Value {
    let queue = scheduler_body.get("queue").unwrap_or(&Value::Null);
    let reconcile = queue.get("reconcile").unwrap_or(&Value::Null);
    let orphaned_active_count = queue
        .get("orphanedActiveTaskCount")
        .and_then(Value::as_i64)
        .unwrap_or(0);
    let active_run_slot_waiters = queue
        .get("activeRunSlotWaiters")
        .and_then(Value::as_array)
        .map(|items| items.len() as i64)
        .unwrap_or(0);
    let recoverable_orphaned_active_count = reconcile
        .get("recoverableOrphanedActiveTaskCount")
        .and_then(Value::as_i64)
        .unwrap_or(orphaned_active_count);
    let retry_wait_count = reconcile
        .get("retryWaitTaskCount")
        .and_then(Value::as_i64)
        .or_else(|| {
            queue
                .get("counts")
                .and_then(|counts| counts.get("retry_wait"))
                .and_then(Value::as_i64)
        })
        .unwrap_or(0);
    let active_without_local_run = recoverable_orphaned_active_count > 0;
    json!({
        "ok": !active_without_local_run,
        "status": if active_without_local_run { "recovery-required" } else { "idle-or-reconciled" },
        "orphanedActiveTaskCount": orphaned_active_count,
        "recoverableOrphanedActiveTaskCount": recoverable_orphaned_active_count,
        "retryWaitTaskCount": retry_wait_count,
        "activeRunSlotWaiterCount": active_run_slot_waiters,
        "lastRunAt": reconcile.get("lastRunAt").cloned().unwrap_or(Value::Null),
        "lastReason": reconcile.get("lastReason").cloned().unwrap_or(Value::Null),
        "lastRecovered": reconcile.get("lastRecovered").cloned().unwrap_or(Value::Null),
        "source": "scheduler.health.queue.reconcile"
    })
}

fn code_queue_dependency_check(service_id: &str, diagnostics: &Value) -> Value {
    let checks = diagnostics.get("checks").unwrap_or(&Value::Null);
    let deployment_available = json_bool(checks, &["deployment", "available"]).unwrap_or(false);
    let endpoint_non_empty = json_i64(checks, &["endpoint", "readyAddressCount"])
        .map(|count| count > 0)
        .unwrap_or_else(|| json_bool(checks, &["endpoint", "ok"]).unwrap_or(false));
    let target_ok = json_bool(checks, &["targetService", "ok"])
        .or_else(|| json_bool(checks, &["kubernetesApiServiceProxy", "ok"]))
        .or_else(|| json_bool(checks, &["managedService", "ok"]))
        .unwrap_or(false);
    let ok = diagnostics
        .get("ok")
        .and_then(Value::as_bool)
        .unwrap_or(false)
        && deployment_available
        && endpoint_non_empty
        && target_ok;
    json!({
        "ok": ok,
        "serviceId": service_id,
        "deploymentAvailable": deployment_available,
        "endpointNonEmpty": endpoint_non_empty,
        "targetServiceOk": target_ok,
        "deployment": compact_json(checks.get("deployment").unwrap_or(&json!({ "ok": false, "skipped": true }))),
        "endpoint": compact_json(checks.get("endpoint").unwrap_or(&json!({ "ok": false, "skipped": true }))),
        "targetService": compact_json(checks.get("targetService").unwrap_or(&json!({ "ok": false, "skipped": true }))),
        "managedService": compact_json(checks.get("managedService").unwrap_or(&json!({ "ok": false, "skipped": true }))),
        "kubernetesApiServiceProxy": compact_json(checks.get("kubernetesApiServiceProxy").unwrap_or(&json!({ "ok": false, "skipped": true })))
    })
}

async fn code_queue_adapter_diagnostics(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    k3s_service_id: &str,
) -> Value {
    let adapter_service_id = service
        .deployment
        .adapter_service_id
        .as_deref()
        .unwrap_or("k3sctl-adapter");
    let Some(adapter) = service_by_id(state, adapter_service_id) else {
        return json!({
            "ok": false,
            "serviceId": k3s_service_id,
            "error": format!("k3sctl adapter microservice not found: {adapter_service_id}")
        });
    };
    let adapter_path = format!(
        "/api/services/{}/diagnostics",
        urlencoding_like(k3s_service_id)
    );
    let response = fetch_microservice_upstream_response(
        state,
        &adapter,
        &Method::GET,
        &adapter_path,
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let status = response.status().as_u16();
    let (status2, content_type, body_text) = response_to_text(response).await.unwrap_or((
        status,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let body: Value = serde_json::from_str(&body_text)
        .unwrap_or_else(|_| json!({ "text": truncate_text(&body_text, 4000) }));
    if (200..300).contains(&status2) && body.get("ok").and_then(Value::as_bool) == Some(true) {
        body
    } else {
        json!({
            "ok": false,
            "serviceId": k3s_service_id,
            "status": status2,
            "contentType": content_type,
            "body": body
        })
    }
}

fn code_queue_scheduler_only_path(method: &Method, target_path: &str) -> bool {
    code_queue_k3s_service_id_for_request(method, target_path) == "code-queue-scheduler"
        && !(target_path == "/" || target_path == "/health")
}

fn code_queue_fallback_workdirs_response(
    service: &MicroserviceConfig,
    status: u16,
    content_type: &str,
    body_text: &str,
) -> Option<Response> {
    if status != 404 {
        return None;
    }
    let body = serde_json::from_str::<Value>(body_text).ok()?;
    if body.get("path").and_then(Value::as_str) != Some("/api/workdirs") {
        return None;
    }
    let provider_id = service.provider_id.clone();
    let fallback = json!({
        "ok": true,
        "workdirs": [{
            "providerId": provider_id,
            "executionMode": "default",
            "path": "/workspace",
            "createdAt": null,
            "updatedAt": null,
            "source": "backend-core-compat"
        }],
        "defaultProviderId": provider_id,
        "defaultWorkdir": "/workspace",
        "remoteDefaultWorkdir": "/home/ubuntu",
        "source": "backend-core-compat-code-queue-workdirs",
        "compat": {
            "reason": "code-queue-mgr version does not yet expose /api/workdirs",
            "upstreamStatus": status,
            "upstreamContentType": content_type
        }
    });
    Some(response_with_body(
        200,
        "application/json; charset=utf-8",
        Body::from(
            serde_json::to_string(&fallback)
                .unwrap_or_else(|_| "{\"ok\":true,\"workdirs\":[]}".to_string()),
        ),
    ))
}

fn path_matches_task_detail(path: &str) -> bool {
    path.strip_prefix("/api/tasks/")
        .is_some_and(|rest| !rest.is_empty() && !rest.contains('/'))
}

fn path_matches_task_action(path: &str, action: &str) -> bool {
    path.strip_prefix("/api/tasks/")
        .and_then(|rest| rest.strip_suffix(&format!("/{action}")))
        .is_some_and(|task_id| !task_id.is_empty() && !task_id.contains('/'))
}

fn code_queue_task_id_for_action(target_path: &str, action: &str) -> Option<String> {
    target_path
        .strip_prefix("/api/tasks/")
        .and_then(|rest| rest.strip_suffix(&format!("/{action}")))
        .map(percent_decode)
}

fn microservice_cache_ttl_ms(service_id: &str, target_path: &str) -> i64 {
    if target_path == "/health" || target_path.ends_with("/stream") {
        0
    } else if service_id == "pipeline" && target_path == "/api/snapshot" {
        6_000
    } else if service_id == "pipeline" && target_path.starts_with("/api/oa-event-flow/") {
        20_000
    } else if service_id == "pipeline" && target_path.starts_with("/api/model-quota/") {
        60_000
    } else if service_id == "pipeline" && target_path.starts_with("/api/runs/") {
        6_000
    } else if service_id == "findjob"
        && matches!(target_path, "/api/summary" | "/api/jobs" | "/api/drafts")
    {
        8_000
    } else if service_id == "met-nonlinear"
        && matches!(target_path, "/api/images" | "/api/projects")
    {
        15_000
    } else if service_id == "met-nonlinear"
        && matches!(target_path, "/api/queue" | "/api/summary" | "/api/history")
    {
        5_000
    } else if service_id == "code-queue" && target_path.contains("/transcript") {
        1_000
    } else if service_id == "code-queue" && target_path == "/api/tasks/overview" {
        3_000
    } else if service_id == "code-queue" && target_path.starts_with("/api/tasks/") {
        2_000
    } else {
        750
    }
}

fn microservice_cache_stale_ms(service_id: &str, target_path: &str) -> i64 {
    if service_id == "pipeline" && target_path.starts_with("/api/model-quota/") {
        5 * 60_000
    } else if service_id == "pipeline" || service_id == "met-nonlinear" {
        45_000
    } else if service_id == "findjob" {
        60_000
    } else if service_id == "code-queue" {
        15_000
    } else {
        5_000
    }
}

fn provider_microservice_cache_ttl_ms(service_id: &str, target_path: &str) -> i64 {
    if target_path == "/health" {
        0
    } else if service_id == "met-nonlinear"
        && matches!(target_path, "/api/images" | "/api/projects")
    {
        60_000
    } else if service_id == "met-nonlinear" && target_path == "/api/history" {
        10_000
    } else if service_id == "met-nonlinear" && matches!(target_path, "/api/queue" | "/api/summary")
    {
        3_000
    } else if service_id == "pipeline"
        && (target_path == "/api/snapshot" || target_path.starts_with("/api/oa-event-flow/"))
    {
        2_000
    } else if service_id == "findjob" && matches!(target_path, "/api/summary" | "/api/jobs") {
        2_000
    } else {
        1_000
    }
}

#[derive(Clone)]
struct ProxyOptions {
    query: String,
    json_array_limits: Value,
}

fn read_microservice_array_limits(uri: &Uri) -> ProxyOptions {
    let query = uri.query().unwrap_or("");
    let mut params = Vec::<(String, String)>::new();
    let mut limits = serde_json::Map::new();
    for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
        if key == "__unideskArrayLimit" {
            for entry in value.split(',') {
                let mut parts = entry.splitn(2, ':');
                let path = parts.next().unwrap_or("");
                let limit = parts
                    .next()
                    .and_then(|text| text.parse::<i64>().ok())
                    .unwrap_or(0);
                if path
                    .chars()
                    .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | '-'))
                    && limit > 0
                    && limit <= 500
                {
                    limits.insert(path.to_string(), json!(limit));
                }
            }
        } else {
            params.push((key.into_owned(), value.into_owned()));
        }
    }
    let query = if params.is_empty() {
        String::new()
    } else {
        format!(
            "?{}",
            url::form_urlencoded::Serializer::new(String::new())
                .extend_pairs(params)
                .finish()
        )
    };
    ProxyOptions {
        query,
        json_array_limits: Value::Object(limits),
    }
}

fn read_microservice_request_headers(headers: &HeaderMap) -> Value {
    let mut result = serde_json::Map::new();
    for name in FORWARD_REQUEST_HEADERS {
        if let Some(value) = headers
            .get(*name)
            .and_then(|value| value.to_str().ok())
            .filter(|value| !value.is_empty())
        {
            result.insert(
                (*name).to_string(),
                Value::String(truncate_text(value, 4096)),
            );
        }
    }
    Value::Object(result)
}

fn headers_from_microservice_request(request_headers: &Value) -> reqwest::header::HeaderMap {
    let mut headers = reqwest::header::HeaderMap::new();
    if let Some(object) = request_headers.as_object() {
        for name in FORWARD_REQUEST_HEADERS {
            if let Some(value) = object.get(*name).and_then(Value::as_str) {
                if let (Ok(name), Ok(value)) = (
                    reqwest::header::HeaderName::from_bytes(name.as_bytes()),
                    reqwest::header::HeaderValue::from_str(value),
                ) {
                    headers.insert(name, value);
                }
            }
        }
    }
    headers
}

fn forwarded_response_headers_from_reqwest(
    headers: &reqwest::header::HeaderMap,
) -> serde_json::Map<String, Value> {
    let mut result = serde_json::Map::new();
    for name in FORWARD_RESPONSE_HEADERS {
        if let Some(value) = headers
            .get(*name)
            .and_then(|value| value.to_str().ok())
            .filter(|value| !value.is_empty())
        {
            result.insert((*name).to_string(), Value::String(truncate_text(value, 8192)));
        }
    }
    result
}

fn copy_forwarded_response_headers_from_json(response: &mut Response, result: &Value) {
    let Some(headers) = result.get("responseHeaders").and_then(Value::as_object) else {
        return;
    };
    for name in FORWARD_RESPONSE_HEADERS {
        if let Some(value) = headers.get(*name).and_then(Value::as_str) {
            add_header(response, name, value);
        }
    }
}

fn copy_forwarded_response_headers_from_reqwest(
    response: &mut Response,
    headers: &reqwest::header::HeaderMap,
) {
    for name in FORWARD_RESPONSE_HEADERS {
        if let Some(value) = headers.get(*name).and_then(|value| value.to_str().ok()) {
            add_header(response, name, value);
        }
    }
}

fn content_type_is_json(content_type: &str) -> bool {
    content_type.to_ascii_lowercase().contains("json")
}

const TODO_NOTE_ACTION_TYPES: &[&str] = &[
    "addTodo",
    "updateTodoTitle",
    "toggleTodoCompleted",
    "toggleTodoExpanded",
    "setAllTodosExpanded",
    "moveTodo",
    "deleteTodo",
    "renameInstance",
    "setTodoReminder",
];

const TODO_NOTE_WRITABLE_ENDPOINTS: &[(&str, &str, &str)] = &[
    (
        "POST",
        "/api/instances",
        "Create a new todo list; body {name}.",
    ),
    (
        "DELETE",
        "/api/instances/:instanceId",
        "Delete a todo list.",
    ),
    (
        "POST",
        "/api/instances/:instanceId/actions",
        "Apply a typed action; body {action: {type, ...}}. Use the action queue pattern, not REST collection paths like /api/instances/:id/todos.",
    ),
    (
        "POST",
        "/api/instances/:instanceId/undo",
        "Undo the last applied action.",
    ),
    (
        "POST",
        "/api/instances/:instanceId/redo",
        "Redo the last undone action.",
    ),
];

fn is_todo_note_misleading_route_not_found(
    status: u16,
    content_type: &str,
    body_text: &str,
) -> Option<Value> {
    if status != 404 || !content_type_is_json(content_type) {
        return None;
    }
    let body = serde_json::from_str::<Value>(body_text).ok()?;
    let record = body.as_object()?;
    if record.get("ok").and_then(Value::as_bool) == Some(false)
        && record.get("error").and_then(Value::as_str)
            == Some("Todo Note is running in backend-only mode")
    {
        Some(body)
    } else {
        None
    }
}

fn todo_note_route_not_found_body(
    method: &Method,
    target_path: &str,
    upstream_body: Value,
) -> Value {
    json!({
        "ok": false,
        "error": "Todo Note route not found",
        "backendOnly": true,
        "method": method.as_str(),
        "path": target_path,
        "bodyRewritten": true,
        "rewriteReason": "Todo Note catch-all 404 body was misleading; UniDesk proxy wrapped it with a structured route diagnostic. See docs/reference/cli.md and pikasTech/unidesk issue #198 for context.",
        "writableApiEndpoints": TODO_NOTE_WRITABLE_ENDPOINTS.iter().map(|(method, path, hint)| json!({ "method": method, "path": path, "hint": hint })).collect::<Vec<Value>>(),
        "actionTypes": TODO_NOTE_ACTION_TYPES,
        "hint": "Write operations use the action queue pattern: POST /api/instances/:instanceId/actions with body {action: {type, ...}}. The requested path/method is not a registered Todo Note API route; TODO_NOTE_BACKEND_ONLY only disables the upstream Vite SPA and does not lock registered API writes.",
        "issueReference": "pikasTech/unidesk#198",
        "upstreamBody": upstream_body,
    })
}

async fn rewrite_todo_note_misleading_route_not_found_response(
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    response: Response,
) -> Response {
    if service.id != "todo-note" {
        return response;
    }
    let status = response.status().as_u16();
    let headers = response.headers().clone();
    let content_type = headers
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("application/octet-stream")
        .to_string();
    if status != 404 || !content_type_is_json(&content_type) {
        return response;
    }
    let bytes = to_bytes(response.into_body(), 16 * 1024 * 1024)
        .await
        .unwrap_or_default();
    let body_text = String::from_utf8_lossy(&bytes).to_string();
    let Some(upstream_body) =
        is_todo_note_misleading_route_not_found(status, &content_type, &body_text)
    else {
        let mut rebuilt = response_with_body(status, &content_type, body_text);
        for name in FORWARD_RESPONSE_HEADERS {
            if let Some(value) = headers.get(*name).and_then(|value| value.to_str().ok()) {
                add_header(&mut rebuilt, name, value);
            }
        }
        if let Some(value) = headers
            .get("x-unidesk-proxy-mode")
            .and_then(|value| value.to_str().ok())
        {
            add_header(&mut rebuilt, "x-unidesk-proxy-mode", value);
        }
        if let Some(value) = headers
            .get("x-unidesk-upstream-proxy-mode")
            .and_then(|value| value.to_str().ok())
        {
            add_header(&mut rebuilt, "x-unidesk-upstream-proxy-mode", value);
        }
        return rebuilt;
    };
    let mut rewritten = json_response(
        todo_note_route_not_found_body(method, target_path, upstream_body),
        404,
    );
    for name in FORWARD_RESPONSE_HEADERS {
        if let Some(value) = headers.get(*name).and_then(|value| value.to_str().ok()) {
            add_header(&mut rewritten, name, value);
        }
    }
    if let Some(value) = headers
        .get("x-unidesk-proxy-mode")
        .and_then(|value| value.to_str().ok())
    {
        add_header(&mut rewritten, "x-unidesk-proxy-mode", value);
    }
    if let Some(value) = headers
        .get("x-unidesk-upstream-proxy-mode")
        .and_then(|value| value.to_str().ok())
    {
        add_header(&mut rewritten, "x-unidesk-upstream-proxy-mode", value);
    }
    add_header(
        &mut rewritten,
        "x-unidesk-todo-note-route-diagnostic",
        "true",
    );
    rewritten
}

fn apply_json_array_limits(body_text: String, content_type: &str, limits: &Value) -> String {
    let Some(limits) = limits.as_object() else {
        return body_text;
    };
    if limits.is_empty() || !content_type_is_json(content_type) {
        return body_text;
    }
    let Ok(mut parsed) = serde_json::from_str::<Value>(&body_text) else {
        return body_text;
    };
    let Some(root) = parsed.as_object_mut() else {
        return body_text;
    };
    let mut applied = serde_json::Map::new();
    for (path, limit) in limits {
        let limit = limit.as_u64().unwrap_or(0) as usize;
        if limit == 0 || limit > 500 {
            continue;
        }
        if let Some(array) = array_at_path_mut(root, path) {
            let original_length = array.len();
            if array.len() > limit {
                array.truncate(limit);
            }
            applied.insert(path.clone(), json!({ "limit": limit, "originalLength": original_length, "returnedLength": array.len() }));
        }
    }
    root.insert("_unidesk".to_string(), json!({ "arrayLimits": applied }));
    serde_json::to_string(&parsed).unwrap_or(body_text)
}

fn array_at_path_mut<'a>(
    root: &'a mut serde_json::Map<String, Value>,
    path: &str,
) -> Option<&'a mut Vec<Value>> {
    let mut current = root;
    let mut parts = path.split('.').peekable();
    while let Some(part) = parts.next() {
        if parts.peek().is_none() {
            return current.get_mut(part).and_then(Value::as_array_mut);
        }
        current = current.get_mut(part)?.as_object_mut()?;
    }
    None
}

fn bounded_microservice_body_text(
    body_text: String,
    content_type: &str,
    metadata: Value,
) -> (String, bool) {
    if body_text.len() <= MICROSERVICE_PROXY_MAX_BODY_TEXT_LENGTH {
        return (body_text, false);
    }
    if content_type_is_json(content_type) {
        return (json!({
            "ok": false,
            "error": "microservice proxy response body is too large",
            "serviceId": metadata.get("serviceId").cloned().unwrap_or(Value::Null),
            "targetPath": metadata.get("targetPath").cloned().unwrap_or(Value::Null),
            "upstreamStatus": metadata.get("status").cloned().unwrap_or(Value::Null),
            "upstreamBodyBytes": metadata.get("upstreamBodyBytes").cloned().unwrap_or(Value::Null),
            "transformedBodyBytes": body_text.len(),
            "responseBodyLimitBytes": MICROSERVICE_PROXY_MAX_BODY_TEXT_LENGTH,
            "hint": "Use a paged endpoint or tighten __unideskArrayLimit so the response stays below the proxy safety limit."
        }).to_string(), true);
    }
    (
        truncate_text(&body_text, MICROSERVICE_PROXY_MAX_BODY_TEXT_LENGTH),
        true,
    )
}

fn response_from_provider_microservice_result(result: Value, proxy_mode: &str) -> Response {
    let status = result.get("status").and_then(Value::as_u64).unwrap_or(0);
    let content_type = result
        .get("contentType")
        .and_then(Value::as_str)
        .unwrap_or("application/json; charset=utf-8");
    let body_text = result
        .get("bodyText")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string();
    if !(100..=599).contains(&status) {
        return json_response(
            json!({ "ok": false, "error": "microservice proxy returned invalid upstream status", "result": result }),
            502,
        );
    }
    let mut response = response_with_body(status as u16, content_type, body_text);
    add_header(&mut response, "x-unidesk-proxy-mode", proxy_mode);
    add_header(
        &mut response,
        "x-unidesk-upstream-proxy-mode",
        result
            .get("proxyMode")
            .and_then(Value::as_str)
            .unwrap_or(""),
    );
    add_header(
        &mut response,
        "x-unidesk-response-truncated",
        if result.get("truncated").and_then(Value::as_bool) == Some(true) {
            "true"
        } else {
            "false"
        },
    );
    copy_forwarded_response_headers_from_json(&mut response, &result);
    response
}

async fn direct_microservice_response(
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
    request_headers: &Value,
    body_text: String,
) -> Response {
    let base = match reqwest::Url::parse(&service.backend.node_base_url) {
        Ok(url) => url,
        Err(error) => {
            return json_response(
                json!({ "ok": false, "error": "invalid microservice base URL", "serviceId": service.id, "detail": error.to_string() }),
                502,
            )
        }
    };
    let mut upstream = match base.join(target_path.trim_start_matches('/')) {
        Ok(url) => url,
        Err(error) => {
            return json_response(
                json!({ "ok": false, "error": "invalid microservice target URL", "serviceId": service.id, "detail": error.to_string() }),
                502,
            )
        }
    };
    upstream.set_query(proxy_options.query.strip_prefix('?'));
    let client = reqwest::Client::new();
    let mut request = client
        .request(
            method.as_str().parse().unwrap_or(reqwest::Method::GET),
            upstream,
        )
        .headers(headers_from_microservice_request(request_headers));
    if method != Method::GET && method != Method::HEAD {
        request = request.body(body_text);
    }
    let result = tokio::time::timeout(
        Duration::from_millis(service.backend.timeout_ms.max(1000)),
        request.send(),
    )
    .await;
    let response = match result {
        Ok(Ok(response)) => response,
        Ok(Err(error)) => {
            return json_response(
                json!({ "ok": false, "error": "direct microservice proxy failed", "serviceId": service.id, "detail": error.to_string() }),
                502,
            )
        }
        Err(_) => {
            return json_response(
                json!({ "ok": false, "error": "direct microservice proxy timed out", "serviceId": service.id, "timeoutMs": service.backend.timeout_ms }),
                504,
            )
        }
    };
    let status = response.status().as_u16();
    let response_headers = response.headers().clone();
    let content_type = response
        .headers()
        .get(reqwest::header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("text/plain; charset=utf-8")
        .to_string();
    let raw_body = match response.text().await {
        Ok(text) => text,
        Err(error) => {
            return json_response(
                json!({ "ok": false, "error": "direct microservice body read failed", "serviceId": service.id, "detail": error.to_string() }),
                502,
            )
        }
    };
    let limited = apply_json_array_limits(
        raw_body.clone(),
        &content_type,
        &proxy_options.json_array_limits,
    );
    let (body_text, truncated) = bounded_microservice_body_text(
        limited,
        &content_type,
        json!({ "serviceId": service.id, "targetPath": target_path, "status": status, "upstreamBodyBytes": raw_body.len() }),
    );
    let mut response = response_with_body(status, &content_type, body_text);
    add_header(&mut response, "x-unidesk-proxy-mode", "direct");
    add_header(
        &mut response,
        "x-unidesk-response-truncated",
        if truncated { "true" } else { "false" },
    );
    copy_forwarded_response_headers_from_reqwest(&mut response, &response_headers);
    response
}

async fn k3sctl_adapter_microservice_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
    request_headers: &Value,
    body_text: String,
) -> Response {
    let adapter_service_id = service
        .deployment
        .adapter_service_id
        .as_deref()
        .unwrap_or("k3sctl-adapter");
    let Some(adapter) = service_by_id(state, adapter_service_id) else {
        return json_response(
            json!({ "ok": false, "error": format!("k3sctl adapter microservice not found: {adapter_service_id}"), "serviceId": service.id }),
            502,
        );
    };
    if adapter.id == service.id || is_k3sctl_managed_microservice(&adapter) {
        return json_response(
            json!({ "ok": false, "error": "k3sctl adapter must be a UniDesk-direct microservice", "serviceId": service.id, "adapterServiceId": adapter_service_id }),
            500,
        );
    }
    let k3s_service_id = if service.id == "code-queue" {
        code_queue_k3s_service_id_for_request(method, target_path).to_string()
    } else {
        service
            .deployment
            .k3s_service_id
            .clone()
            .unwrap_or_else(|| service.id.clone())
    };
    let adapter_target_path = format!(
        "/api/services/{}/proxy{}",
        urlencoding_like(&k3s_service_id),
        target_path
    );
    Box::pin(fetch_microservice_upstream_response(
        state,
        &adapter,
        method,
        &adapter_target_path,
        proxy_options,
        request_headers,
        body_text,
    ))
    .await
}

async fn code_queue_k3s_scheduler_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
    request_headers: &Value,
    body_text: String,
) -> Response {
    let mut scheduler_service = service.clone();
    scheduler_service.deployment.k3s_service_id = Some("code-queue-scheduler".to_string());
    k3sctl_adapter_microservice_response(
        state,
        &scheduler_service,
        method,
        target_path,
        proxy_options,
        request_headers,
        body_text,
    )
    .await
}

fn provider_http_tunnel_request_id(provider_id: &str) -> String {
    format!(
        "{provider_id}:http_{}_{}",
        chrono::Utc::now().timestamp_millis(),
        rand::random::<u64>()
    )
}

fn can_retry_provider_http_tunnel(method: &Method, target_path: &str) -> bool {
    (method == Method::GET || method == Method::HEAD) && !target_path.ends_with("/stream")
}

fn tunnel_error_body(
    service: &MicroserviceConfig,
    request_id: &str,
    error: &str,
    stage: &str,
    status: u16,
    extra: Value,
) -> Response {
    let mut object = serde_json::Map::new();
    object.insert("ok".to_string(), Value::Bool(false));
    object.insert("error".to_string(), Value::String(error.to_string()));
    object.insert("stage".to_string(), Value::String(stage.to_string()));
    object.insert(
        "providerId".to_string(),
        Value::String(service.provider_id.clone()),
    );
    object.insert("serviceId".to_string(), Value::String(service.id.clone()));
    object.insert(
        "requestId".to_string(),
        Value::String(request_id.to_string()),
    );
    if let Some(extra) = extra.as_object() {
        for (key, value) in extra {
            object.insert(key.clone(), value.clone());
        }
    }
    let mut response = json_response(Value::Object(object), status);
    add_header(&mut response, "x-unidesk-request-id", request_id);
    add_header(&mut response, "x-unidesk-provider-id", &service.provider_id);
    add_header(&mut response, "x-unidesk-service-id", &service.id);
    add_header(&mut response, "x-unidesk-tunnel-error", stage);
    if matches!(status, 502 | 503 | 504) {
        add_header(&mut response, "x-unidesk-transient-error", "true");
    }
    response
}

async fn wait_for_provider_http_tunnel_response(
    state: &Arc<AppState>,
    provider_id: &str,
    request_id: &str,
    timeout_ms: u64,
) -> Result<HttpTunnelResponse, String> {
    let (tx, rx) = tokio::sync::oneshot::channel();
    state
        .http_tunnel_waiters
        .lock()
        .await
        .insert(request_id.to_string(), tx);
    match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), rx).await {
        Ok(Ok(Ok(message))) if message.provider_id == provider_id => Ok(message),
        Ok(Ok(Ok(_))) => Err("provider-mismatch".to_string()),
        Ok(Ok(Err(reason))) => Err(reason),
        Ok(Err(_)) => Err("aborted".to_string()),
        Err(_) => {
            state.http_tunnel_waiters.lock().await.remove(request_id);
            Err("timeout".to_string())
        }
    }
}

async fn provider_http_tunnel_microservice_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
    request_headers: &Value,
    body_text: String,
) -> Response {
    let retryable = can_retry_provider_http_tunnel(method, target_path);
    let max_attempts = if retryable {
        PROVIDER_HTTP_TUNNEL_MAX_ATTEMPTS
    } else {
        1
    };
    let mut attempts = Vec::<Value>::new();
    let mut last_request_id = String::new();
    for attempt in 1..=max_attempts {
        let request_id = provider_http_tunnel_request_id(&service.provider_id);
        last_request_id = request_id.clone();
        let provider = {
            state
                .active_providers
                .read()
                .await
                .get(&service.provider_id)
                .cloned()
        };
        let Some(provider) = provider else {
            attempts.push(json!({ "attempt": attempt, "requestId": request_id, "ok": false, "reason": "provider-offline" }));
            return tunnel_error_body(
                service,
                &request_id,
                &format!("provider is offline: {}", service.provider_id),
                "provider-gateway-online",
                503,
                json!({ "retryable": retryable, "attempts": attempts }),
            );
        };
        let timeout_ms = service.backend.timeout_ms.max(1000) + 3000;
        let started = millis_now();
        let payload = json!({
            "type": "http_tunnel_request",
            "requestId": request_id,
            "payload": {
                "source": "microservice-frontend-proxy",
                "serviceId": service.id,
                "method": method.as_str(),
                "targetBaseUrl": service.backend.node_base_url,
                "path": target_path,
                "query": proxy_options.query,
                "requestHeaders": request_headers,
                "bodyText": body_text,
                "jsonArrayLimits": proxy_options.json_array_limits,
                "timeoutMs": service.backend.timeout_ms,
                "cacheTtlMs": provider_microservice_cache_ttl_ms(&service.id, target_path),
            }
        });
        if provider
            .sender
            .send(axum::extract::ws::Message::Text(payload.to_string()))
            .is_err()
        {
            attempts.push(json!({ "attempt": attempt, "requestId": request_id, "ok": false, "reason": "send-failed", "durationMs": millis_now() - started }));
            if attempt < max_attempts {
                tokio::time::sleep(Duration::from_millis((100 * attempt) as u64)).await;
                continue;
            }
            return tunnel_error_body(
                service,
                &request_id,
                "provider HTTP tunnel send failed",
                "http-tunnel-send",
                502,
                json!({ "retryable": retryable, "attempts": attempts }),
            );
        }
        match wait_for_provider_http_tunnel_response(
            state,
            &service.provider_id,
            &request_id,
            timeout_ms,
        )
        .await
        {
            Ok(message) if message.ok => {
                attempts.push(json!({ "attempt": attempt, "requestId": request_id, "ok": true, "durationMs": millis_now() - started }));
                let mut response = response_from_provider_microservice_result(
                    message.result,
                    "provider-ws-http-tunnel",
                );
                add_header(&mut response, "x-unidesk-request-id", &request_id);
                add_header(
                    &mut response,
                    "x-unidesk-http-tunnel-attempt",
                    &attempt.to_string(),
                );
                add_header(
                    &mut response,
                    "x-unidesk-http-tunnel-attempts",
                    &attempts.len().to_string(),
                );
                add_header(&mut response, "x-unidesk-provider-id", &service.provider_id);
                return response;
            }
            Ok(message) => {
                attempts.push(json!({ "attempt": attempt, "requestId": request_id, "ok": false, "durationMs": millis_now() - started, "result": compact_json(&message.result) }));
                if retryable && attempt < max_attempts {
                    tokio::time::sleep(Duration::from_millis((100 * attempt) as u64)).await;
                    continue;
                }
                return tunnel_error_body(
                    service,
                    &request_id,
                    "provider HTTP tunnel failed",
                    "provider-gateway-http-fetch",
                    502,
                    json!({ "retryable": retryable, "attempts": attempts, "result": message.result }),
                );
            }
            Err(reason) => {
                attempts.push(json!({ "attempt": attempt, "requestId": request_id, "ok": false, "reason": reason, "durationMs": millis_now() - started, "timeoutMs": timeout_ms }));
                if retryable
                    && attempt < max_attempts
                    && matches!(
                        reason.as_str(),
                        "timeout" | "provider-disconnected" | "send-failed"
                    )
                {
                    tokio::time::sleep(Duration::from_millis((100 * attempt) as u64)).await;
                    continue;
                }
                let status = match reason.as_str() {
                    "aborted" => 499,
                    "provider-disconnected" => 503,
                    "send-failed" => 502,
                    _ => 504,
                };
                return tunnel_error_body(
                    service,
                    &request_id,
                    "provider HTTP tunnel timed out or disconnected",
                    if reason == "aborted" {
                        "client-aborted"
                    } else {
                        "http-tunnel-wait"
                    },
                    status,
                    json!({ "retryable": retryable, "attempts": attempts, "timeoutMs": timeout_ms, "failureReason": reason }),
                );
            }
        }
    }
    tunnel_error_body(
        service,
        &last_request_id,
        "provider HTTP tunnel exhausted attempts",
        "http-tunnel-wait",
        504,
        json!({ "retryable": retryable, "attempts": attempts }),
    )
}

async fn fetch_microservice_upstream_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
    request_headers: &Value,
    body_text: String,
) -> Response {
    if service.id == "code-queue" {
        if target_path == "/health" || target_path == "/" {
            return direct_microservice_response(
                &direct_code_queue_mgr_service(service, None),
                method,
                target_path,
                proxy_options,
                request_headers,
                body_text,
            )
            .await;
        }
        if path_matches_task_action(target_path, "interrupt") {
            if let Some(task_id) = code_queue_task_id_for_action(target_path, "interrupt") {
                let meta = direct_microservice_response(
                    &direct_code_queue_mgr_service(service, Some(3000)),
                    &Method::GET,
                    &format!("/api/tasks/{}", urlencoding_like(&task_id)),
                    &ProxyOptions {
                        query: "?meta=1".to_string(),
                        json_array_limits: json!({}),
                    },
                    &json!({ "accept": "application/json" }),
                    String::new(),
                )
                .await;
                if meta.status().is_success() {
                    if let Ok((_, _, text)) = response_to_text(meta).await {
                        if let Ok(body) = serde_json::from_str::<Value>(&text) {
                            let status = body
                                .get("status")
                                .or_else(|| body.pointer("/task/status"))
                                .and_then(Value::as_str)
                                .unwrap_or("");
                            if status == "queued" || status == "retry_wait" {
                                return direct_microservice_response(
                                    &direct_code_queue_mgr_service(service, None),
                                    method,
                                    target_path,
                                    proxy_options,
                                    request_headers,
                                    body_text,
                                )
                                .await;
                            }
                        }
                    }
                }
            }
            return code_queue_k3s_scheduler_response(
                state,
                service,
                method,
                target_path,
                proxy_options,
                request_headers,
                body_text,
            )
            .await;
        }
        if code_queue_scheduler_only_path(method, target_path) {
            return code_queue_k3s_scheduler_response(
                state,
                service,
                method,
                target_path,
                proxy_options,
                request_headers,
                body_text,
            )
            .await;
        }
        let response = direct_microservice_response(
            &direct_code_queue_mgr_service(service, None),
            method,
            target_path,
            proxy_options,
            request_headers,
            body_text,
        )
        .await;
        if method == Method::GET && target_path == "/api/workdirs" {
            let (status, content_type, response_body) =
                response_to_text(response).await.unwrap_or((
                    502,
                    "text/plain".to_string(),
                    "failed to read code-queue workdirs response".to_string(),
                ));
            if let Some(fallback) = code_queue_fallback_workdirs_response(
                service,
                status,
                &content_type,
                &response_body,
            ) {
                return fallback;
            }
            return response_with_body(status, &content_type, Body::from(response_body));
        }
        return response;
    }
    if is_k3sctl_managed_microservice(service) {
        return k3sctl_adapter_microservice_response(
            state,
            service,
            method,
            target_path,
            proxy_options,
            request_headers,
            body_text,
        )
        .await;
    }
    if can_direct_proxy_microservice(service) {
        return direct_microservice_response(
            service,
            method,
            target_path,
            proxy_options,
            request_headers,
            body_text,
        )
        .await;
    }
    if !provider_supports(state, &service.provider_id, "microservice.http")
        .await
        .unwrap_or(false)
    {
        return json_response(
            json!({ "ok": false, "error": format!("provider does not declare microservice.http capability: {}", service.provider_id) }),
            409,
        );
    }
    if provider_supports(state, &service.provider_id, "microservice.http.tunnel")
        .await
        .unwrap_or(false)
    {
        return provider_http_tunnel_microservice_response(
            state,
            service,
            method,
            target_path,
            proxy_options,
            request_headers,
            body_text,
        )
        .await;
    }
    let payload = json!({
        "source": "microservice-frontend-proxy",
        "serviceId": service.id,
        "method": method.as_str(),
        "targetBaseUrl": service.backend.node_base_url,
        "path": target_path,
        "query": proxy_options.query,
        "requestHeaders": request_headers,
        "bodyText": body_text,
        "jsonArrayLimits": proxy_options.json_array_limits,
        "timeoutMs": service.backend.timeout_ms,
        "cacheTtlMs": provider_microservice_cache_ttl_ms(&service.id, target_path),
    });
    match create_and_send_task(state, &service.provider_id, "microservice.http", payload).await {
        Ok((task_id, true)) => {
            match wait_for_task_terminal(state, &task_id, service.backend.timeout_ms + 3000).await {
                Ok(Some(task)) if task.status == "succeeded" => {
                    response_from_provider_microservice_result(
                        task.result.unwrap_or(Value::Null),
                        "provider-task",
                    )
                }
                Ok(task) => json_response(
                    json!({ "ok": false, "error": "microservice proxy task failed", "task": task.map(|task| json!({ "id": task.id, "status": task.status, "result": task.result })) }),
                    502,
                ),
                Err(error) => json_response(
                    json!({ "ok": false, "error": error.to_string(), "taskId": task_id }),
                    502,
                ),
            }
        }
        Ok((task_id, false)) => json_response(
            json!({ "ok": false, "error": format!("provider is offline: {}", service.provider_id), "taskId": task_id }),
            503,
        ),
        Err(error) => json_response(json!({ "ok": false, "error": error.to_string() }), 500),
    }
}

fn microservice_cache_key(
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    proxy_options: &ProxyOptions,
) -> String {
    json!([
        service.id,
        method.as_str(),
        target_path,
        proxy_options.query,
        proxy_options.json_array_limits
    ])
    .to_string()
}

fn microservice_path_fallback_cache_key(
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
) -> String {
    json!([
        service.id,
        method.as_str(),
        target_path,
        "__path_fallback__"
    ])
    .to_string()
}

fn can_use_microservice_path_fallback(
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
) -> bool {
    service.id == "code-queue"
        && target_path == "/api/tasks/overview"
        && (method == Method::GET || method == Method::HEAD)
}

fn response_from_microservice_cache(entry: &MicroserviceProxyCacheEntry, state: &str) -> Response {
    let mut response =
        response_with_body(entry.status, &entry.content_type, entry.body_text.clone());
    add_header(&mut response, "x-unidesk-cache", state);
    response
}

async fn read_microservice_cache(state: &Arc<AppState>, key: &str) -> Option<Response> {
    let now = millis_now();
    let mut cache = state.microservice_proxy_cache.lock().await;
    let entry = cache.get(key)?.clone();
    if entry.stale_expires_at_ms <= now {
        cache.remove(key);
        return None;
    }
    if entry.expires_at_ms <= now {
        return None;
    }
    Some(response_from_microservice_cache(&entry, "hit"))
}

async fn read_stale_microservice_cache(state: &Arc<AppState>, key: &str) -> Option<Response> {
    let now = millis_now();
    let mut cache = state.microservice_proxy_cache.lock().await;
    let entry = cache.get(key)?.clone();
    if entry.stale_expires_at_ms <= now {
        cache.remove(key);
        return None;
    }
    if entry.expires_at_ms > now {
        return None;
    }
    Some(response_from_microservice_cache(&entry, "stale"))
}

async fn read_microservice_path_fallback(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
) -> Option<Response> {
    if !can_use_microservice_path_fallback(service, method, target_path) {
        return None;
    }
    let key = microservice_path_fallback_cache_key(service, method, target_path);
    let now = millis_now();
    let mut cache = state.microservice_proxy_cache.lock().await;
    let entry = cache.get(&key)?.clone();
    if entry.stale_expires_at_ms <= now {
        cache.remove(&key);
        return None;
    }
    Some(response_from_microservice_cache(&entry, "path-stale"))
}

async fn cacheable_response_snapshot(
    response: Response,
) -> (Response, Option<MicroserviceProxyCacheEntry>) {
    let status = response.status().as_u16();
    let headers = response.headers().clone();
    let content_type = headers
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("application/octet-stream")
        .to_string();
    let bytes = to_bytes(response.into_body(), 16 * 1024 * 1024)
        .await
        .unwrap_or_default();
    let body_text = String::from_utf8_lossy(&bytes).to_string();
    let rebuilt = response_with_body(status, &content_type, body_text.clone());
    if !(200..300).contains(&status)
        || headers
            .get("x-unidesk-response-truncated")
            .and_then(|value| value.to_str().ok())
            == Some("true")
        || body_text.len() > 2 * 1024 * 1024
    {
        return (rebuilt, None);
    }
    (
        rebuilt,
        Some(MicroserviceProxyCacheEntry {
            expires_at_ms: 0,
            stale_expires_at_ms: 0,
            status,
            content_type,
            body_text,
        }),
    )
}

async fn remember_microservice_cache(
    state: &Arc<AppState>,
    key: String,
    ttl_ms: i64,
    mut entry: Option<MicroserviceProxyCacheEntry>,
) {
    if ttl_ms <= 0 {
        return;
    }
    let Some(ref mut entry) = entry else { return };
    let parsed: Value = serde_json::from_str(&key).unwrap_or(Value::Null);
    let service_id = parsed.get(0).and_then(Value::as_str).unwrap_or("");
    let target_path = parsed.get(2).and_then(Value::as_str).unwrap_or("");
    let now = millis_now();
    entry.expires_at_ms = now + ttl_ms;
    entry.stale_expires_at_ms =
        entry.expires_at_ms + microservice_cache_stale_ms(service_id, target_path);
    let mut cache = state.microservice_proxy_cache.lock().await;
    cache.insert(key, entry.clone());
    if cache.len() > 300 {
        let now = millis_now();
        cache.retain(|_, entry| entry.expires_at_ms > now);
        if cache.len() > 240 {
            cache.clear();
        }
    }
}

async fn remember_microservice_path_fallback(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    method: &Method,
    target_path: &str,
    entry: Option<MicroserviceProxyCacheEntry>,
) {
    if !can_use_microservice_path_fallback(service, method, target_path) {
        return;
    }
    let Some(mut entry) = entry else { return };
    let now = millis_now();
    entry.expires_at_ms = now;
    entry.stale_expires_at_ms = now + CODE_QUEUE_OVERVIEW_PATH_FALLBACK_STALE_MS;
    state.microservice_proxy_cache.lock().await.insert(
        microservice_path_fallback_cache_key(service, method, target_path),
        entry,
    );
}

async fn invalidate_microservice_cache(state: &Arc<AppState>, service_id: &str) {
    let prefix = format!("[\"{service_id}\",");
    state
        .microservice_proxy_cache
        .lock()
        .await
        .retain(|key, _| !key.starts_with(&prefix));
}

fn is_microservice_transient_failure_response(response: &Response) -> bool {
    matches!(response.status().as_u16(), 502 | 503 | 504)
        && (response
            .headers()
            .get("x-unidesk-transient-error")
            .is_some()
            || response.headers().get("x-unidesk-tunnel-error").is_some()
            || response
                .headers()
                .get("x-unidesk-upstream-proxy-mode")
                .and_then(|value| value.to_str().ok())
                == Some("provider-gateway-http-fetch"))
}

fn json_body_from_text(body_text: &str, content_type: &str) -> Value {
    if !content_type_is_json(content_type) {
        return if body_text.is_empty() {
            Value::Null
        } else {
            Value::String(truncate_text(body_text, 4000))
        };
    }
    serde_json::from_str::<Value>(body_text)
        .map(|value| compact_json(&value))
        .unwrap_or_else(|_| Value::String(truncate_text(body_text, 4000)))
}

fn generic_microservice_health_assessment(
    upstream_status: u16,
    body: &Value,
) -> (bool, String, Value) {
    let mut checks = serde_json::Map::new();
    checks.insert("upstreamStatus".to_string(), json!(upstream_status));
    checks.insert(
        "upstream2xx".to_string(),
        json!((200..300).contains(&upstream_status)),
    );
    if !(200..300).contains(&upstream_status) {
        return (
            false,
            format!("health endpoint returned HTTP {upstream_status}"),
            Value::Object(checks),
        );
    }
    if let Some(object) = body.as_object() {
        let ok = object.get("ok").and_then(Value::as_bool);
        let success = object.get("success").and_then(Value::as_bool);
        let status = object.get("status").and_then(Value::as_str).unwrap_or("");
        checks.insert(
            "okField".to_string(),
            ok.map(Value::Bool).unwrap_or(Value::Null),
        );
        checks.insert(
            "successField".to_string(),
            success.map(Value::Bool).unwrap_or(Value::Null),
        );
        checks.insert(
            "statusField".to_string(),
            if status.is_empty() {
                Value::Null
            } else {
                Value::String(status.to_string())
            },
        );
        if ok == Some(false) {
            return (
                false,
                "health body ok=false".to_string(),
                Value::Object(checks),
            );
        }
        if success == Some(false) {
            return (
                false,
                "health body success=false".to_string(),
                Value::Object(checks),
            );
        }
        if matches!(
            status.to_ascii_lowercase().as_str(),
            "error"
                | "failed"
                | "fail"
                | "unhealthy"
                | "down"
                | "offline"
                | "not_ready"
                | "not-ready"
        ) {
            return (
                false,
                format!("health body status={status}"),
                Value::Object(checks),
            );
        }
    }
    (
        true,
        "health endpoint returned a usable service state".to_string(),
        Value::Object(checks),
    )
}

fn health_probe_from_assessment(
    service: &MicroserviceConfig,
    upstream_status: u16,
    content_type: &str,
    healthy: bool,
    reason: String,
    checks: Value,
    body: Option<Value>,
) -> Value {
    let mut probe = json!({
        "serviceId": service.id,
        "name": service.name,
        "providerId": service.provider_id,
        "healthPath": service.backend.health_path,
        "healthy": healthy,
        "status": if healthy { "healthy" } else { "unhealthy" },
        "reason": reason,
        "upstreamStatus": upstream_status,
        "contentType": content_type,
        "checks": checks,
        "checkedAt": crate::json_util::now_iso(),
    });
    if let Some(body) = body {
        probe["body"] = body;
    }
    probe
}

async fn remember_microservice_availability(state: &Arc<AppState>, service_id: &str, probe: Value) {
    state.microservice_availability_cache.lock().await.insert(
        service_id.to_string(),
        (millis_now() + MICROSERVICE_AVAILABILITY_TTL_MS, probe),
    );
}

async fn cached_microservice_availability(
    state: &Arc<AppState>,
    service_id: &str,
) -> Option<Value> {
    let now = millis_now();
    let mut cache = state.microservice_availability_cache.lock().await;
    let (expires, probe) = cache.get(service_id)?.clone();
    if expires <= now {
        cache.remove(service_id);
        None
    } else {
        Some(probe)
    }
}

pub async fn strict_microservice_health_probe(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    timeout_ms: Option<u64>,
) -> Value {
    let mut probe_service = service.clone();
    if let Some(timeout_ms) = timeout_ms {
        probe_service.backend.timeout_ms = timeout_ms;
    }
    let response = if service.id == "code-queue" {
        code_queue_health_response(state, &probe_service, false).await
    } else {
        fetch_microservice_upstream_response(
            state,
            &probe_service,
            &Method::GET,
            &service.backend.health_path,
            &ProxyOptions {
                query: String::new(),
                json_array_limits: json!({}),
            },
            &json!({ "accept": "application/json" }),
            String::new(),
        )
        .await
    };
    let (status, content_type, body_text) = response_to_text(response).await.unwrap_or((
        502,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let body = json_body_from_text(&body_text, &content_type);
    let (healthy, reason, checks) = generic_microservice_health_assessment(status, &body);
    let probe = health_probe_from_assessment(
        service,
        status,
        &content_type,
        healthy,
        reason,
        checks,
        Some(body),
    );
    remember_microservice_availability(state, &service.id, probe.clone()).await;
    probe
}

async fn microservice_availability(state: &Arc<AppState>, service: &MicroserviceConfig) -> Value {
    if let Some(cached) = cached_microservice_availability(state, &service.id).await {
        return cached;
    }
    let timeout_ms = service.backend.timeout_ms.clamp(2500, 10_000);
    strict_microservice_health_probe(state, service, Some(timeout_ms)).await
}

async fn code_queue_health_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    head_only: bool,
) -> Response {
    let mgr_response = direct_microservice_response(
        &direct_code_queue_mgr_service(service, Some(4000)),
        &Method::GET,
        "/health",
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let (mgr_status, _, mgr_text) = response_to_text(mgr_response).await.unwrap_or((
        502,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let mgr_body: Value = serde_json::from_str(&mgr_text)
        .unwrap_or_else(|_| json!({ "text": truncate_text(&mgr_text, 4000) }));
    let trace_body = mgr_body
        .get("traceRead")
        .cloned()
        .unwrap_or_else(|| json!({}));
    let mgr_healthy = (200..300).contains(&mgr_status)
        && mgr_body.get("ok").and_then(Value::as_bool) != Some(false);

    let scheduler_response = code_queue_k3s_scheduler_response(
        state,
        service,
        &Method::GET,
        "/health",
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let (scheduler_status, _, scheduler_text) = response_to_text(scheduler_response)
        .await
        .unwrap_or((504, "application/json".to_string(), "{}".to_string()));
    let scheduler_body: Value = serde_json::from_str(&scheduler_text)
        .unwrap_or_else(|_| json!({ "text": truncate_text(&scheduler_text, 4000) }));
    let scheduler_healthy = (200..300).contains(&scheduler_status)
        && scheduler_body.get("ok").and_then(Value::as_bool) != Some(false);
    let scheduler_diagnostics =
        code_queue_adapter_diagnostics(state, service, "code-queue-scheduler").await;
    let (provider_egress_service, tcp_egress_service) = code_queue_infra_service_names(service);
    let provider_egress_diagnostics =
        code_queue_adapter_diagnostics(state, service, &provider_egress_service).await;
    let tcp_egress_diagnostics =
        code_queue_adapter_diagnostics(state, service, &tcp_egress_service).await;
    let scheduler_dependency =
        code_queue_dependency_check("code-queue-scheduler", &scheduler_diagnostics);
    let provider_egress =
        code_queue_dependency_check(&provider_egress_service, &provider_egress_diagnostics);
    let tcp_egress =
        code_queue_dependency_check(&tcp_egress_service, &tcp_egress_diagnostics);
    let postgres_route = code_queue_route_postgres_check(&scheduler_body, &tcp_egress_service);
    let stale_reconcile = code_queue_reconcile_check(&scheduler_body);
    let dependencies_ok = scheduler_dependency
        .get("ok")
        .and_then(Value::as_bool)
        .unwrap_or(false)
        && provider_egress
            .get("ok")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        && tcp_egress
            .get("ok")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        && postgres_route
            .get("ok")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        && stale_reconcile
            .get("ok")
            .and_then(Value::as_bool)
            .unwrap_or(false);
    let ok = mgr_healthy && scheduler_healthy && dependencies_ok;
    let status = if ok { 200 } else { 503 };
    let body = json!({
        "ok": ok,
        "service": "code-queue",
        "role": "hybrid-master-control-plane",
        "status": if ok { "healthy" } else { "degraded" },
        "checkedAt": crate::json_util::now_iso(),
        "manager": { "ok": mgr_healthy, "status": mgr_status, "body": mgr_body, "timedOut": false },
        "trace": { "ok": mgr_healthy, "source": "code-queue-mgr", "readonlyUserConfigured": trace_body.get("readonlyUserConfigured").cloned().unwrap_or(Value::Null), "body": trace_body },
        "scheduler": { "ok": scheduler_healthy, "status": scheduler_status, "body": scheduler_body, "timedOut": false, "requiredFor": ["running task steer", "running task interrupt", "scheduler claim/runner execution", "dev-container control"] },
        "checks": {
            "scheduler": scheduler_dependency,
            "providerEgressProxy": provider_egress,
            "tcpEgressGateway": tcp_egress,
            "postgresRoute": postgres_route,
            "staleActiveReconcile": stale_reconcile
        },
        "diagnostics": {
            "scheduler": scheduler_diagnostics,
            "providerEgressProxy": provider_egress_diagnostics,
            "tcpEgressGateway": tcp_egress_diagnostics
        },
    });
    if head_only {
        response_with_body(status, "application/json; charset=utf-8", Body::empty())
    } else {
        json_response(body, status)
    }
}

async fn strict_microservice_health_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    head_only: bool,
) -> Response {
    if service.id == "code-queue" {
        return code_queue_health_response(state, service, head_only).await;
    }
    let response = fetch_microservice_upstream_response(
        state,
        service,
        &Method::GET,
        &service.backend.health_path,
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let (upstream_status, content_type, body_text) = response_to_text(response).await.unwrap_or((
        502,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let body = json_body_from_text(&body_text, &content_type);
    let (healthy, reason, checks) = generic_microservice_health_assessment(upstream_status, &body);
    let probe = health_probe_from_assessment(
        service,
        upstream_status,
        &content_type,
        healthy,
        reason.clone(),
        checks.clone(),
        None,
    );
    remember_microservice_availability(state, &service.id, probe).await;
    if healthy {
        let mut response = response_with_body(
            upstream_status,
            &content_type,
            if head_only { String::new() } else { body_text },
        );
        add_header(&mut response, "x-unidesk-health", "healthy");
        add_header(&mut response, "x-unidesk-health-reason", &reason);
        return response;
    }
    json_response(
        json!({
            "ok": false,
            "status": "unhealthy",
            "serviceId": service.id,
            "name": service.name,
            "providerId": service.provider_id,
            "reason": reason,
            "checkedAt": crate::json_util::now_iso(),
            "checks": checks,
            "upstream": { "status": upstream_status, "contentType": content_type, "body": body },
        }),
        if upstream_status >= 500 {
            upstream_status
        } else {
            503
        },
    )
}

async fn k3sctl_managed_diagnostics_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
) -> Response {
    let adapter_service_id = service
        .deployment
        .adapter_service_id
        .as_deref()
        .unwrap_or("k3sctl-adapter");
    let adapter = service_by_id(state, adapter_service_id);
    let provider_id = adapter
        .as_ref()
        .map(|service| service.provider_id.clone())
        .unwrap_or_else(|| service.provider_id.clone());
    let provider_online = state
        .active_providers
        .read()
        .await
        .contains_key(&provider_id);
    let provider_tunnel_capable =
        provider_supports(state, &provider_id, "microservice.http.tunnel")
            .await
            .unwrap_or(false);
    let Some(adapter) = adapter else {
        return json_response(
            json!({
                "ok": false,
                "serviceId": service.id,
                "checkedAt": crate::json_util::now_iso(),
                "requestPath": "/diagnostics",
                "checks": {
                    "providerGateway": { "ok": provider_online, "providerId": provider_id, "online": provider_online },
                    "httpTunnel": { "ok": provider_tunnel_capable, "providerId": provider_id, "capable": provider_tunnel_capable },
                    "k3sctlAdapter": { "ok": false, "serviceId": adapter_service_id, "error": format!("k3sctl adapter microservice not found: {adapter_service_id}") },
                    "kubernetesApiServiceProxy": { "ok": false, "skipped": true },
                    "targetService": { "ok": false, "skipped": true }
                }
            }),
            502,
        );
    };
    let k3s_service_id = if service.id == "code-queue" {
        code_queue_k3s_service_id_for_request(&Method::GET, &service.backend.health_path)
            .to_string()
    } else {
        service
            .deployment
            .k3s_service_id
            .clone()
            .unwrap_or_else(|| service.id.clone())
    };
    let adapter_path = format!(
        "/api/services/{}/diagnostics",
        urlencoding_like(&k3s_service_id)
    );
    let response = fetch_microservice_upstream_response(
        state,
        &adapter,
        &Method::GET,
        &adapter_path,
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let status = response.status().as_u16();
    let proxy_mode = response
        .headers()
        .get("x-unidesk-proxy-mode")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("")
        .to_string();
    let request_id = response
        .headers()
        .get("x-unidesk-request-id")
        .and_then(|value| value.to_str().ok())
        .map(ToOwned::to_owned);
    let (status2, content_type, body_text) = response_to_text(response).await.unwrap_or((
        status,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let adapter_body: Value = serde_json::from_str(&body_text)
        .unwrap_or_else(|_| Value::String(truncate_text(&body_text, 4000)));
    let adapter_checks = adapter_body
        .get("checks")
        .cloned()
        .unwrap_or_else(|| json!({}));
    let mut code_queue_checks = Value::Null;
    let mut code_queue_dependencies = Value::Null;
    let mut code_queue_adapter = Value::Null;
    let mut code_queue_ok = true;
    if service.id == "code-queue" {
        let scheduler_body = adapter_body
            .get("checks")
            .and_then(|checks| checks.get("targetService"))
            .and_then(|target| target.get("body"))
            .cloned()
            .unwrap_or(Value::Null);
        let scheduler_diagnostics =
            code_queue_adapter_diagnostics(state, service, "code-queue-scheduler").await;
        let (provider_egress_service, tcp_egress_service) = code_queue_infra_service_names(service);
        let provider_egress_diagnostics =
            code_queue_adapter_diagnostics(state, service, &provider_egress_service).await;
        let tcp_egress_diagnostics =
            code_queue_adapter_diagnostics(state, service, &tcp_egress_service).await;
        let scheduler_dependency =
            code_queue_dependency_check("code-queue-scheduler", &scheduler_diagnostics);
        let provider_egress =
            code_queue_dependency_check(&provider_egress_service, &provider_egress_diagnostics);
        let tcp_egress =
            code_queue_dependency_check(&tcp_egress_service, &tcp_egress_diagnostics);
        let postgres_route = code_queue_route_postgres_check(&scheduler_body, &tcp_egress_service);
        let stale_reconcile = code_queue_reconcile_check(&scheduler_body);
        code_queue_ok = scheduler_dependency
            .get("ok")
            .and_then(Value::as_bool)
            .unwrap_or(false)
            && provider_egress
                .get("ok")
                .and_then(Value::as_bool)
                .unwrap_or(false)
            && tcp_egress
                .get("ok")
                .and_then(Value::as_bool)
                .unwrap_or(false)
            && postgres_route
                .get("ok")
                .and_then(Value::as_bool)
                .unwrap_or(false)
            && stale_reconcile
                .get("ok")
                .and_then(Value::as_bool)
                .unwrap_or(false);
        code_queue_checks = json!({
            "scheduler": scheduler_dependency,
            "providerEgressProxy": provider_egress,
            "tcpEgressGateway": tcp_egress,
            "postgresRoute": postgres_route,
            "staleActiveReconcile": stale_reconcile
        });
        code_queue_dependencies = json!({
            "postgresRoute": format!("{tcp_egress_service}.unidesk.svc.cluster.local:15432"),
            "requiredDeployments": [tcp_egress_service, provider_egress_service, "code-queue".to_string()],
            "requiredEndpoints": [tcp_egress_service, provider_egress_service, "code-queue-scheduler".to_string()]
        });
        code_queue_adapter = json!({
            "scheduler": scheduler_diagnostics,
            "providerEgressProxy": provider_egress_diagnostics,
            "tcpEgressGateway": tcp_egress_diagnostics
        });
    }
    let ok = (200..300).contains(&status2)
        && provider_online
        && proxy_mode == "provider-ws-http-tunnel"
        && code_queue_ok;
    json_response(
        json!({
            "ok": ok,
            "serviceId": service.id,
            "k3sServiceId": k3s_service_id,
            "checkedAt": crate::json_util::now_iso(),
            "path": service.backend.health_path,
            "chain": "CLI/frontend -> backend-core -> provider-gateway HTTP tunnel -> k3sctl-adapter -> Kubernetes API service proxy -> k3s Service",
            "checks": {
                "providerGateway": { "ok": provider_online, "providerId": provider_id, "online": provider_online, "activeSocketCount": state.active_provider_count().await },
                "httpTunnel": { "ok": proxy_mode == "provider-ws-http-tunnel", "providerId": provider_id, "capable": provider_tunnel_capable, "requestId": request_id, "proxyStatus": status2 },
                "k3sctlAdapter": { "ok": (200..300).contains(&status2), "serviceId": adapter.id, "providerId": adapter.provider_id, "status": status2, "contentType": content_type },
                "kubernetesApiServiceProxy": compact_json(adapter_checks.get("kubernetesApiServiceProxy").unwrap_or(&json!({ "ok": false, "skipped": true }))),
                "targetService": compact_json(adapter_checks.get("targetService").or_else(|| adapter_checks.get("managedService")).unwrap_or(&json!({ "ok": false, "skipped": true }))),
            },
            "codeQueue": {
                "ok": code_queue_ok,
                "checks": code_queue_checks,
                "dependencies": code_queue_dependencies,
                "adapter": code_queue_adapter
            },
            "adapter": adapter_body,
        }),
        if ok {
            200
        } else if (200..300).contains(&status2) {
            503
        } else {
            status2
        },
    )
}

async fn microservice_tunnel_self_test_response(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
) -> Response {
    let tunnel_service = if is_k3sctl_managed_microservice(service) {
        service_by_id(
            state,
            service
                .deployment
                .adapter_service_id
                .as_deref()
                .unwrap_or("k3sctl-adapter"),
        )
    } else {
        Some(service.clone())
    };
    let Some(mut tunnel_service) = tunnel_service else {
        return json_response(
            json!({ "ok": false, "serviceId": service.id, "error": "tunnel service not found", "adapterServiceId": service.deployment.adapter_service_id }),
            502,
        );
    };
    if !provider_supports(
        state,
        &tunnel_service.provider_id,
        "microservice.http.tunnel",
    )
    .await
    .unwrap_or(false)
    {
        return json_response(
            json!({ "ok": false, "serviceId": service.id, "providerId": tunnel_service.provider_id, "error": format!("provider does not declare microservice.http.tunnel capability: {}", tunnel_service.provider_id) }),
            409,
        );
    }
    tunnel_service.backend.node_base_url = "http://127.0.0.1:1".to_string();
    tunnel_service.backend.timeout_ms = 1000;
    let response = provider_http_tunnel_microservice_response(
        state,
        &tunnel_service,
        &Method::GET,
        "/",
        &ProxyOptions {
            query: String::new(),
            json_array_limits: json!({}),
        },
        &json!({ "accept": "application/json" }),
        String::new(),
    )
    .await;
    let status = response.status().as_u16();
    let request_header = response
        .headers()
        .get("x-unidesk-request-id")
        .and_then(|value| value.to_str().ok())
        .map(ToOwned::to_owned);
    let tunnel_error = response
        .headers()
        .get("x-unidesk-tunnel-error")
        .and_then(|value| value.to_str().ok())
        .map(ToOwned::to_owned);
    let (_, _, body_text) = response_to_text(response).await.unwrap_or((
        status,
        "application/json".to_string(),
        "{}".to_string(),
    ));
    let body: Value = serde_json::from_str(&body_text)
        .unwrap_or_else(|_| Value::String(truncate_text(&body_text, 4000)));
    let has_request_id = body
        .get("requestId")
        .and_then(Value::as_str)
        .is_some_and(|value| !value.is_empty());
    let has_stage = body
        .get("stage")
        .and_then(Value::as_str)
        .is_some_and(|value| !value.is_empty());
    let expected_status = status == 502 || status == 504;
    let ok = expected_status
        && has_request_id
        && has_stage
        && request_header.as_deref() == body.get("requestId").and_then(Value::as_str);
    json_response(
        json!({
            "ok": ok,
            "serviceId": service.id,
            "tunnelServiceId": tunnel_service.id,
            "providerId": tunnel_service.provider_id,
            "expectedFailure": true,
            "status": status,
            "checks": {
                "expectedStatus": expected_status,
                "bodyHasRequestId": has_request_id,
                "bodyHasStage": has_stage,
                "headerHasRequestId": request_header.as_ref().is_some_and(|value| !value.is_empty()),
                "headerHasTunnelError": tunnel_error.as_ref().is_some_and(|value| !value.is_empty()),
            },
            "headers": { "requestId": request_header, "tunnelError": tunnel_error },
            "body": body,
        }),
        if ok { 200 } else { 502 },
    )
}

pub async fn get_microservices(state: &Arc<AppState>) -> anyhow::Result<Vec<Value>> {
    let nodes = get_nodes(state).await?;
    let docker_statuses = get_node_docker_statuses(state).await?;
    let mut result = Vec::new();
    for service in &state.config.microservices {
        let node = nodes.iter().find(|item| {
            item.get("providerId").and_then(Value::as_str) == Some(&service.provider_id)
        });
        let docker = docker_statuses.iter().find(|item| {
            item.get("providerId").and_then(Value::as_str) == Some(&service.provider_id)
        });
        let k3s_managed = is_k3sctl_managed_microservice(service);
        let container = if k3s_managed {
            Value::Null
        } else {
            find_container(
                docker.and_then(|item| item.get("dockerStatus")),
                &service.repository.container_name,
            )
        };
        let provider_status = node
            .and_then(|item| item.get("status"))
            .and_then(Value::as_str)
            .unwrap_or("missing");
        let availability =
            inferred_microservice_availability(state, service, provider_status, &container).await;
        let mut service_json = serde_json::to_value(service)?;
        service_json["runtime"] = json!({
            "orchestrator": if k3s_managed { "k3sctl" } else { "unidesk-direct" },
            "providerStatus": provider_status,
            "providerName": node.and_then(|item| item.get("name")).and_then(Value::as_str).unwrap_or(&service.provider_id),
            "providerLastHeartbeat": node.and_then(|item| item.get("lastHeartbeat")).cloned().unwrap_or(Value::Null),
            "container": container,
            "availability": availability,
            "backendPortMapping": {
                "providerId": service.provider_id,
                "node": format!("{}:{}", service.backend.node_bind_host, service.backend.node_port),
                "mainServerAccess": "frontend-only backend proxy",
                "public": service.backend.public,
            },
        });
        result.push(service_json);
    }
    Ok(result)
}

async fn inferred_microservice_availability(
    state: &Arc<AppState>,
    service: &MicroserviceConfig,
    provider_status: &str,
    container: &Value,
) -> Value {
    if let Some(cached) = cached_microservice_availability(state, &service.id).await {
        return cached;
    }
    let container_state = container
        .get("state")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_ascii_lowercase();
    let container_status = container
        .get("status")
        .and_then(Value::as_str)
        .unwrap_or("");
    if provider_status != "online" {
        json!({ "serviceId": service.id, "healthy": false, "status": "unhealthy", "reason": format!("provider is {provider_status}"), "source": "runtime-inference" })
    } else if !container.is_null() && container_state != "running" {
        json!({ "serviceId": service.id, "healthy": false, "status": "unhealthy", "reason": format!("container is {}", if container_status.is_empty() { &container_state } else { container_status }), "source": "runtime-inference" })
    } else {
        json!({ "serviceId": service.id, "healthy": Value::Null, "status": "unknown", "reason": "strict health has not been probed yet", "source": "runtime-inference" })
    }
}

pub async fn microservice_route(
    state: &Arc<AppState>,
    method: Method,
    uri: Uri,
    request: Request<Body>,
) -> anyhow::Result<Response> {
    let path = uri.path().to_string();
    let rest = &path["/api/microservices/".len()..];
    let slash_index = rest.find('/');
    let service_id = percent_decode(match slash_index {
        Some(index) => &rest[..index],
        None => rest,
    });
    let suffix = slash_index.map(|index| &rest[index + 1..]).unwrap_or("");
    let Some(service) = service_by_id(state, &service_id) else {
        return Ok(json_response(
            json!({ "ok": false, "error": format!("microservice not found: {service_id}") }),
            404,
        ));
    };
    if (suffix.is_empty() || suffix == "status")
        && (method == Method::GET || method == Method::HEAD)
    {
        let microservices = get_microservices(state).await?;
        return Ok(json_response(
            json!({ "ok": true, "microservice": microservices.into_iter().find(|item| item.get("id").and_then(Value::as_str) == Some(&service_id)).unwrap_or_else(|| serde_json::to_value(&service).unwrap_or(Value::Null)) }),
            200,
        ));
    }
    let target_path = if suffix == "health" {
        service.backend.health_path.clone()
    } else if suffix == "proxy" {
        "/".to_string()
    } else if let Some(rest) = suffix.strip_prefix("proxy/") {
        format!("/{rest}")
    } else if suffix == "diagnostics" {
        "/diagnostics".to_string()
    } else if suffix == "tunnel-self-test" {
        "/tunnel-self-test".to_string()
    } else {
        String::new()
    };
    if target_path.is_empty() {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice route must be /status, /health, /diagnostics, /tunnel-self-test, or /proxy/<path>" }),
            404,
        ));
    }
    if suffix == "health" && method != Method::GET && method != Method::HEAD {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice health only supports GET/HEAD" }),
            405,
        ));
    }
    if suffix == "diagnostics" && method != Method::GET && method != Method::HEAD {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice diagnostics only supports GET/HEAD" }),
            405,
        ));
    }
    if suffix == "tunnel-self-test" && method != Method::GET && method != Method::HEAD {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice tunnel self-test only supports GET/HEAD" }),
            405,
        ));
    }
    if !is_microservice_method_allowed(&service, &method) {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice method is not allowed", "serviceId": service_id, "method": method.as_str(), "allowedMethods": service.backend.allowed_methods }),
            405,
        ));
    }
    if suffix == "diagnostics" {
        return Ok(if is_k3sctl_managed_microservice(&service) {
            k3sctl_managed_diagnostics_response(state, &service).await
        } else {
            strict_microservice_health_response(state, &service, method == Method::HEAD).await
        });
    }
    if suffix == "tunnel-self-test" {
        return Ok(microservice_tunnel_self_test_response(state, &service).await);
    }
    if !is_microservice_path_allowed(&service, &target_path) {
        return Ok(json_response(
            json!({ "ok": false, "error": "microservice path is not allowed", "serviceId": service_id, "targetPath": target_path }),
            403,
        ));
    }
    if suffix == "health" {
        return Ok(
            strict_microservice_health_response(state, &service, method == Method::HEAD).await,
        );
    }
    let proxy_options = read_microservice_array_limits(&uri);
    let cache_key = microservice_cache_key(&service, &method, &target_path, &proxy_options);
    let cache_ttl_ms = microservice_cache_ttl_ms(&service.id, &target_path);
    if method == Method::GET || method == Method::HEAD {
        if let Some(response) = read_microservice_cache(state, &cache_key).await {
            return Ok(response);
        }
    } else {
        invalidate_microservice_cache(state, &service.id).await;
    }
    let headers = request.headers().clone();
    let body_text = if method == Method::GET || method == Method::HEAD {
        String::new()
    } else {
        let bytes = to_bytes(request.into_body(), 1024 * 1024 + 1).await?;
        if bytes.len() > 1024 * 1024 {
            return Ok(json_response(
                json!({ "ok": false, "error": "microservice request body is too large", "maxBytes": 1024 * 1024 }),
                413,
            ));
        }
        String::from_utf8_lossy(&bytes).to_string()
    };
    let request_headers = read_microservice_request_headers(&headers);
    if method == Method::GET || method == Method::HEAD {
        if let Some(response) = read_stale_microservice_cache(state, &cache_key).await {
            return Ok(response);
        }
        if let Some(response) =
            read_microservice_path_fallback(state, &service, &method, &target_path).await
        {
            return Ok(response);
        }
    }
    let response = fetch_microservice_upstream_response(
        state,
        &service,
        &method,
        &target_path,
        &proxy_options,
        &request_headers,
        body_text,
    )
    .await;
    let response = rewrite_todo_note_misleading_route_not_found_response(
        &service,
        &method,
        &target_path,
        response,
    )
    .await;
    if (method == Method::GET || method == Method::HEAD)
        && is_microservice_transient_failure_response(&response)
    {
        if let Some(mut stale) = read_stale_microservice_cache(state, &cache_key)
            .await
            .or(read_microservice_path_fallback(state, &service, &method, &target_path).await)
        {
            add_header(&mut stale, "x-unidesk-cache", "stale-on-transient-failure");
            add_header(
                &mut stale,
                "x-unidesk-stale-reason",
                &response.status().as_u16().to_string(),
            );
            return Ok(stale);
        }
    }
    if (method == Method::GET || method == Method::HEAD) && cache_ttl_ms > 0 {
        let (rebuilt, snapshot) = cacheable_response_snapshot(response).await;
        remember_microservice_cache(state, cache_key, cache_ttl_ms, snapshot.clone()).await;
        remember_microservice_path_fallback(state, &service, &method, &target_path, snapshot).await;
        Ok(rebuilt)
    } else {
        Ok(response)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn todo_note_route_diagnostic_matches_only_misleading_404_body() {
        let upstream = r#"{"ok":false,"error":"Todo Note is running in backend-only mode"}"#;
        let matched = is_todo_note_misleading_route_not_found(
            404,
            "application/json; charset=utf-8",
            upstream,
        )
        .expect("misleading body should match");
        assert_eq!(
            matched.get("error").and_then(Value::as_str),
            Some("Todo Note is running in backend-only mode")
        );
        assert!(
            is_todo_note_misleading_route_not_found(
                404,
                "application/json",
                r#"{"ok":false,"error":"Cannot POST /api/todos"}"#,
            )
            .is_none()
        );
        assert!(
            is_todo_note_misleading_route_not_found(200, "application/json", upstream).is_none()
        );
    }

    #[test]
    fn todo_note_route_diagnostic_describes_missing_post_path() {
        let body = todo_note_route_not_found_body(
            &Method::POST,
            "/api/instances/instance_probe_bad/todos",
            json!({ "ok": false, "error": "Todo Note is running in backend-only mode" }),
        );
        assert_eq!(
            body.get("error").and_then(Value::as_str),
            Some("Todo Note route not found")
        );
        assert_eq!(body.get("backendOnly").and_then(Value::as_bool), Some(true));
        assert_eq!(body.get("method").and_then(Value::as_str), Some("POST"));
        assert_eq!(
            body.get("path").and_then(Value::as_str),
            Some("/api/instances/instance_probe_bad/todos")
        );
        assert_eq!(
            body.get("bodyRewritten").and_then(Value::as_bool),
            Some(true)
        );
        assert!(
            body.get("writableApiEndpoints")
                .and_then(Value::as_array)
                .is_some_and(|endpoints| endpoints.iter().any(|endpoint| endpoint
                    .get("path")
                    .and_then(Value::as_str)
                    == Some("/api/instances/:instanceId/actions")))
        );
        assert!(
            body.get("actionTypes")
                .and_then(Value::as_array)
                .is_some_and(|actions| actions
                    .iter()
                    .any(|action| action.as_str() == Some("addTodo")))
        );
        assert_eq!(
            body.pointer("/upstreamBody/error").and_then(Value::as_str),
            Some("Todo Note is running in backend-only mode")
        );
    }
}

fn urlencoding_like(value: &str) -> String {
    percent_encoding::utf8_percent_encode(value, percent_encoding::NON_ALPHANUMERIC).to_string()
}
