use std::sync::Arc;

use anyhow::Context;
use chrono::{DateTime, Duration, Utc};
use serde_json::{json, Value};
use tokio_postgres::Row;

use crate::json_util::{
    compact_json, compact_json_for_storage, nested_number, redact_database_url,
};
use crate::state::AppState;
use crate::types::{JsonValue, RawTaskRow};

pub async fn init_database(state: &Arc<AppState>) -> anyhow::Result<()> {
    state.log(
        "info",
        "database_init_start",
        Some(json!({ "databaseUrl": redact_database_url(&state.config.database_url) })),
    );
    let client = state.pool.get().await?;
    client.batch_execute(
        r#"
        CREATE TABLE IF NOT EXISTS unidesk_nodes (
          provider_id TEXT PRIMARY KEY,
          name TEXT NOT NULL,
          labels JSONB NOT NULL DEFAULT '{}'::jsonb,
          status TEXT NOT NULL,
          connected_at TIMESTAMPTZ,
          last_heartbeat TIMESTAMPTZ,
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_events (
          id BIGSERIAL PRIMARY KEY,
          type TEXT NOT NULL,
          source TEXT NOT NULL,
          payload JSONB NOT NULL DEFAULT '{}'::jsonb,
          created_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_tasks (
          id TEXT PRIMARY KEY,
          provider_id TEXT NOT NULL,
          command TEXT NOT NULL,
          status TEXT NOT NULL,
          payload JSONB NOT NULL DEFAULT '{}'::jsonb,
          result JSONB,
          created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_scheduled_tasks (
          id TEXT PRIMARY KEY,
          name TEXT NOT NULL,
          description TEXT NOT NULL DEFAULT '',
          enabled BOOLEAN NOT NULL DEFAULT true,
          schedule_json JSONB NOT NULL DEFAULT '{}'::jsonb,
          action_json JSONB NOT NULL DEFAULT '{}'::jsonb,
          concurrency_policy TEXT NOT NULL DEFAULT 'skip',
          next_run_at TIMESTAMPTZ,
          last_run_at TIMESTAMPTZ,
          last_run_id TEXT,
          created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_scheduled_task_runs (
          id TEXT PRIMARY KEY,
          schedule_id TEXT NOT NULL REFERENCES unidesk_scheduled_tasks(id) ON DELETE CASCADE,
          trigger_type TEXT NOT NULL,
          status TEXT NOT NULL,
          task_id TEXT,
          result JSONB,
          error TEXT,
          started_at TIMESTAMPTZ,
          finished_at TIMESTAMPTZ,
          duration_ms BIGINT,
          created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_node_docker_status (
          provider_id TEXT PRIMARY KEY,
          status JSONB NOT NULL DEFAULT '{}'::jsonb,
          collected_at TIMESTAMPTZ,
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_node_system_status (
          provider_id TEXT PRIMARY KEY,
          status JSONB NOT NULL DEFAULT '{}'::jsonb,
          collected_at TIMESTAMPTZ,
          updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE TABLE IF NOT EXISTS unidesk_node_metric_samples (
          id BIGSERIAL PRIMARY KEY,
          provider_id TEXT NOT NULL,
          collected_at TIMESTAMPTZ NOT NULL,
          cpu_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
          memory_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
          disk_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
          sample JSONB NOT NULL DEFAULT '{}'::jsonb,
          created_at TIMESTAMPTZ NOT NULL DEFAULT now()
        );
        CREATE INDEX IF NOT EXISTS idx_unidesk_tasks_updated_at ON unidesk_tasks(updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_unidesk_tasks_status_updated_at ON unidesk_tasks(status, updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_tasks_next_run ON unidesk_scheduled_tasks(enabled, next_run_at);
        CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_task_runs_schedule_updated ON unidesk_scheduled_task_runs(schedule_id, updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_task_runs_status_updated ON unidesk_scheduled_task_runs(status, updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_unidesk_node_system_status_updated_at ON unidesk_node_system_status(updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_unidesk_node_metric_samples_provider_time ON unidesk_node_metric_samples(provider_id, collected_at DESC);
        "#,
    )
    .await
    .context("create backend-core tables")?;
    state.log("info", "database_init_complete", None);
    Ok(())
}

pub async fn record_event(
    state: &Arc<AppState>,
    event_type: &str,
    source: &str,
    payload: JsonValue,
) {
    state.log(
        "info",
        event_type,
        Some(json!({ "source": source, "payload": payload })),
    );
    if !state.db_ready() {
        return;
    }
    let stored_payload = compact_json_for_storage(&payload);
    let result = async {
        let client = state.pool.get().await?;
        client
            .execute(
                "INSERT INTO unidesk_events (type, source, payload) VALUES ($1, $2, $3)",
                &[&event_type, &source, &stored_payload],
            )
            .await?;
        anyhow::Ok(())
    }
    .await;
    if let Err(error) = result {
        state.log(
            "error",
            "event_insert_failed",
            Some(json!({ "type": event_type, "source": source, "error": error.to_string() })),
        );
    }
}

pub async fn upsert_provider_node(
    state: &Arc<AppState>,
    provider_id: &str,
    name: &str,
    labels: &JsonValue,
) -> anyhow::Result<()> {
    let client = state.pool.get().await?;
    client.execute(
        r#"
        INSERT INTO unidesk_nodes (provider_id, name, labels, status, connected_at, last_heartbeat, updated_at)
        VALUES ($1, $2, $3, 'online', now(), now(), now())
        ON CONFLICT (provider_id) DO UPDATE SET
          name = EXCLUDED.name,
          labels = EXCLUDED.labels,
          status = 'online',
          connected_at = COALESCE(unidesk_nodes.connected_at, EXCLUDED.connected_at),
          last_heartbeat = now(),
          updated_at = now()
        "#,
        &[&provider_id, &name, labels],
    ).await?;
    Ok(())
}

pub async fn update_provider_heartbeat(
    state: &Arc<AppState>,
    provider_id: &str,
    labels: &JsonValue,
) -> anyhow::Result<()> {
    let client = state.pool.get().await?;
    client
        .execute(
            r#"
        UPDATE unidesk_nodes
        SET labels = unidesk_nodes.labels || $2::jsonb,
            status = 'online',
            last_heartbeat = now(),
            updated_at = now()
        WHERE provider_id = $1
        "#,
            &[&provider_id, labels],
        )
        .await?;
    Ok(())
}

fn provider_collected_at(
    state: &Arc<AppState>,
    provider_id: &str,
    source: &str,
    value: &str,
) -> DateTime<Utc> {
    match DateTime::parse_from_rfc3339(value) {
        Ok(parsed) => parsed.with_timezone(&Utc),
        Err(error) => {
            state.log(
                "warn",
                "provider_status_collected_at_invalid",
                Some(json!({
                    "providerId": provider_id,
                    "source": source,
                    "value": value,
                    "error": error.to_string()
                })),
            );
            Utc::now()
        }
    }
}

pub async fn upsert_docker_status(
    state: &Arc<AppState>,
    provider_id: &str,
    status: &JsonValue,
    collected_at: &str,
) -> anyhow::Result<()> {
    let collected_at = provider_collected_at(state, provider_id, "docker_status", collected_at);
    let client = state.pool.get().await?;
    client
        .execute(
            r#"
        INSERT INTO unidesk_node_docker_status (provider_id, status, collected_at, updated_at)
        VALUES ($1, $2, $3, now())
        ON CONFLICT (provider_id) DO UPDATE SET
          status = EXCLUDED.status,
          collected_at = EXCLUDED.collected_at,
          updated_at = now()
        "#,
            &[&provider_id, status, &collected_at],
        )
        .await?;
    Ok(())
}

pub async fn upsert_system_status(
    state: &Arc<AppState>,
    provider_id: &str,
    status: &JsonValue,
    collected_at: &str,
) -> anyhow::Result<()> {
    let cpu_percent = nested_number(status, "cpu", "percent");
    let memory_percent = nested_number(status, "memory", "percent");
    let disk_percent = nested_number(status, "disk", "percent");
    let collected_at = provider_collected_at(state, provider_id, "system_status", collected_at);
    let mut client = state.pool.get().await?;
    let tx = client.transaction().await?;
    tx.execute(
        r#"
        INSERT INTO unidesk_node_system_status (provider_id, status, collected_at, updated_at)
        VALUES ($1, $2, $3, now())
        ON CONFLICT (provider_id) DO UPDATE SET
          status = EXCLUDED.status,
          collected_at = EXCLUDED.collected_at,
          updated_at = now()
        "#,
        &[&provider_id, status, &collected_at],
    )
    .await?;
    tx.execute(
        r#"
        INSERT INTO unidesk_node_metric_samples (provider_id, collected_at, cpu_percent, memory_percent, disk_percent, sample)
        VALUES ($1, $2, $3, $4, $5, $6)
        "#,
        &[
            &provider_id,
            &collected_at,
            &cpu_percent,
            &memory_percent,
            &disk_percent,
            status,
        ],
    )
    .await?;
    tx.execute(
        r#"
        DELETE FROM unidesk_node_metric_samples
        WHERE provider_id = $1
          AND id NOT IN (
            SELECT id FROM unidesk_node_metric_samples
            WHERE provider_id = $1
            ORDER BY collected_at DESC
            LIMIT 720
          )
        "#,
        &[&provider_id],
    )
    .await?;
    tx.commit().await?;
    Ok(())
}

pub async fn get_nodes(state: &Arc<AppState>) -> anyhow::Result<Vec<JsonValue>> {
    let client = state.pool.get().await?;
    let rows = client.query(
        "SELECT provider_id, name, labels, status, connected_at, last_heartbeat FROM unidesk_nodes ORDER BY status DESC, provider_id ASC",
        &[],
    ).await?;
    Ok(rows.into_iter().map(|row| {
        let connected_at: Option<DateTime<Utc>> = row.get("connected_at");
        let last_heartbeat: Option<DateTime<Utc>> = row.get("last_heartbeat");
        json!({
            "providerId": row.get::<_, String>("provider_id"),
            "name": row.get::<_, String>("name"),
            "status": if row.get::<_, String>("status") == "online" { "online" } else { "offline" },
            "labels": row.get::<_, JsonValue>("labels"),
            "connectedAt": connected_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
            "lastHeartbeat": last_heartbeat.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
        })
    }).collect())
}

pub async fn get_node_docker_statuses(state: &Arc<AppState>) -> anyhow::Result<Vec<JsonValue>> {
    let client = state.pool.get().await?;
    let rows = client.query(
        r#"
        SELECT n.provider_id, n.name, n.status AS node_status, d.status AS docker_status, d.updated_at
        FROM unidesk_nodes n
        LEFT JOIN unidesk_node_docker_status d ON d.provider_id = n.provider_id
        ORDER BY n.status DESC, n.provider_id ASC
        "#,
        &[],
    ).await?;
    Ok(rows.into_iter().map(|row| {
        let updated_at: Option<DateTime<Utc>> = row.get("updated_at");
        let docker_status: Option<JsonValue> = row.get("docker_status");
        json!({
            "providerId": row.get::<_, String>("provider_id"),
            "name": row.get::<_, String>("name"),
            "nodeStatus": if row.get::<_, String>("node_status") == "online" { "online" } else { "offline" },
            "dockerStatus": docker_status,
            "updatedAt": updated_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
        })
    }).collect())
}

fn metric_point_from_sample(sample: &JsonValue, collected_at: &str) -> JsonValue {
    json!({
        "at": collected_at,
        "cpuPercent": nested_number(sample, "cpu", "percent"),
        "memoryPercent": nested_number(sample, "memory", "percent"),
        "diskPercent": nested_number(sample, "disk", "percent"),
        "memoryUsedBytes": nested_number(sample, "memory", "usedBytes"),
        "memoryTotalBytes": nested_number(sample, "memory", "totalBytes"),
        "diskUsedBytes": nested_number(sample, "disk", "usedBytes"),
        "diskTotalBytes": nested_number(sample, "disk", "totalBytes"),
        "load1": nested_number(sample, "cpu", "load1"),
    })
}

const SYSTEM_STATUS_STALE_AFTER_SECONDS: i64 = 300;

fn system_status_is_stale(collected_at: Option<DateTime<Utc>>, now: DateTime<Utc>) -> bool {
    collected_at
        .map(|value| now.signed_duration_since(value) > Duration::seconds(SYSTEM_STATUS_STALE_AFTER_SECONDS))
        .unwrap_or(false)
}

fn system_status_age_seconds(collected_at: Option<DateTime<Utc>>, now: DateTime<Utc>) -> Option<i64> {
    collected_at.map(|value| now.signed_duration_since(value).num_seconds().max(0))
}

pub async fn get_node_system_statuses(
    state: &Arc<AppState>,
    limit: i64,
) -> anyhow::Result<Vec<JsonValue>> {
    let client = state.pool.get().await?;
    let current_rows = client.query(
        r#"
        SELECT n.provider_id, n.name, n.status AS node_status, s.status AS system_status, s.collected_at AS system_collected_at, s.updated_at
        FROM unidesk_nodes n
        LEFT JOIN unidesk_node_system_status s ON s.provider_id = n.provider_id
        ORDER BY n.status DESC, n.provider_id ASC
        "#,
        &[],
    ).await?;
    let sample_rows = client
        .query(
            r#"
        SELECT n.provider_id, recent.collected_at, recent.sample
        FROM unidesk_nodes n
        LEFT JOIN LATERAL (
          SELECT collected_at, sample
          FROM unidesk_node_metric_samples m
          WHERE m.provider_id = n.provider_id
          ORDER BY collected_at DESC
          LIMIT $1
        ) recent ON true
        WHERE recent.collected_at IS NOT NULL
        ORDER BY n.provider_id ASC, recent.collected_at ASC
        "#,
            &[&limit],
        )
        .await?;
    let mut history_by_provider = std::collections::HashMap::<String, Vec<JsonValue>>::new();
    for row in sample_rows {
        let provider_id: String = row.get("provider_id");
        let collected_at: DateTime<Utc> = row.get("collected_at");
        let sample: JsonValue = row.get("sample");
        history_by_provider
            .entry(provider_id)
            .or_default()
            .push(metric_point_from_sample(
                &sample,
                &collected_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            ));
    }
    let now = Utc::now();
    Ok(current_rows.into_iter().map(|row| {
        let provider_id: String = row.get("provider_id");
        let updated_at: Option<DateTime<Utc>> = row.get("updated_at");
        let system_collected_at: Option<DateTime<Utc>> = row.get("system_collected_at");
        let system_status: Option<JsonValue> = row.get("system_status");
        let stale = system_status_is_stale(system_collected_at, now);
        let current = if stale { None } else { system_status.clone() };
        json!({
            "providerId": provider_id,
            "name": row.get::<_, String>("name"),
            "nodeStatus": if row.get::<_, String>("node_status") == "online" { "online" } else { "offline" },
            "current": current,
            "lastKnown": system_status,
            "currentCollectedAt": system_collected_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
            "stale": stale,
            "staleSeconds": system_status_age_seconds(system_collected_at, now),
            "history": history_by_provider.remove(&provider_id).unwrap_or_default(),
            "updatedAt": updated_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
        })
    }).collect())
}

pub async fn get_events(state: &Arc<AppState>, limit: i64) -> anyhow::Result<Vec<JsonValue>> {
    let client = state.pool.get().await?;
    let rows = client.query(
        "SELECT id, type, source, payload, created_at FROM unidesk_events ORDER BY id DESC LIMIT $1",
        &[&limit],
    ).await?;
    Ok(rows
        .into_iter()
        .map(|row| {
            let created_at: DateTime<Utc> = row.get("created_at");
            let payload: JsonValue = row.get("payload");
            json!({
                "id": row.get::<_, i64>("id"),
                "type": row.get::<_, String>("type"),
                "source": row.get::<_, String>("source"),
                "payload": compact_json(&payload),
                "createdAt": created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            })
        })
        .collect())
}

fn task_json_summary(value: Option<&JsonValue>, prefix: &str) -> JsonValue {
    let Some(value) = value else {
        return if prefix == "result" {
            Value::Null
        } else {
            json!({})
        };
    };
    let Some(object) = value.as_object() else {
        return json!({ "summaryOnly": true, "type": type_name(value) });
    };
    let mut result = serde_json::Map::new();
    result.insert("summaryOnly".to_string(), Value::Bool(true));
    result.insert("type".to_string(), Value::String("object".to_string()));
    let fields: &[&str] = if prefix == "payload" {
        &[
            "source",
            "serviceId",
            "method",
            "path",
            "mode",
            "targetBaseUrl",
            "timeoutMs",
            "targetProviderGatewayVersion",
            "providerGatewayVersion",
        ]
    } else {
        &[
            "error",
            "reason",
            "message",
            "status",
            "exitCode",
            "code",
            "signal",
            "timeoutMs",
            "previousStatus",
            "mode",
            "policy",
            "targetProviderGatewayVersion",
            "providerGatewayVersion",
            "updaterContainerId",
        ]
    };
    for field in fields {
        if let Some(field_value) = object.get(*field) {
            if !(field_value.is_null() || field_value.as_str().is_some_and(|text| text.is_empty()))
            {
                result.insert((*field).to_string(), field_value.clone());
            }
        }
    }
    if let Some(body_text) = object.get("bodyText").and_then(Value::as_str) {
        result.insert(
            "bodyText".to_string(),
            Value::String(format!("<omitted:{} chars>", body_text.len())),
        );
    }
    Value::Object(result)
}

fn type_name(value: &JsonValue) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

fn omit_body_text(value: JsonValue) -> JsonValue {
    match value {
        Value::Object(mut object) => {
            if let Some(Value::String(text)) = object.get("bodyText") {
                object.insert(
                    "bodyText".to_string(),
                    Value::String(format!("<omitted:{} chars>", text.len())),
                );
            }
            Value::Object(object)
        }
        other => other,
    }
}

pub async fn get_task(state: &Arc<AppState>, task_id: &str) -> anyhow::Result<Option<JsonValue>> {
    let client = state.pool.get().await?;
    let row = client.query_opt(
        "SELECT id, provider_id, command, status, payload, result, created_at, updated_at FROM unidesk_tasks WHERE id = $1 LIMIT 1",
        &[&task_id],
    ).await?;
    Ok(row.map(task_full_from_row))
}

pub async fn get_tasks(
    state: &Arc<AppState>,
    limit: i64,
    status_filter: &str,
    lite: bool,
    summary: bool,
) -> anyhow::Result<Vec<JsonValue>> {
    crate::task_dispatcher::maybe_mark_stale_tasks_failed(state, 15_000).await?;
    let client = state.pool.get().await?;
    let rows = if status_filter == "pending" {
        client.query(
            "SELECT id, provider_id, command, status, payload, result, created_at, updated_at FROM unidesk_tasks WHERE status IN ('queued','dispatched','running') ORDER BY updated_at DESC LIMIT $1",
            &[&limit],
        ).await?
    } else {
        client.query(
            "SELECT id, provider_id, command, status, payload, result, created_at, updated_at FROM unidesk_tasks ORDER BY updated_at DESC LIMIT $1",
            &[&limit],
        ).await?
    };
    Ok(rows.into_iter().map(|row| {
        if summary && !lite {
            task_summary_from_row(&row)
        } else {
            let created_at: DateTime<Utc> = row.get("created_at");
            let updated_at: DateTime<Utc> = row.get("updated_at");
            json!({
                "id": row.get::<_, String>("id"),
                "providerId": row.get::<_, String>("provider_id"),
                "command": row.get::<_, String>("command"),
                "status": row.get::<_, String>("status"),
                "payload": if lite { json!({}) } else { compact_json(&omit_body_text(row.get::<_, JsonValue>("payload"))) },
                "result": if lite { Value::Null } else { row.get::<_, Option<JsonValue>>("result").map(|value| compact_json(&omit_body_text(value))).unwrap_or(Value::Null) },
                "createdAt": created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
                "updatedAt": updated_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            })
        }
    }).collect())
}

fn task_full_from_row(row: Row) -> JsonValue {
    let created_at: DateTime<Utc> = row.get("created_at");
    let updated_at: DateTime<Utc> = row.get("updated_at");
    json!({
        "id": row.get::<_, String>("id"),
        "providerId": row.get::<_, String>("provider_id"),
        "command": row.get::<_, String>("command"),
        "status": row.get::<_, String>("status"),
        "payload": compact_json(&omit_body_text(row.get::<_, JsonValue>("payload"))),
        "result": row.get::<_, Option<JsonValue>>("result").map(|value| compact_json(&omit_body_text(value))).unwrap_or(Value::Null),
        "createdAt": created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
        "updatedAt": updated_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
    })
}

fn task_summary_from_row(row: &Row) -> JsonValue {
    let created_at: DateTime<Utc> = row.get("created_at");
    let updated_at: DateTime<Utc> = row.get("updated_at");
    let payload: JsonValue = row.get("payload");
    let result: Option<JsonValue> = row.get("result");
    json!({
        "id": row.get::<_, String>("id"),
        "providerId": row.get::<_, String>("provider_id"),
        "command": row.get::<_, String>("command"),
        "status": row.get::<_, String>("status"),
        "payload": task_json_summary(Some(&payload), "payload"),
        "result": task_json_summary(result.as_ref(), "result"),
        "createdAt": created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
        "updatedAt": updated_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
        "_summaryOnly": true,
    })
}

pub async fn raw_task(state: &Arc<AppState>, task_id: &str) -> anyhow::Result<Option<RawTaskRow>> {
    let client = state.pool.get().await?;
    let row = client.query_opt(
        "SELECT id, provider_id, command, status, payload, result, updated_at FROM unidesk_tasks WHERE id = $1 LIMIT 1",
        &[&task_id],
    ).await?;
    Ok(row.map(|row| RawTaskRow {
        id: row.get("id"),
        provider_id: row.get("provider_id"),
        command: row.get("command"),
        status: row.get("status"),
        payload: row.get("payload"),
        result: row.get("result"),
        updated_at: row.get("updated_at"),
    }))
}

pub async fn provider_supports(
    state: &Arc<AppState>,
    provider_id: &str,
    capability: &str,
) -> anyhow::Result<bool> {
    if !state.db_ready() {
        return Ok(false);
    }
    let client = state.pool.get().await?;
    let row = client
        .query_opt(
            "SELECT labels FROM unidesk_nodes WHERE provider_id = $1 LIMIT 1",
            &[&provider_id],
        )
        .await?;
    let Some(row) = row else {
        return Ok(false);
    };
    let labels: JsonValue = row.get("labels");
    Ok(labels
        .get("unideskCapabilities")
        .and_then(Value::as_array)
        .is_some_and(|items| items.iter().any(|item| item.as_str() == Some(capability))))
}
