use crate::{ControlPlaneError, DaemonId, LeaseToken};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::fmt::{Debug, Formatter, Result as FmtResult};
use uuid::Uuid;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct CloneJobId(pub Uuid);

impl CloneJobId {
    pub fn new() -> Self {
        Self(Uuid::new_v4())
    }
}

impl Default for CloneJobId {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Clone, Serialize, Deserialize)]
pub struct CloneSourceSpec {
    pub source_uri: String,
}

impl CloneSourceSpec {
    pub fn validate(&self) -> Result<(), ControlPlaneError> {
        let trimmed = self.source_uri.trim();
        if trimmed.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "source_uri must not be empty".to_string(),
            ));
        }
        if !(trimmed.starts_with("postgres://") || trimmed.starts_with("postgresql://")) {
            return Err(ControlPlaneError::InvalidInput(
                "source_uri must use postgres:// or postgresql://".to_string(),
            ));
        }
        Ok(())
    }

    pub fn redacted_uri(&self) -> String {
        redact_uri(&self.source_uri)
    }
}

impl Debug for CloneSourceSpec {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        f.debug_struct("CloneSourceSpec")
            .field("source_uri", &self.redacted_uri())
            .finish()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum CloneTargetMode {
    CreateManagedInstance,
    ReuseManagedInstance,
}

#[derive(Clone, Serialize, Deserialize)]
pub struct CreateManagedTargetSpec {
    pub client_name: String,
    pub container_name: Option<String>,
    pub image: Option<String>,
    pub host: Option<String>,
    pub host_port: Option<u16>,
    pub database_name: Option<String>,
    pub username: Option<String>,
    pub password: Option<String>,
    #[serde(default)]
    pub reuse_existing_container: bool,
    pub startup_timeout_secs: Option<u64>,
}

impl Debug for CreateManagedTargetSpec {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        f.debug_struct("CreateManagedTargetSpec")
            .field("client_name", &self.client_name)
            .field("container_name", &self.container_name)
            .field("image", &self.image)
            .field("host", &self.host)
            .field("host_port", &self.host_port)
            .field("database_name", &self.database_name)
            .field("username", &self.username)
            .field("password", &redact_secret(self.password.as_deref()))
            .field("reuse_existing_container", &self.reuse_existing_container)
            .field("startup_timeout_secs", &self.startup_timeout_secs)
            .finish()
    }
}

#[derive(Clone, Serialize, Deserialize)]
pub struct ReuseManagedTargetSpec {
    pub existing_client_name: String,
    pub target_client_name: String,
    pub database_name: String,
    pub description: Option<String>,
}

impl Debug for ReuseManagedTargetSpec {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        f.debug_struct("ReuseManagedTargetSpec")
            .field("existing_client_name", &self.existing_client_name)
            .field("target_client_name", &self.target_client_name)
            .field("database_name", &self.database_name)
            .field("description", &self.description)
            .finish()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "mode", rename_all = "snake_case")]
pub enum CloneTargetSpec {
    CreateManagedInstance(CreateManagedTargetSpec),
    ReuseManagedInstance(ReuseManagedTargetSpec),
}

impl CloneTargetSpec {
    pub fn mode(&self) -> CloneTargetMode {
        match self {
            Self::CreateManagedInstance(_) => CloneTargetMode::CreateManagedInstance,
            Self::ReuseManagedInstance(_) => CloneTargetMode::ReuseManagedInstance,
        }
    }

    pub fn summary_json(&self) -> Value {
        match self {
            Self::CreateManagedInstance(spec) => json!({
                "mode": "create_managed_instance",
                "client_name": spec.client_name,
                "container_name": spec.container_name,
                "image": spec.image,
                "host": spec.host,
                "host_port": spec.host_port,
                "database_name": spec.database_name,
                "username": spec.username,
                "password": redact_secret(spec.password.as_deref()),
                "reuse_existing_container": spec.reuse_existing_container,
                "startup_timeout_secs": spec.startup_timeout_secs,
            }),
            Self::ReuseManagedInstance(spec) => json!({
                "mode": "reuse_managed_instance",
                "existing_client_name": spec.existing_client_name,
                "target_client_name": spec.target_client_name,
                "database_name": spec.database_name,
                "description": spec.description,
            }),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct CloneRegistrationPolicy {
    #[serde(default)]
    pub register_runtime: bool,
    #[serde(default)]
    pub register_catalog: bool,
    pub register_name: Option<String>,
    pub description: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct CloneProvisioningPolicy {
    #[serde(default)]
    pub provision_schema: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CloneJobRequest {
    pub source: CloneSourceSpec,
    pub target: CloneTargetSpec,
    #[serde(default)]
    pub registration: CloneRegistrationPolicy,
    #[serde(default)]
    pub provisioning: CloneProvisioningPolicy,
    pub requested_by: Option<String>,
    pub max_attempts: Option<i32>,
}

impl CloneJobRequest {
    pub fn validate(&self) -> Result<(), ControlPlaneError> {
        self.source.validate()?;
        match &self.target {
            CloneTargetSpec::CreateManagedInstance(spec) => {
                if spec.client_name.trim().is_empty() {
                    return Err(ControlPlaneError::InvalidInput(
                        "target.client_name must not be empty".to_string(),
                    ));
                }
            }
            CloneTargetSpec::ReuseManagedInstance(spec) => {
                if spec.existing_client_name.trim().is_empty() {
                    return Err(ControlPlaneError::InvalidInput(
                        "target.existing_client_name must not be empty".to_string(),
                    ));
                }
                if spec.target_client_name.trim().is_empty() {
                    return Err(ControlPlaneError::InvalidInput(
                        "target.target_client_name must not be empty".to_string(),
                    ));
                }
                if spec.database_name.trim().is_empty() {
                    return Err(ControlPlaneError::InvalidInput(
                        "target.database_name must not be empty".to_string(),
                    ));
                }
            }
        }
        if (self.registration.register_catalog || self.registration.register_runtime)
            && self
                .registration
                .register_name
                .as_deref()
                .unwrap_or("")
                .trim()
                .is_empty()
        {
            return Err(ControlPlaneError::InvalidInput(
                "registration.register_name is required when registration is enabled".to_string(),
            ));
        }
        Ok(())
    }

    pub fn max_attempts_or_default(&self) -> i32 {
        self.max_attempts.unwrap_or(3).clamp(1, 10)
    }

    pub fn redacted_summary(&self) -> Value {
        json!({
            "source": {
                "source_uri": self.source.redacted_uri(),
            },
            "target": self.target.summary_json(),
            "registration": self.registration,
            "provisioning": self.provisioning,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CloneJobStatus {
    Queued,
    Preparing,
    Running,
    Verifying,
    Completed,
    FailedRetryable,
    FailedTerminal,
    CancelRequested,
    Cancelled,
}

impl CloneJobStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Queued => "queued",
            Self::Preparing => "preparing",
            Self::Running => "running",
            Self::Verifying => "verifying",
            Self::Completed => "completed",
            Self::FailedRetryable => "failed_retryable",
            Self::FailedTerminal => "failed_terminal",
            Self::CancelRequested => "cancel_requested",
            Self::Cancelled => "cancelled",
        }
    }

    pub fn parse(value: &str) -> Result<Self, ControlPlaneError> {
        match value {
            "queued" => Ok(Self::Queued),
            "preparing" => Ok(Self::Preparing),
            "running" => Ok(Self::Running),
            "verifying" => Ok(Self::Verifying),
            "completed" => Ok(Self::Completed),
            "failed_retryable" => Ok(Self::FailedRetryable),
            "failed_terminal" => Ok(Self::FailedTerminal),
            "cancel_requested" => Ok(Self::CancelRequested),
            "cancelled" => Ok(Self::Cancelled),
            _ => Err(ControlPlaneError::Storage(format!(
                "unknown clone job status `{value}`"
            ))),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CloneJobPhase {
    ValidateSource,
    PrepareTarget,
    ResolveTools,
    DumpSource,
    RestoreTarget,
    BootstrapAthena,
    RegisterClient,
    VerifyTarget,
    Complete,
}

impl CloneJobPhase {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::ValidateSource => "validate_source",
            Self::PrepareTarget => "prepare_target",
            Self::ResolveTools => "resolve_tools",
            Self::DumpSource => "dump_source",
            Self::RestoreTarget => "restore_target",
            Self::BootstrapAthena => "bootstrap_athena",
            Self::RegisterClient => "register_client",
            Self::VerifyTarget => "verify_target",
            Self::Complete => "complete",
        }
    }

    pub fn parse(value: &str) -> Result<Self, ControlPlaneError> {
        match value {
            "validate_source" => Ok(Self::ValidateSource),
            "prepare_target" => Ok(Self::PrepareTarget),
            "resolve_tools" => Ok(Self::ResolveTools),
            "dump_source" => Ok(Self::DumpSource),
            "restore_target" => Ok(Self::RestoreTarget),
            "bootstrap_athena" => Ok(Self::BootstrapAthena),
            "register_client" => Ok(Self::RegisterClient),
            "verify_target" => Ok(Self::VerifyTarget),
            "complete" => Ok(Self::Complete),
            _ => Err(ControlPlaneError::Storage(format!(
                "unknown clone job phase `{value}`"
            ))),
        }
    }
}

#[derive(Debug, Clone)]
pub struct CloneJobRecord {
    pub job_id: CloneJobId,
    pub source: CloneSourceSpec,
    pub target: CloneTargetSpec,
    pub registration: CloneRegistrationPolicy,
    pub provisioning: CloneProvisioningPolicy,
    pub requested_by: Option<String>,
    pub status: CloneJobStatus,
    pub phase: CloneJobPhase,
    pub attempt_count: i32,
    pub max_attempts: i32,
    pub execution_daemon_id: Option<DaemonId>,
    pub lease_token: Option<LeaseToken>,
    pub lease_expires_at: Option<DateTime<Utc>>,
    pub cancel_requested: bool,
    pub redacted_summary: Value,
    pub verification_summary: Option<Value>,
    pub failure_summary: Option<Value>,
    pub created_at: DateTime<Utc>,
    pub started_at: Option<DateTime<Utc>>,
    pub updated_at: DateTime<Utc>,
    pub completed_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Clone, Serialize)]
pub struct CloneJobSummary {
    pub job_id: Uuid,
    pub status: CloneJobStatus,
    pub phase: CloneJobPhase,
    pub attempt_count: i32,
    pub max_attempts: i32,
    pub execution_daemon_id: Option<String>,
    pub cancel_requested: bool,
    pub requested_by: Option<String>,
    pub redacted_summary: Value,
    pub verification_summary: Option<Value>,
    pub failure_summary: Option<Value>,
    pub created_at: DateTime<Utc>,
    pub started_at: Option<DateTime<Utc>>,
    pub updated_at: DateTime<Utc>,
    pub completed_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Clone, Serialize)]
pub struct CloneJobDetail {
    pub summary: CloneJobSummary,
    pub source: Value,
    pub target: Value,
    pub registration: CloneRegistrationPolicy,
    pub provisioning: CloneProvisioningPolicy,
    pub lease_expires_at: Option<DateTime<Utc>>,
}

impl CloneJobRecord {
    pub fn summary(&self) -> CloneJobSummary {
        CloneJobSummary {
            job_id: self.job_id.0,
            status: self.status,
            phase: self.phase,
            attempt_count: self.attempt_count,
            max_attempts: self.max_attempts,
            execution_daemon_id: self
                .execution_daemon_id
                .as_ref()
                .map(|value| value.as_str().to_string()),
            cancel_requested: self.cancel_requested,
            requested_by: self.requested_by.clone(),
            redacted_summary: self.redacted_summary.clone(),
            verification_summary: self.verification_summary.clone(),
            failure_summary: self.failure_summary.clone(),
            created_at: self.created_at,
            started_at: self.started_at,
            updated_at: self.updated_at,
            completed_at: self.completed_at,
        }
    }

    pub fn detail(&self) -> CloneJobDetail {
        CloneJobDetail {
            summary: self.summary(),
            source: json!({ "source_uri": self.source.redacted_uri() }),
            target: self.target.summary_json(),
            registration: self.registration.clone(),
            provisioning: self.provisioning.clone(),
            lease_expires_at: self.lease_expires_at,
        }
    }
}

fn redact_secret(value: Option<&str>) -> Option<&'static str> {
    value.map(|value| {
        if value.is_empty() {
            "<empty>"
        } else {
            "<redacted>"
        }
    })
}

fn redact_uri(uri: &str) -> String {
    let trimmed = uri.trim();
    if trimmed.is_empty() {
        return "<empty>".to_string();
    }

    let Some(scheme_idx) = trimmed.find("://") else {
        return "<redacted-uri>".to_string();
    };

    let scheme = &trimmed[..scheme_idx + 3];
    let rest = &trimmed[scheme_idx + 3..];
    let masked = if let Some(at_idx) = rest.rfind('@') {
        let after_at = &rest[at_idx + 1..];
        format!("<redacted>@{after_at}")
    } else {
        rest.to_string()
    };

    format!("{scheme}{masked}")
}

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

    #[test]
    fn source_spec_rejects_non_postgres_uri() {
        let err = CloneSourceSpec {
            source_uri: "mysql://user:pass@localhost/db".to_string(),
        }
        .validate()
        .expect_err("non-postgres uri should fail");
        assert!(err.to_string().contains("postgres://"));
    }

    #[test]
    fn request_requires_register_name_when_registration_enabled() {
        let err = CloneJobRequest {
            source: CloneSourceSpec {
                source_uri: "postgres://user:pass@localhost:5432/source".to_string(),
            },
            target: CloneTargetSpec::CreateManagedInstance(CreateManagedTargetSpec {
                client_name: "target".to_string(),
                container_name: None,
                image: None,
                host: None,
                host_port: None,
                database_name: None,
                username: None,
                password: None,
                reuse_existing_container: false,
                startup_timeout_secs: None,
            }),
            registration: CloneRegistrationPolicy {
                register_runtime: true,
                ..CloneRegistrationPolicy::default()
            },
            provisioning: CloneProvisioningPolicy::default(),
            requested_by: None,
            max_attempts: None,
        }
        .validate()
        .expect_err("missing register name should fail");
        assert!(err.to_string().contains("register_name"));
    }

    #[test]
    fn redacted_summary_masks_source_and_target_password() {
        let request = CloneJobRequest {
            source: CloneSourceSpec {
                source_uri: "postgres://user:pass@localhost:5432/source".to_string(),
            },
            target: CloneTargetSpec::CreateManagedInstance(CreateManagedTargetSpec {
                client_name: "target".to_string(),
                container_name: None,
                image: None,
                host: None,
                host_port: None,
                database_name: None,
                username: Some("postgres".to_string()),
                password: Some("super-secret".to_string()),
                reuse_existing_container: false,
                startup_timeout_secs: None,
            }),
            registration: CloneRegistrationPolicy::default(),
            provisioning: CloneProvisioningPolicy::default(),
            requested_by: None,
            max_attempts: None,
        };

        let redacted = request.redacted_summary().to_string();
        assert!(redacted.contains("<redacted>@localhost"));
        assert!(redacted.contains("<redacted>"));
        assert!(!redacted.contains("super-secret"));
    }
}
