use std::collections::BTreeMap;

use serde::{Deserialize, Serialize};

use crate::compat::CompatibilityPolicy;
use crate::schema::SchemaDescriptor;

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ResourceKind {
    Topic,
    Service,
    Action,
    Mission,
}

impl ResourceKind {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Topic => "topic",
            Self::Service => "service",
            Self::Action => "action",
            Self::Mission => "mission",
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct TopicContract {
    pub message: SchemaDescriptor,
    pub encoding_hint: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ServiceContract {
    pub request: SchemaDescriptor,
    pub response: SchemaDescriptor,
    pub encoding_hint: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ActionContract {
    pub goal: SchemaDescriptor,
    pub feedback: SchemaDescriptor,
    pub result: SchemaDescriptor,
    pub encoding_hint: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct MissionContract {
    pub open: SchemaDescriptor,
    pub command: SchemaDescriptor,
    pub ack: SchemaDescriptor,
    pub update: SchemaDescriptor,
    pub feedback: Option<SchemaDescriptor>,
    pub encoding_hint: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "contract", rename_all = "snake_case")]
pub enum ResourceContract {
    Topic(TopicContract),
    Service(ServiceContract),
    Action(ActionContract),
    Mission(MissionContract),
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
pub struct ContractKey {
    pub resource_kind: ResourceKind,
    pub resolved_name: String,
}

impl ContractKey {
    pub fn new(resource_kind: ResourceKind, resolved_name: impl Into<String>) -> Self {
        Self {
            resource_kind,
            resolved_name: resolved_name.into(),
        }
    }
}

impl TopicContract {
    pub fn schema_summary(&self) -> Vec<String> {
        vec![format!("msg={}", schema_ref_summary(&self.message))]
    }

    pub fn is_compatible_with(&self, other: &Self, policy: CompatibilityPolicy) -> bool {
        self.encoding_hint == other.encoding_hint
            && schema_compatible(&self.message, &other.message, policy)
    }
}

impl ServiceContract {
    pub fn schema_summary(&self) -> Vec<String> {
        vec![
            format!("req={}", schema_ref_summary(&self.request)),
            format!("res={}", schema_ref_summary(&self.response)),
        ]
    }

    pub fn is_compatible_with(&self, other: &Self, policy: CompatibilityPolicy) -> bool {
        self.encoding_hint == other.encoding_hint
            && schema_compatible(&self.request, &other.request, policy)
            && schema_compatible(&self.response, &other.response, policy)
    }
}

impl ActionContract {
    pub fn schema_summary(&self) -> Vec<String> {
        vec![
            format!("goal={}", schema_ref_summary(&self.goal)),
            format!("feedback={}", schema_ref_summary(&self.feedback)),
            format!("result={}", schema_ref_summary(&self.result)),
        ]
    }

    pub fn is_compatible_with(&self, other: &Self, policy: CompatibilityPolicy) -> bool {
        self.encoding_hint == other.encoding_hint
            && schema_compatible(&self.goal, &other.goal, policy)
            && schema_compatible(&self.feedback, &other.feedback, policy)
            && schema_compatible(&self.result, &other.result, policy)
    }
}

impl MissionContract {
    pub fn schema_summary(&self) -> Vec<String> {
        let mut parts = vec![
            format!("open={}", schema_ref_summary(&self.open)),
            format!("command={}", schema_ref_summary(&self.command)),
            format!("ack={}", schema_ref_summary(&self.ack)),
            format!("update={}", schema_ref_summary(&self.update)),
        ];
        parts.push(match &self.feedback {
            Some(feedback) => format!("feedback={}", schema_ref_summary(feedback)),
            None => String::from("feedback=?"),
        });
        parts
    }

    pub fn is_compatible_with(&self, other: &Self, policy: CompatibilityPolicy) -> bool {
        self.encoding_hint == other.encoding_hint
            && schema_compatible(&self.open, &other.open, policy)
            && schema_compatible(&self.command, &other.command, policy)
            && schema_compatible(&self.ack, &other.ack, policy)
            && schema_compatible(&self.update, &other.update, policy)
            && optional_schema_compatible(self.feedback.as_ref(), other.feedback.as_ref(), policy)
    }
}

impl ResourceContract {
    pub fn resource_kind(&self) -> ResourceKind {
        match self {
            Self::Topic(_) => ResourceKind::Topic,
            Self::Service(_) => ResourceKind::Service,
            Self::Action(_) => ResourceKind::Action,
            Self::Mission(_) => ResourceKind::Mission,
        }
    }

    pub fn encoding_hint(&self) -> Option<&str> {
        match self {
            Self::Topic(contract) => contract.encoding_hint.as_deref(),
            Self::Service(contract) => contract.encoding_hint.as_deref(),
            Self::Action(contract) => contract.encoding_hint.as_deref(),
            Self::Mission(contract) => contract.encoding_hint.as_deref(),
        }
    }

    pub fn schema_summary(&self) -> Vec<String> {
        match self {
            Self::Topic(contract) => contract.schema_summary(),
            Self::Service(contract) => contract.schema_summary(),
            Self::Action(contract) => contract.schema_summary(),
            Self::Mission(contract) => contract.schema_summary(),
        }
    }

    pub fn schema_summary_text(&self) -> String {
        self.schema_summary().join(", ")
    }

    pub fn is_compatible_with(&self, other: &Self, policy: CompatibilityPolicy) -> bool {
        match (self, other) {
            (Self::Topic(left), Self::Topic(right)) => left.is_compatible_with(right, policy),
            (Self::Service(left), Self::Service(right)) => left.is_compatible_with(right, policy),
            (Self::Action(left), Self::Action(right)) => left.is_compatible_with(right, policy),
            (Self::Mission(left), Self::Mission(right)) => left.is_compatible_with(right, policy),
            _ => false,
        }
    }

    pub fn contract_hash(&self) -> String {
        let canonical = self.canonical_string();
        let mut hash = 0xcbf29ce484222325u64;
        for byte in canonical.as_bytes() {
            hash ^= u64::from(*byte);
            hash = hash.wrapping_mul(0x100000001b3);
        }
        format!("{:016x}", hash)
    }

    pub fn canonical_string(&self) -> String {
        let mut fields = BTreeMap::new();
        fields.insert(
            String::from("kind"),
            self.resource_kind().as_str().to_string(),
        );

        match self {
            Self::Topic(contract) => {
                insert_schema_fields(&mut fields, "message", &contract.message);
                insert_optional_string(
                    &mut fields,
                    "encoding_hint",
                    contract.encoding_hint.as_deref(),
                );
            }
            Self::Service(contract) => {
                insert_schema_fields(&mut fields, "request", &contract.request);
                insert_schema_fields(&mut fields, "response", &contract.response);
                insert_optional_string(
                    &mut fields,
                    "encoding_hint",
                    contract.encoding_hint.as_deref(),
                );
            }
            Self::Action(contract) => {
                insert_schema_fields(&mut fields, "goal", &contract.goal);
                insert_schema_fields(&mut fields, "feedback", &contract.feedback);
                insert_schema_fields(&mut fields, "result", &contract.result);
                insert_optional_string(
                    &mut fields,
                    "encoding_hint",
                    contract.encoding_hint.as_deref(),
                );
            }
            Self::Mission(contract) => {
                insert_schema_fields(&mut fields, "open", &contract.open);
                insert_schema_fields(&mut fields, "command", &contract.command);
                insert_schema_fields(&mut fields, "ack", &contract.ack);
                insert_schema_fields(&mut fields, "update", &contract.update);
                match &contract.feedback {
                    Some(feedback) => insert_schema_fields(&mut fields, "feedback", feedback),
                    None => {
                        fields.insert(String::from("feedback.id"), String::new());
                        fields.insert(String::from("feedback.version"), String::new());
                        fields.insert(String::from("feedback.name"), String::new());
                    }
                }
                insert_optional_string(
                    &mut fields,
                    "encoding_hint",
                    contract.encoding_hint.as_deref(),
                );
            }
        }

        fields
            .into_iter()
            .map(|(key, value)| format!("{key}={value}"))
            .collect::<Vec<_>>()
            .join("|")
    }
}

fn insert_schema_fields(
    fields: &mut BTreeMap<String, String>,
    prefix: &str,
    schema: &SchemaDescriptor,
) {
    fields.insert(format!("{prefix}.id"), schema.id.0.clone());
    fields.insert(format!("{prefix}.version"), schema.version.0.to_string());
    fields.insert(format!("{prefix}.name"), schema.name.clone());
}

fn insert_optional_string(fields: &mut BTreeMap<String, String>, key: &str, value: Option<&str>) {
    fields.insert(key.to_string(), value.unwrap_or_default().to_string());
}

pub fn schema_ref_summary(schema: &SchemaDescriptor) -> String {
    format!("{}@{}", schema.id.0, schema.version.0)
}

fn schema_compatible(
    registered: &SchemaDescriptor,
    candidate: &SchemaDescriptor,
    policy: CompatibilityPolicy,
) -> bool {
    registered.id == candidate.id
        && registered.name == candidate.name
        && policy.check(registered.version, candidate.version)
}

fn optional_schema_compatible(
    registered: Option<&SchemaDescriptor>,
    candidate: Option<&SchemaDescriptor>,
    policy: CompatibilityPolicy,
) -> bool {
    match (registered, candidate) {
        (Some(registered), Some(candidate)) => schema_compatible(registered, candidate, policy),
        (None, None) => true,
        _ => false,
    }
}
