use std::collections::HashMap;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sqlx::{PgPool, Row};

use crate::{ControlPlaneError, ControlPlaneSchemaService, DaemonId};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonRuntimeHeartbeat {
    pub daemon_id: DaemonId,
    pub runtime_kind: String,
    pub host: String,
    pub version: String,
    pub workers: Vec<String>,
    pub metadata: Value,
    pub started_at: DateTime<Utc>,
}

impl DaemonRuntimeHeartbeat {
    pub fn new(
        daemon_id: DaemonId,
        runtime_kind: impl Into<String>,
        host: impl Into<String>,
        version: impl Into<String>,
        workers: Vec<String>,
        metadata: Value,
        started_at: DateTime<Utc>,
    ) -> Result<Self, ControlPlaneError> {
        let runtime_kind = runtime_kind.into().trim().to_string();
        if runtime_kind.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "runtime_kind must not be empty".to_string(),
            ));
        }

        let host = host.into().trim().to_string();
        if host.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "host must not be empty".to_string(),
            ));
        }

        let version = version.into().trim().to_string();
        if version.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "version must not be empty".to_string(),
            ));
        }

        let mut workers = workers
            .into_iter()
            .map(|worker| worker.trim().to_string())
            .filter(|worker| !worker.is_empty())
            .collect::<Vec<_>>();
        workers.sort();
        workers.dedup();

        Ok(Self {
            daemon_id,
            runtime_kind,
            host,
            version,
            workers,
            metadata,
            started_at,
        })
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonInventoryRecord {
    pub daemon_id: String,
    pub runtime_kind: String,
    pub host: String,
    pub version: String,
    pub workers: Vec<String>,
    pub started_at: DateTime<Utc>,
    pub last_seen_at: DateTime<Utc>,
    pub status: String,
    pub active_backup_jobs: i64,
    pub active_typesense_jobs: i64,
    pub active_clone_jobs: i64,
    pub pinned_backup_schedules: i64,
    pub pinned_typesense_jobs: i64,
    pub pinned_clone_jobs: i64,
}

pub struct DaemonRegistryService {
    pool: PgPool,
}

impl DaemonRegistryService {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    pub async fn upsert_heartbeat(
        &self,
        heartbeat: &DaemonRuntimeHeartbeat,
    ) -> Result<(), ControlPlaneError> {
        ControlPlaneSchemaService::ensure_daemon_registry_table(&self.pool).await?;
        sqlx::query(
            r#"
            INSERT INTO public.daemon_runtime_registry (
                daemon_id,
                runtime_kind,
                host,
                version,
                workers_json,
                metadata_json,
                started_at,
                last_seen_at
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7, now())
            ON CONFLICT (daemon_id)
            DO UPDATE
            SET
                runtime_kind = EXCLUDED.runtime_kind,
                host = EXCLUDED.host,
                version = EXCLUDED.version,
                workers_json = EXCLUDED.workers_json,
                metadata_json = EXCLUDED.metadata_json,
                started_at = EXCLUDED.started_at,
                last_seen_at = now()
            "#,
        )
        .bind(heartbeat.daemon_id.as_str())
        .bind(&heartbeat.runtime_kind)
        .bind(&heartbeat.host)
        .bind(&heartbeat.version)
        .bind(json!(heartbeat.workers))
        .bind(&heartbeat.metadata)
        .bind(heartbeat.started_at)
        .execute(&self.pool)
        .await
        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;

        Ok(())
    }

    pub async fn list_inventory(&self) -> Result<Vec<DaemonInventoryRecord>, ControlPlaneError> {
        ControlPlaneSchemaService::ensure_daemon_registry_table(&self.pool).await?;

        let rows = sqlx::query(
            r#"
            SELECT
                daemon_id,
                runtime_kind,
                host,
                version,
                workers_json,
                started_at,
                last_seen_at
            FROM public.daemon_runtime_registry
            ORDER BY last_seen_at DESC, daemon_id ASC
            "#,
        )
        .fetch_all(&self.pool)
        .await
        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;

        let active_backup_jobs = self
            .fetch_count_map_if_present(
                "public.backup_jobs",
                r#"
                SELECT execution_daemon_id, COUNT(*)::bigint AS count
                FROM public.backup_jobs
                WHERE status = 'running'
                  AND NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NOT NULL
                  AND (lease_expires_at IS NULL OR lease_expires_at >= now())
                GROUP BY execution_daemon_id
                "#,
            )
            .await?;
        let active_typesense_jobs = self
            .fetch_count_map_if_present(
                "public.typesense_sync_runs",
                r#"
                SELECT j.execution_daemon_id, COUNT(DISTINCT r.job_id)::bigint AS count
                FROM public.typesense_sync_runs AS r
                JOIN public.typesense_sync_jobs AS j
                  ON j.id = r.job_id
                WHERE r.status = 'running'
                  AND NULLIF(BTRIM(COALESCE(j.execution_daemon_id, '')), '') IS NOT NULL
                GROUP BY j.execution_daemon_id
                "#,
            )
            .await?;
        let active_clone_jobs = self
            .fetch_count_map_if_present(
                "public.database_clone_jobs",
                r#"
                SELECT execution_daemon_id, COUNT(*)::bigint AS count
                FROM public.database_clone_jobs
                WHERE NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NOT NULL
                  AND status IN ('preparing', 'running', 'verifying', 'cancel_requested')
                  AND (lease_expires_at IS NULL OR lease_expires_at >= now())
                GROUP BY execution_daemon_id
                "#,
            )
            .await?;
        let pinned_backup_schedules = self
            .fetch_count_map_if_present(
                "public.backup_schedules",
                r#"
                SELECT execution_daemon_id, COUNT(*)::bigint AS count
                FROM public.backup_schedules
                WHERE enabled = true
                  AND NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NOT NULL
                GROUP BY execution_daemon_id
                "#,
            )
            .await?;
        let pinned_typesense_jobs = self
            .fetch_count_map_if_present(
                "public.typesense_sync_jobs",
                r#"
                SELECT execution_daemon_id, COUNT(*)::bigint AS count
                FROM public.typesense_sync_jobs
                WHERE NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NOT NULL
                GROUP BY execution_daemon_id
                "#,
            )
            .await?;
        let pinned_clone_jobs = self
            .fetch_count_map_if_present(
                "public.database_clone_jobs",
                r#"
                SELECT execution_daemon_id, COUNT(*)::bigint AS count
                FROM public.database_clone_jobs
                WHERE NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NOT NULL
                  AND status NOT IN ('completed', 'failed_retryable', 'failed_terminal', 'cancelled')
                GROUP BY execution_daemon_id
                "#,
            )
            .await?;

        rows.into_iter()
            .map(|row| {
                let daemon_id = row
                    .try_get::<String, _>("daemon_id")
                    .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
                let last_seen_at = row
                    .try_get::<DateTime<Utc>, _>("last_seen_at")
                    .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
                let workers = row
                    .try_get::<Value, _>("workers_json")
                    .map_err(|err| ControlPlaneError::Storage(err.to_string()))
                    .and_then(|value| {
                        serde_json::from_value::<Vec<String>>(value)
                            .map_err(|err| ControlPlaneError::Storage(err.to_string()))
                    })?;
                let status = if last_seen_at >= Utc::now() - chrono::Duration::seconds(60) {
                    "online".to_string()
                } else {
                    "offline".to_string()
                };

                Ok(DaemonInventoryRecord {
                    runtime_kind: row
                        .try_get("runtime_kind")
                        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?,
                    host: row
                        .try_get("host")
                        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?,
                    version: row
                        .try_get("version")
                        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?,
                    started_at: row
                        .try_get("started_at")
                        .map_err(|err| ControlPlaneError::Storage(err.to_string()))?,
                    daemon_id: daemon_id.clone(),
                    workers,
                    last_seen_at,
                    status,
                    active_backup_jobs: active_backup_jobs.get(&daemon_id).copied().unwrap_or(0),
                    active_typesense_jobs: active_typesense_jobs
                        .get(&daemon_id)
                        .copied()
                        .unwrap_or(0),
                    active_clone_jobs: active_clone_jobs.get(&daemon_id).copied().unwrap_or(0),
                    pinned_backup_schedules: pinned_backup_schedules
                        .get(&daemon_id)
                        .copied()
                        .unwrap_or(0),
                    pinned_typesense_jobs: pinned_typesense_jobs
                        .get(&daemon_id)
                        .copied()
                        .unwrap_or(0),
                    pinned_clone_jobs: pinned_clone_jobs.get(&daemon_id).copied().unwrap_or(0),
                })
            })
            .collect()
    }

    async fn fetch_count_map_if_present(
        &self,
        relation_name: &str,
        sql: &str,
    ) -> Result<HashMap<String, i64>, ControlPlaneError> {
        if !self.relation_exists(relation_name).await? {
            return Ok(HashMap::new());
        }

        let rows = sqlx::query(sql)
            .fetch_all(&self.pool)
            .await
            .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
        let mut counts = HashMap::new();
        for row in rows {
            let daemon_id = row
                .try_get::<String, _>("execution_daemon_id")
                .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
            let count = row
                .try_get::<i64, _>("count")
                .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
            counts.insert(daemon_id, count);
        }
        Ok(counts)
    }

    async fn relation_exists(&self, relation_name: &str) -> Result<bool, ControlPlaneError> {
        let relation = sqlx::query_scalar::<_, Option<String>>("SELECT to_regclass($1)::text")
            .bind(relation_name)
            .fetch_one(&self.pool)
            .await
            .map_err(|err| ControlPlaneError::Storage(err.to_string()))?;
        Ok(relation.is_some())
    }
}
