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

use anyhow::bail;
use axum::body::Bytes;
use rand::Rng;
use serde_json::{json, Value};

use crate::db::{provider_supports, raw_task, record_event};
use crate::json_util::{compact_json_for_storage, now_iso};
use crate::state::AppState;
use crate::types::{JsonValue, RawTaskRow};

pub fn is_terminal_task_status(status: &str) -> bool {
    status == "succeeded" || status == "failed"
}

pub fn is_provider_dispatch_command(command: &str) -> bool {
    matches!(
        command,
        "docker.ps" | "provider.upgrade" | "host.ssh" | "microservice.http" | "echo"
    )
}

pub async fn notify_task_terminal(state: &Arc<AppState>, task_id: &str) -> anyhow::Result<()> {
    let waiters = {
        let mut waiters_by_task = state.task_terminal_waiters.lock().await;
        waiters_by_task.remove(task_id)
    };
    let Some(waiters) = waiters else {
        return Ok(());
    };
    let task = raw_task(state, task_id).await?;
    if task
        .as_ref()
        .is_none_or(|task| !is_terminal_task_status(&task.status))
    {
        return Ok(());
    }
    for waiter in waiters {
        let _ = waiter.send(task.clone());
    }
    Ok(())
}

pub async fn wait_for_task_terminal(
    state: &Arc<AppState>,
    task_id: &str,
    timeout_ms: u64,
) -> anyhow::Result<Option<RawTaskRow>> {
    let latest = raw_task(state, task_id).await?;
    if latest
        .as_ref()
        .is_some_and(|task| is_terminal_task_status(&task.status))
    {
        return Ok(latest);
    }
    let (tx, rx) = tokio::sync::oneshot::channel();
    {
        let mut waiters = state.task_terminal_waiters.lock().await;
        waiters.entry(task_id.to_string()).or_default().push(tx);
    }
    match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), rx).await {
        Ok(Ok(task)) => Ok(task),
        _ => {
            let task = raw_task(state, task_id).await?;
            Ok(task.or(latest))
        }
    }
}

pub async fn create_and_send_task(
    state: &Arc<AppState>,
    provider_id: &str,
    command: &str,
    payload: JsonValue,
) -> anyhow::Result<(String, bool)> {
    if !is_provider_dispatch_command(command) {
        bail!("unsupported dispatch command: {command}");
    }
    let suffix: u64 = rand::thread_rng().gen();
    let task_id = format!(
        "task_{}_{}",
        chrono::Utc::now().timestamp_millis(),
        format!("{suffix:x}")
    );
    let client = state.pool.get().await?;
    client.execute(
        "INSERT INTO unidesk_tasks (id, provider_id, command, status, payload, result) VALUES ($1, $2, $3, 'queued', $4, NULL)",
        &[&task_id, &provider_id, &command, &payload],
    ).await?;
    let dispatch =
        json!({ "type": "dispatch", "taskId": task_id, "command": command, "payload": payload });
    let provider_online = state.send_provider(provider_id, dispatch).await;
    if !provider_online {
        record_event(
            state,
            "task_queued_provider_offline",
            provider_id,
            json!({ "taskId": task_id, "providerId": provider_id, "command": command }),
        )
        .await;
        return Ok((task_id, false));
    }
    client.execute("UPDATE unidesk_tasks SET status = 'dispatched', updated_at = now() WHERE id = $1 AND status = 'queued'", &[&task_id]).await?;
    record_event(
        state,
        "task_dispatched",
        provider_id,
        json!({ "taskId": task_id, "providerId": provider_id, "command": command }),
    )
    .await;
    Ok((task_id, true))
}

pub async fn mark_stale_tasks_failed(state: &Arc<AppState>) -> anyhow::Result<()> {
    if !state.db_ready() {
        return Ok(());
    }
    let timeout_ms = state.config.task_pending_timeout_ms as i64;
    let client = state.pool.get().await?;
    let rows = client
        .query(
            r#"
        WITH stale AS (
          SELECT id, provider_id, command, status AS previous_status, updated_at
          FROM unidesk_tasks
          WHERE status IN ('queued', 'dispatched', 'running')
            AND updated_at < now() - ($1::bigint * interval '1 millisecond')
          FOR UPDATE
        ),
        updated AS (
          UPDATE unidesk_tasks task
          SET
            status = 'failed',
            result = jsonb_build_object(
              'error', 'task timed out without terminal provider status',
              'timeoutMs', $1::bigint,
              'previousStatus', stale.previous_status,
              'previousResult', task.result,
              'timedOutAt', now()
            ),
            updated_at = now()
          FROM stale
          WHERE task.id = stale.id
          RETURNING task.id, task.provider_id, task.command, stale.previous_status, stale.updated_at
        )
        SELECT * FROM updated
        "#,
            &[&timeout_ms],
        )
        .await?;
    for row in rows {
        let task_id: String = row.get("id");
        let provider_id: String = row.get("provider_id");
        let previous_status: String = row.get("previous_status");
        let updated_at: chrono::DateTime<chrono::Utc> = row.get("updated_at");
        record_event(state, "task_timeout", &provider_id, json!({
            "taskId": task_id,
            "providerId": provider_id,
            "command": row.get::<_, String>("command"),
            "previousStatus": previous_status,
            "previousUpdatedAt": updated_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            "timeoutMs": timeout_ms,
        })).await;
        if let Err(error) = notify_task_terminal(state, &task_id).await {
            state.log(
                "error",
                "task_waiter_notify_failed",
                Some(json!({ "taskId": task_id, "error": error.to_string() })),
            );
        }
    }
    Ok(())
}

pub async fn maybe_mark_stale_tasks_failed(
    state: &Arc<AppState>,
    _min_interval_ms: u64,
) -> anyhow::Result<()> {
    // The Rust implementation keeps this simple; the periodic sweeper handles
    // the steady-state load, and list calls may also opportunistically mark.
    mark_stale_tasks_failed(state).await
}

pub async fn dispatch_task(state: &Arc<AppState>, body: Bytes) -> crate::http::HttpResponse {
    let parsed: Value = serde_json::from_slice(&body).unwrap_or_else(|_| json!({}));
    let provider_id = parsed
        .get("providerId")
        .and_then(Value::as_str)
        .unwrap_or("");
    let command = parsed.get("command").and_then(Value::as_str).unwrap_or("");
    let payload = parsed
        .get("payload")
        .filter(|value| value.is_object())
        .cloned()
        .unwrap_or_else(|| json!({}));
    if provider_id.is_empty() {
        return crate::http::json_response(
            json!({ "ok": false, "error": "providerId is required" }),
            400,
        );
    }
    if !is_provider_dispatch_command(command) {
        return crate::http::json_response(
            json!({ "ok": false, "error": "command must be one of docker.ps, provider.upgrade, host.ssh, microservice.http, echo" }),
            400,
        );
    }
    if command == "host.ssh"
        && !provider_supports(state, provider_id, "host.ssh")
            .await
            .unwrap_or(false)
    {
        return crate::http::json_response(
            json!({ "ok": false, "error": format!("provider does not declare host.ssh capability: {provider_id}") }),
            409,
        );
    }
    if command == "microservice.http"
        && !provider_supports(state, provider_id, "microservice.http")
            .await
            .unwrap_or(false)
    {
        return crate::http::json_response(
            json!({ "ok": false, "error": format!("provider does not declare microservice.http capability: {provider_id}") }),
            409,
        );
    }
    match create_and_send_task(
        state,
        provider_id,
        command,
        compact_json_for_storage(&payload),
    )
    .await
    {
        Ok((task_id, provider_online)) => crate::http::json_response(
            json!({ "ok": true, "taskId": task_id, "status": if provider_online { "dispatched" } else { "queued" }, "providerOnline": provider_online }),
            200,
        ),
        Err(error) => crate::http::json_response(
            json!({ "ok": false, "error": error.to_string(), "at": now_iso() }),
            500,
        ),
    }
}
