use std::collections::BTreeMap;

use axum::extract::FromRequestParts;
use axum::http::header;
use ci_protocol::runner::{LeaseDto, RegisterResponse, StepPayload};
use models::ci::runner::{self, RunnerState};
use models::ci::runner_token;
use models::ci::workflow_job;
use models::ci::workflow_run;
use models::ci::workflow_step;
use sea_orm::{ActiveModelTrait, ColumnTrait, Condition, EntityTrait, QueryFilter, QueryOrder, Set};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use uuid::Uuid;

use crate::errors::{CiError, CiResult};
use crate::secrets::{constant_time_eq, random_secret};
use crate::service::CiService;

#[derive(Debug, Clone)]
pub struct RunnerAuth {
    pub runner_id: Uuid,
}

pub fn hash_token(secret: &str, salt: &str) -> String {
    let digest = Sha256::digest(format!("{secret}{salt}").as_bytes());
    let mut out = String::with_capacity(digest.len() * 2);
    for byte in digest {
        out.push_str(&format!("{byte:02x}"));
    }
    out
}

pub fn generate_credential(runner_id: Uuid) -> (String, String, String, String) {
    let secret = random_secret(32);
    let salt = random_secret(16);
    let hash = hash_token(&secret, &salt);
    let last_eight: String = secret
        .chars()
        .rev()
        .take(8)
        .collect::<String>()
        .chars()
        .rev()
        .collect();
    let token = format!("{runner_id}.{secret}");
    (token, hash, salt, last_eight)
}

pub async fn verify_runner_token(service: &CiService, token: &str) -> CiResult<Uuid> {
    let Some((id, secret)) = token.split_once('.') else {
        return Err(CiError::Unauthorized);
    };
    let Ok(runner_id) = Uuid::parse_str(id) else {
        return Err(CiError::Unauthorized);
    };
    let row = runner_token::Entity::find()
        .filter(runner_token::Column::RunnerId.eq(runner_id))
        .one(service.db_read())
        .await?;
    let Some(row) = row else {
        return Err(CiError::Unauthorized);
    };
    if row.revoked_at.is_some() {
        return Err(CiError::Unauthorized);
    }
    if let Some(expires_at) = row.expires_at
        && expires_at <= OffsetDateTime::now_utc()
    {
        return Err(CiError::Unauthorized);
    }
    let computed = hash_token(secret, &row.token_salt);
    if !constant_time_eq(computed.as_bytes(), row.token_hash.as_bytes()) {
        return Err(CiError::Unauthorized);
    }
    let runner = runner::Entity::find_by_id(runner_id)
        .one(service.db_read())
        .await?
        .ok_or(CiError::Unauthorized)?;
    if runner.state != RunnerState::Active {
        return Err(CiError::Forbidden);
    }
    Ok(runner_id)
}

impl FromRequestParts<CiService> for RunnerAuth {
    type Rejection = crate::errors::ApiError;

    async fn from_request_parts(
        parts: &mut axum::http::request::Parts,
        state: &CiService,
    ) -> Result<Self, Self::Rejection> {
        let Some(value) = parts.headers.get(header::AUTHORIZATION) else {
            return Err(crate::errors::ApiError::unauthorized());
        };
        let raw = value
            .to_str()
            .map_err(|_| crate::errors::ApiError::bad_request("malformed authorization header"))?;
        let Some(token) = raw.strip_prefix("Bearer ") else {
            return Err(crate::errors::ApiError::unauthorized());
        };
        let runner_id = verify_runner_token(state, token)
            .await
            .map_err(crate::errors::ApiError::from)?;
        Ok(Self { runner_id })
    }
}

impl CiService {
    pub async fn register_runner(
        &self,
        token: &str,
        name: &str,
        labels: Vec<String>,
        capacity: i32,
        version: Option<String>,
    ) -> CiResult<RegisterResponse> {
        self.register_runner_inner(token, name, labels, capacity, version)
            .await
    }

    pub async fn configure_runner_protocol(
        &self,
        runner_id: Uuid,
        versions: &[u32],
        capabilities: &[String],
    ) -> CiResult<u32> {
        let version = if versions.contains(&2) {
            2
        } else if versions.is_empty() || versions.contains(&1) {
            1
        } else {
            return Err(CiError::InvalidArgument(
                "no common runner protocol version".into(),
            ));
        };
        let row = runner::Entity::find_by_id(runner_id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        let mut active: runner::ActiveModel = row.into();
        active.protocol_version = Set(Some(version as i32));
        active.capabilities = Set(Some(serde_json::to_value(capabilities)?));
        active.update(self.db_write()).await?;
        Ok(version)
    }

    async fn register_runner_inner(
        &self,
        token: &str,
        name: &str,
        labels: Vec<String>,
        capacity: i32,
        version: Option<String>,
    ) -> CiResult<RegisterResponse> {
        self.ensure_enabled()?;
        if name.trim().is_empty() || name.len() > 200 {
            return Err(CiError::InvalidArgument(
                "runner name must be 1-200 characters".to_string(),
            ));
        }
        let mut labels: Vec<String> = labels
            .into_iter()
            .map(|l| l.trim().to_string())
            .filter(|l| !l.is_empty())
            .collect();
        labels.sort();
        labels.dedup();
        if labels.is_empty() {
            return Err(CiError::InvalidArgument(
                "runner must declare at least one label".to_string(),
            ));
        }
        if capacity <= 0 || capacity > 256 {
            return Err(CiError::InvalidArgument(
                "runner capacity must be between 1 and 256".to_string(),
            ));
        }

        let (registration_token, space_scope) = self.verify_registration_token(token).await?;

        let now = OffsetDateTime::now_utc();
        let runner_id = Uuid::now_v7();
        runner::ActiveModel {
            id: Set(runner_id),
            name: Set(name.trim().to_string()),
            labels: Set(serde_json::to_value(&labels)?),
            capacity: Set(capacity),
            version: Set(version),
            state: Set(RunnerState::Active),
            space_id: Set(space_scope),
            last_seen_at: Set(Some(now)),
            created_at: Set(now),
            updated_at: Set(now),
            ..Default::default()
        }
        .insert(self.db_write())
        .await?;

        let (credential, hash, salt, last_eight) = generate_credential(runner_id);
        runner_token::ActiveModel {
            id: Set(Uuid::now_v7()),
            runner_id: Set(Some(runner_id)),
            name: Set(format!("runner:{name}")),
            token_hash: Set(hash),
            token_salt: Set(salt),
            token_last_eight: Set(last_eight),
            space_id: Set(space_scope),
            single_use: Set(false),
            used_at: Set(None),
            revoked_at: Set(None),
            expires_at: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
        }
        .insert(self.db_write())
        .await?;

        if let Some(registration_token_id) = registration_token {
            let mut am: runner_token::ActiveModel = registration_token_id.into();
            am.used_at = Set(Some(now));
            am.updated_at = Set(now);
            let _ = am.update(self.db_write()).await;
        }

        tracing::info!(runner_id = %runner_id, name, "runner registered");
        Ok(RegisterResponse {
            runner_id,
            credential,
            heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,
            poll_timeout_secs: self.config().ci.long_poll_timeout_secs,
            protocol_version: 1,
        })
    }

    async fn verify_registration_token(
        &self,
        token: &str,
    ) -> CiResult<(Option<runner_token::Model>, Option<Uuid>)> {
        let Some((id, secret)) = token.split_once('.') else {
            return Err(CiError::Unauthorized);
        };
        let Ok(token_id) = Uuid::parse_str(id) else {
            return Err(CiError::Unauthorized);
        };
        let row = runner_token::Entity::find_by_id(token_id)
            .one(self.db_read())
            .await?
            .ok_or(CiError::Unauthorized)?;
        if row.runner_id.is_some() || row.revoked_at.is_some() {
            return Err(CiError::Unauthorized);
        }
        if row.single_use && row.used_at.is_some() {
            return Err(CiError::Unauthorized);
        }
        if let Some(expires_at) = row.expires_at
            && expires_at <= OffsetDateTime::now_utc()
        {
            return Err(CiError::Unauthorized);
        }
        let computed = hash_token(secret, &row.token_salt);
        if !constant_time_eq(computed.as_bytes(), row.token_hash.as_bytes()) {
            return Err(CiError::Unauthorized);
        }
        Ok((Some(row.clone()), row.space_id))
    }

    pub async fn runner_heartbeat(&self, runner_id: Uuid) -> CiResult<Vec<Uuid>> {
        let now = OffsetDateTime::now_utc();
        runner::Entity::update_many()
            .col_expr(
                runner::Column::LastSeenAt,
                sea_orm::sea_query::Expr::value(Some(now)),
            )
            .col_expr(
                runner::Column::UpdatedAt,
                sea_orm::sea_query::Expr::value(now),
            )
            .filter(runner::Column::Id.eq(runner_id))
            .exec(self.db_write())
            .await?;
        self.cancel_list_for_runner(runner_id).await
    }

    /// Disables runners that stopped heartbeating and revokes their credentials.
    pub async fn reap_stale_runners(&self) -> CiResult<u64> {
        let now = OffsetDateTime::now_utc();
        let stale_secs = self.config().ci.runner_stale_secs.max(1) as i64;
        let cutoff = now - time::Duration::seconds(stale_secs);
        let stale = runner::Entity::find()
            .filter(runner::Column::State.eq(RunnerState::Active))
            .filter(
                Condition::any()
                    .add(runner::Column::LastSeenAt.lt(cutoff))
                    .add(runner::Column::LastSeenAt.is_null()),
            )
            .all(self.db_write())
            .await?;
        let mut ids = Vec::with_capacity(stale.len());
        for row in stale {
            let id = row.id;
            let mut am: runner::ActiveModel = row.into();
            am.state = Set(RunnerState::Disabled);
            am.updated_at = Set(now);
            am.update(self.db_write()).await?;
            ids.push(id);
        }
        if ids.is_empty() {
            return Ok(0);
        }
        runner_token::Entity::update_many()
            .col_expr(
                runner_token::Column::RevokedAt,
                sea_orm::sea_query::Expr::value(Some(now)),
            )
            .col_expr(
                runner_token::Column::UpdatedAt,
                sea_orm::sea_query::Expr::value(now),
            )
            .filter(runner_token::Column::RunnerId.is_in(ids.clone()))
            .filter(runner_token::Column::RevokedAt.is_null())
            .exec(self.db_write())
            .await?;
        tracing::info!(count = ids.len(), "stale runners disabled");
        Ok(ids.len() as u64)
    }

    pub async fn runner_labels(&self, runner_id: Uuid) -> CiResult<Vec<String>> {
        let runner = runner::Entity::find_by_id(runner_id)
            .one(self.db_read())
            .await?
            .ok_or(CiError::NotFound)?;
        Ok(serde_json::from_value(runner.labels).unwrap_or_default())
    }

    /// Builds the full claim payload for a job that this runner has leased.
    pub async fn build_job_payload(
        &self,
        job_id: Uuid,
        _runner_id: Uuid,
        lease_token: String,
    ) -> CiResult<ci_protocol::runner::JobPayload> {
        let job = workflow_job::Entity::find_by_id(job_id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        let run = workflow_run::Entity::find_by_id(job.run_id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        if job.actions_context.is_some() {
            return self
                .build_actions_job_payload(&job, &run, lease_token)
                .await;
        }
        let repo = self.repo_by_id(job.repo_id).await?;
        let space = self.space_of_repo(&repo).await?;
        let definition: serde_json::Value = job.definition.clone();

        let steps = workflow_step::Entity::find()
            .filter(workflow_step::Column::JobId.eq(job_id))
            .order_by_asc(workflow_step::Column::StepIndex)
            .all(self.db_read())
            .await?;
        let step_env: BTreeMap<String, String> = definition
            .get("env")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default();
        let mut step_payloads = Vec::with_capacity(steps.len());
        for step in steps {
            step_payloads.push(StepPayload {
                index: step.step_index,
                name: step.name.clone(),
                run: step.script.clone(),
                shell: step.shell.clone(),
                working_directory: step.working_directory.clone(),
                env: step_env.clone(),
                timeout_secs: definition
                    .get("timeout-minutes")
                    .and_then(|v| v.as_u64())
                    .map(|m| m * 60),
            });
        }

        let workflow_env: BTreeMap<String, String> = run
            .definition
            .get("env")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default();
        let mut env = BTreeMap::new();
        env.insert("GITDATA_RUN_ID".to_string(), run.id.to_string());
        env.insert("GITDATA_JOB_ID".to_string(), job.id.to_string());
        env.insert("GITDATA_REPO_ID".to_string(), repo.id.to_string());
        env.insert("GITDATA_SPACE".to_string(), space.lower_name.clone());
        env.insert("GITDATA_REPO".to_string(), repo.name.clone());
        env.insert("GITDATA_SHA".to_string(), run.commit_sha.clone());
        env.insert("GITDATA_REF".to_string(), run.git_ref.clone());
        env.insert("GITDATA_EVENT".to_string(), run.event.as_str().to_string());
        env.insert("GITDATA_WORKFLOW".to_string(), run.workflow_name.clone());
        env.insert("CI".to_string(), "true".to_string());
        env.insert("GITDATA_CI".to_string(), "true".to_string());
        if let Some(branch) = &run.branch {
            env.insert("GITDATA_BRANCH".to_string(), branch.clone());
        }
        if let Some(tag) = &run.tag {
            env.insert("GITDATA_TAG".to_string(), tag.clone());
        }
        if let Some(base_sha) = &run.base_sha {
            env.insert("GITDATA_BASE_SHA".to_string(), base_sha.clone());
        }
        if let Some(base_ref) = &run.base_ref {
            env.insert("GITDATA_BASE_REF".to_string(), base_ref.clone());
        }
        if let Some(number) = run.pr_number {
            env.insert("GITDATA_PR_NUMBER".to_string(), number.to_string());
        }
        if let Some(environment) = &job.environment {
            env.insert("GITDATA_ENVIRONMENT".to_string(), environment.clone());
        }
        for (key, value) in workflow_env {
            env.insert(key, value);
        }

        let secrets: BTreeMap<String, String> = self
            .resolved_secrets(repo.id, space.id, job.environment.as_deref())
            .await?
            .into_iter()
            .collect();

        let contents_write = definition
            .get("permissions")
            .and_then(|p| p.get("contents"))
            .and_then(|c| c.as_str())
            .is_some_and(|c| c == "write");

        let artifacts: Vec<ci_protocol::runner::ArtifactSpecDto> = definition
            .get("artifacts")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default();

        let needs_artifacts = self.needs_artifacts(run.id, job.id).await?;
        let artifacts: Vec<ci_protocol::runner::ArtifactSpecDto> = artifacts
            .into_iter()
            .map(|spec| ci_protocol::runner::ArtifactSpecDto {
                name: spec.name,
                path: spec.path,
                always: spec.always,
            })
            .collect();

        let clone_url = format!(
            "{}/{}/{}.git",
            self.config().git.base_url.trim_end_matches('/'),
            space.lower_name,
            repo.name
        );

        let payload = ci_protocol::runner::JobPayload {
            job_id: job.id,
            run_id: run.id,
            repo_id: repo.id,
            space_id: space.id,
            run_attempt: job.attempt,
            job_name: job.name.clone(),
            clone_url,
            commit_sha: run.commit_sha.clone(),
            git_ref: run.git_ref.clone(),
            branch: run.branch.clone(),
            tag: run.tag.clone(),
            base_sha: run.base_sha.clone(),
            base_ref: run.base_ref.clone(),
            pr_number: run.pr_number,
            event: run.event.as_str().to_string(),
            workflow_name: run.workflow_name.clone(),
            workflow_file: run.workflow_file.clone(),
            env,
            secrets,
            allow_git_token: contents_write,
            container: definition
                .get("container")
                .and_then(|v| v.as_str())
                .map(|s| s.to_string()),
            timeout_secs: definition
                .get("timeout-minutes")
                .and_then(|v| v.as_u64())
                .map(|m| m * 60)
                .unwrap_or(self.config().ci.job_timeout_default_secs)
                .min(self.config().ci.job_timeout_max_secs),
            steps: step_payloads,
            artifacts,
            needs_artifacts,
            lease: LeaseDto {
                token: lease_token,
                expires_at: OffsetDateTime::now_utc()
                    + time::Duration::seconds(self.config().ci.lease_secs as i64),
                heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,
                attempt: job.attempt,
            },
            actions: None,
        };
        Ok(payload)
    }
}

impl CiService {
    /// Returns the claim payload (with a fresh lease token) for a leased job.
    pub async fn claim_payload(
        &self,
        job_id: Uuid,
        runner_id: Uuid,
    ) -> CiResult<Option<ci_protocol::runner::JobPayload>> {
        let job = workflow_job::Entity::find_by_id(job_id)
            .one(self.db_read())
            .await?
            .ok_or(CiError::NotFound)?;
        if job.runner_id != Some(runner_id) {
            return Ok(None);
        }
        let lease_token = job
            .lease_token
            .clone()
            .unwrap_or_else(|| crate::secrets::random_secret(24));
        self.build_job_payload(job_id, runner_id, lease_token)
            .await
            .map(Some)
    }
}