use std::{borrow::Cow, marker::PhantomData};

use serde::{Serialize, de::DeserializeOwned};
use zenoh::{
    Wait,
    bytes::{Encoding, ZBytes},
    handlers::FifoChannelHandler,
    key_expr::OwnedKeyExpr,
    pubsub::Subscriber,
    query::{Query as ZenohQuery, QueryConsolidation, QueryTarget, Queryable, ReplyKeyExpr},
    sample::{Locality, Sample},
};

use crate::bus::metadata::BusMetadata;
use crate::bus::query::{Retry, retry_query};
use crate::bus::topic::{PubSub, Query as TopicQuery, Topic};
use crate::bus::zenoh::{deserialize_payload, serialize_payload, typed_encoding};
use crate::bus::{Bus, Error, Result};

#[derive(Debug, Clone, PartialEq)]
pub struct Received<T> {
    pub at_ns: Option<u64>,
    pub value: T,
}

pub struct TypedTopicSubscriber<T> {
    inner: Subscriber<FifoChannelHandler<Sample>>,
    topic_key: String,
    schema: &'static str,
    schema_version: u32,
    _phantom: PhantomData<T>,
}

impl<T: DeserializeOwned> TypedTopicSubscriber<T> {
    pub async fn recv(&self) -> Result<Received<T>> {
        let sample = self.inner.recv_async().await?;
        decode_sample(&sample, &self.topic_key, self.schema, self.schema_version)
    }

    pub fn key_expr(&self) -> &zenoh::key_expr::KeyExpr<'static> {
        self.inner.key_expr()
    }
}

pub struct TypedTopicResponder<Req, Resp> {
    inner: Queryable<FifoChannelHandler<ZenohQuery>>,
    topic_key: String,
    schema: &'static str,
    schema_version: u32,
    _phantom: PhantomData<(Req, Resp)>,
}

impl<Req, Resp> TypedTopicResponder<Req, Resp> {
    pub async fn recv(&self) -> Result<TypedTopicQuery<Req, Resp>> {
        let query = self.inner.recv_async().await?;
        Ok(TypedTopicQuery {
            inner: query,
            topic_key: self.topic_key.clone(),
            schema: self.schema,
            schema_version: self.schema_version,
            _phantom: PhantomData,
        })
    }
}

pub struct TypedTopicQuery<Req, Resp> {
    inner: ZenohQuery,
    topic_key: String,
    schema: &'static str,
    schema_version: u32,
    _phantom: PhantomData<(Req, Resp)>,
}

impl<Req: DeserializeOwned, Resp: Serialize> TypedTopicQuery<Req, Resp> {
    pub fn request(&self) -> Result<Req> {
        let payload = self.inner.payload().ok_or_else(|| {
            Error::TypedDecode(format!(
                "topic '{}' schema '{}' query is missing a request payload",
                self.topic_key, self.schema
            ))
        })?;
        validate_contract(
            self.inner.encoding(),
            self.inner.attachment(),
            &self.topic_key,
            self.schema,
            self.schema_version,
        )?;
        deserialize_payload(payload).map_err(Error::from)
    }

    pub async fn reply(&self, response: &Resp) -> Result<()> {
        let payload = serialize_payload(response)?;
        self.inner
            .reply(self.inner.key_expr(), payload)
            .encoding(typed_encoding(self.schema))
            .attachment(
                BusMetadata {
                    schema_version: self.schema_version,
                    produced_at_ns: None,
                }
                .encode(),
            )
            .await?;
        Ok(())
    }

    pub async fn reply_err<Payload: Into<ZBytes>>(&self, payload: Payload) -> Result<()> {
        self.inner.reply_err(payload).await?;
        Ok(())
    }
}

impl Bus {
    pub async fn publish<T: Serialize>(
        &self,
        topic: &Topic<PubSub<T>>,
        at_ns: u64,
        data: &T,
    ) -> Result<()> {
        let key = concrete_publish_key(topic)?;
        let key = owned_key(self.topic(key.as_ref()))?;
        let payload = serialize_payload(data)?;
        self.session()
            .put(key, payload)
            .encoding(typed_encoding(topic.schema()))
            .attachment(
                BusMetadata {
                    schema_version: topic.version(),
                    produced_at_ns: Some(at_ns),
                }
                .encode(),
            )
            .await?;
        Ok(())
    }

    pub async fn subscriber<T: DeserializeOwned>(
        &self,
        topic: &Topic<PubSub<T>>,
    ) -> Result<TypedTopicSubscriber<T>> {
        let key = self.topic(topic.key().as_ref());
        let inner = declare_subscriber(self, &key).await?;
        Ok(TypedTopicSubscriber {
            inner,
            topic_key: key,
            schema: topic.schema(),
            schema_version: topic.version(),
            _phantom: PhantomData,
        })
    }

    pub async fn request<Req: Serialize, Resp: DeserializeOwned>(
        &self,
        topic: &Topic<TopicQuery<Req, Resp>>,
        request: &Req,
        retry: &Retry,
    ) -> Result<Option<Resp>> {
        let name = topic.key().into_owned();
        retry_query(&name, retry, || async {
            self.request_once(topic, request).await
        })
        .await
    }

    pub async fn responder<Req: DeserializeOwned, Resp: Serialize>(
        &self,
        topic: &Topic<TopicQuery<Req, Resp>>,
    ) -> Result<TypedTopicResponder<Req, Resp>> {
        let key = self.topic(topic.key().as_ref());
        let inner = self
            .session()
            .declare_queryable(owned_key(key.clone())?)
            .await?;
        Ok(TypedTopicResponder {
            inner,
            topic_key: key,
            schema: topic.schema(),
            schema_version: topic.version(),
            _phantom: PhantomData,
        })
    }

    async fn request_once<Req: Serialize, Resp: DeserializeOwned>(
        &self,
        topic: &Topic<TopicQuery<Req, Resp>>,
        request: &Req,
    ) -> Result<Option<Resp>> {
        let key = self.topic(topic.key().as_ref());
        let receiver = self
            .session()
            .get(owned_key(key.clone())?)
            .payload(serialize_payload(request)?)
            .encoding(typed_encoding(topic.schema()))
            .attachment(
                BusMetadata {
                    schema_version: topic.version(),
                    produced_at_ns: None,
                }
                .encode(),
            )
            .target(QueryTarget::DEFAULT)
            .consolidation(QueryConsolidation::DEFAULT)
            .allowed_destination(Locality::default())
            .accept_replies(ReplyKeyExpr::MatchingQuery)
            .wait()?;

        let Ok(reply) = receiver.recv_async().await else {
            return Ok(None);
        };

        match reply.into_result() {
            Ok(sample) => Ok(Some(decode_response(
                &sample,
                &key,
                topic.schema(),
                topic.version(),
            )?)),
            Err(reply_error) => Err(Error::TypedDecode(format!(
                "topic '{key}' schema '{}' returned an error reply with {} bytes",
                topic.schema(),
                reply_error.payload().to_bytes().len()
            ))),
        }
    }
}

fn concrete_publish_key<T>(topic: &Topic<PubSub<T>>) -> Result<Cow<'static, str>> {
    topic.publish_key()
}

async fn declare_subscriber(
    bus: &Bus,
    key: &str,
) -> Result<Subscriber<FifoChannelHandler<Sample>>> {
    Ok(bus
        .session()
        .declare_subscriber(owned_key(key.to_string())?)
        .await?)
}

fn owned_key(key: String) -> Result<OwnedKeyExpr> {
    OwnedKeyExpr::new(key).map_err(|error| Error::InvalidTopic(error.to_string()))
}

fn decode_sample<T: DeserializeOwned>(
    sample: &Sample,
    topic_key: &str,
    schema: &'static str,
    schema_version: u32,
) -> Result<Received<T>> {
    let metadata = validate_contract(
        Some(sample.encoding()),
        sample.attachment(),
        topic_key,
        schema,
        schema_version,
    )?;
    let value = deserialize_payload(sample.payload()).map_err(Error::from)?;
    Ok(Received {
        at_ns: metadata.produced_at_ns,
        value,
    })
}

fn decode_response<T: DeserializeOwned>(
    sample: &Sample,
    topic_key: &str,
    schema: &'static str,
    schema_version: u32,
) -> Result<T> {
    validate_contract(
        Some(sample.encoding()),
        sample.attachment(),
        topic_key,
        schema,
        schema_version,
    )?;
    deserialize_payload(sample.payload()).map_err(Error::from)
}

fn validate_contract(
    encoding: Option<&Encoding>,
    attachment: Option<&ZBytes>,
    topic_key: &str,
    schema: &'static str,
    schema_version: u32,
) -> Result<BusMetadata> {
    validate_encoding(encoding, topic_key, schema)?;
    let metadata = decode_metadata(attachment, topic_key, schema)?;
    if metadata.schema_version != schema_version {
        return Err(Error::TypedDecode(format!(
            "topic '{topic_key}' schema '{schema}' version mismatch: expected {schema_version}, received {}",
            metadata.schema_version
        )));
    }
    Ok(metadata)
}

fn validate_encoding(
    encoding: Option<&Encoding>,
    topic_key: &str,
    schema: &'static str,
) -> Result<()> {
    let expected = typed_encoding(schema);
    let Some(received) = encoding else {
        return Err(Error::TypedDecode(format!(
            "topic '{topic_key}' schema '{schema}' encoding mismatch: expected {expected}, received <none>"
        )));
    };

    if *received != expected {
        return Err(Error::TypedDecode(format!(
            "topic '{topic_key}' schema '{schema}' encoding mismatch: expected {expected}, received {received}"
        )));
    }
    Ok(())
}

fn decode_metadata(
    attachment: Option<&ZBytes>,
    topic_key: &str,
    schema: &'static str,
) -> Result<BusMetadata> {
    let attachment = attachment.ok_or_else(|| {
        Error::TypedDecode(format!(
            "topic '{topic_key}' schema '{schema}' is missing BusMetadata attachment"
        ))
    })?;
    BusMetadata::decode(attachment.to_bytes().as_ref()).map_err(|error| {
        Error::TypedDecode(format!(
            "topic '{topic_key}' schema '{schema}' has invalid BusMetadata attachment: {error}"
        ))
    })
}

#[cfg(test)]
mod tests {
    use std::time::{Duration, SystemTime, UNIX_EPOCH};

    use serde::{Deserialize, Serialize};
    use serial_test::serial;
    use tokio::time::timeout;
    use zenoh::{key_expr::KeyExpr, sample::SampleBuilder};

    use super::{concrete_publish_key, decode_sample};
    use crate::api::v1::{asset, component::capability::motor, topic};
    use crate::bus::metadata::BusMetadata;
    use crate::bus::query::Retry;
    use crate::bus::topic::ANY;
    use crate::bus::zenoh::{serialize_payload, typed_encoding};
    use crate::bus::{Bus, Error};

    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
    struct TestPayload {
        value: u32,
    }

    #[test]
    fn typed_topic_keys_are_ready_for_bus_prefix_application() -> crate::bus::Result<()> {
        let concrete = topic::new()
            .v1()
            .component("base")
            .motor("left_wheel")
            .command();
        let wildcard = topic::new().v1().component(ANY).motor(ANY).command();

        assert_eq!(
            concrete_publish_key(&concrete)?.as_ref(),
            "v1/component/base/motor/left_wheel/command"
        );
        assert_eq!(wildcard.key().as_ref(), "v1/component/*/motor/*/command");
        assert_eq!(
            format!("robot-a/{}", concrete_publish_key(&concrete)?),
            "robot-a/v1/component/base/motor/left_wheel/command"
        );
        assert_eq!(
            format!("robot-a/{}", wildcard.key()),
            "robot-a/v1/component/*/motor/*/command"
        );
        Ok(())
    }

    #[test]
    fn typed_publish_rejects_wildcard_topic_before_transport() {
        let topic = topic::new().v1().component(ANY).motor(ANY).command();
        assert!(matches!(
            concrete_publish_key(&topic),
            Err(Error::InvalidTopic(_))
        ));
    }

    #[test]
    fn typed_decode_error_names_schema_mismatch_context() {
        let sample = sample_with_contract("other/schema", 1);
        let error = decode_sample::<TestPayload>(
            &sample,
            "v1/component/*/motor/*/command",
            "v1/component/motor/command",
            1,
        )
        .expect_err("schema mismatch should fail")
        .to_string();

        assert!(error.contains("v1/component/*/motor/*/command"));
        assert!(error.contains("v1/component/motor/command"));
        assert!(error.contains("other/schema"));
        assert!(error.contains("encoding mismatch"));
    }

    #[test]
    fn typed_decode_error_names_version_mismatch_context() {
        let sample = sample_with_contract("v1/component/motor/command", 2);
        let error = decode_sample::<TestPayload>(
            &sample,
            "v1/component/*/motor/*/command",
            "v1/component/motor/command",
            1,
        )
        .expect_err("version mismatch should fail")
        .to_string();

        assert!(error.contains("v1/component/*/motor/*/command"));
        assert!(error.contains("v1/component/motor/command"));
        assert!(error.contains("version mismatch"));
        assert!(error.contains("expected 1"));
        assert!(error.contains("received 2"));
    }

    #[serial]
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn typed_live_pubsub_roundtrip_supports_wildcard_subscriber() -> crate::bus::Result<()> {
        let bus = open_bus("pubsub").await;
        let wildcard = topic::new().v1().component(ANY).motor(ANY).command();
        let wildcard_error = bus
            .publish(&wildcard, 1, &motor::Command::Velocity(0.0))
            .await
            .expect_err("publishing to a wildcard topic should fail");
        assert!(matches!(wildcard_error, Error::InvalidTopic(_)));

        let subscriber = bus.subscriber(&wildcard).await?;
        let concrete = topic::new()
            .v1()
            .component("base")
            .motor("left_wheel")
            .command();
        let now_ns = 42_000_000;
        let command = motor::Command::Velocity(0.75);

        bus.publish(&concrete, now_ns, &command).await?;

        let received = timeout(Duration::from_secs(5), subscriber.recv())
            .await
            .expect("typed subscriber should receive within timeout")?;
        assert_eq!(received.at_ns, Some(now_ns));
        match received.value {
            motor::Command::Velocity(value) => assert_eq!(value, 0.75),
            _ => panic!("expected velocity motor command"),
        }

        bus.close().await?;
        Ok(())
    }

    #[serial]
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn typed_live_query_roundtrip_returns_response() -> crate::bus::Result<()> {
        let bus = open_bus("query").await;
        let topic = topic::new().v1().asset().get();
        let responder = bus.responder(&topic).await?;
        let request = asset::GetRequest {
            path: "fixture-map".to_string(),
        };
        let retry = Retry::new(1)
            .with_initial_backoff(Duration::from_millis(10))
            .with_max_backoff(Duration::from_millis(10));

        let (handled, response) = timeout(Duration::from_secs(5), async {
            let responder_task = async {
                let query = responder.recv().await?;
                let request = query.request()?;
                query
                    .reply(&asset::GetResponse::Ok {
                        bytes: vec![1, 2, 3, 5, 8],
                    })
                    .await?;
                crate::bus::Result::Ok(request)
            };
            tokio::join!(responder_task, bus.request(&topic, &request, &retry))
        })
        .await
        .expect("typed query should complete within timeout");

        assert_eq!(handled?, request);
        assert_eq!(
            response?,
            Some(asset::GetResponse::Ok {
                bytes: vec![1, 2, 3, 5, 8]
            })
        );

        bus.close().await?;
        Ok(())
    }

    fn sample_with_contract(schema: &'static str, version: u32) -> zenoh::sample::Sample {
        let payload =
            serialize_payload(&TestPayload { value: 42 }).expect("test payload should serialize");
        let key: KeyExpr<'static> = KeyExpr::try_from("v1/component/base/motor/left_wheel/command")
            .expect("test key should be valid");
        SampleBuilder::put(key, payload)
            .encoding(typed_encoding(schema))
            .attachment(
                BusMetadata {
                    schema_version: version,
                    produced_at_ns: Some(9),
                }
                .encode(),
            )
            .into()
    }

    async fn open_bus(suffix: &str) -> Bus {
        let session = zenoh::open(local_test_config())
            .await
            .expect("test zenoh session should open");
        Bus::new(session, unique_prefix(suffix))
    }

    fn local_test_config() -> zenoh::Config {
        let mut config = zenoh::Config::default();
        config.insert_json5("listen/endpoints", "[]").unwrap();
        config
            .insert_json5("scouting/multicast/enabled", "false")
            .unwrap();
        config
    }

    fn unique_prefix(suffix: &str) -> String {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        format!("phoxal/tests/typed/{suffix}/{nanos}")
    }
}
