gmqtt 0.1.1

use heapless::Vec;

use crate::{utils::*, MqttError, Pid, QoS, QosPid};

use super::{property, Header, PropertyType};

/// A PUBLISH packet is sent from a Client to a Server or from a Server to a Client to transport an Application Message.
///
/// [MQTT5 SPECIFICATION](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901100)
#[derive(Debug, Clone, PartialEq)]
pub struct Publish<'a> {
    /// Position: byte 1, bit 3.
    ///
    /// If the DUP flag is set to 0, it indicates that this is the first occasion that the Client or Server has attempted to send this PUBLISH packet.
    ///
    /// If the DUP flag is set to 1, it indicates that this might be re-delivery of an earlier attempt to send the packet.
    pub dup: bool,

    /// QoS:
    /// Position: byte 1, bits 2-1.
    ///
    /// This field indicates the level of assurance for delivery of an Application Message.
    ///
    /// Pid:
    /// The Packet Identifier field is only present in PUBLISH packets where the QoS level is 1 or 2
    pub qospid: QosPid,

    /// Position: byte 1, bit 0.
    ///
    /// If the RETAIN flag is set to 1 in a PUBLISH packet sent by a Client to a Server,
    /// the Server MUST replace any existing retained message for this topic and store the Application Message,
    /// so that it can be delivered to future subscribers whose subscriptions match its Topic Name
    pub retain: bool,

    /// The Topic Name identifies the information channel to which Payload data is published.
    pub topic: &'a str,

    pub properties: Option<PublishProperties<'a>>,

    /// The Payload contains the Application Message that is being published.
    /// The content and format of the data is application specific.
    /// The length of the Payload can be calculated by subtracting the length of the Variable Header from the Remaining Length field that is in the Fixed Header.
    /// It is valid for a PUBLISH packet to contain a zero length Payload.
    pub payload: &'a [u8],
}

impl<'a> Publish<'a> {
    pub fn new(topic: &'a str, qospid: QosPid, payload: &'a [u8]) -> Self {
        Publish {
            dup: false,
            qospid,
            retain: false,
            topic,
            properties: None,
            payload,
        }
    }

    pub fn read(header: Header, buf: &'a [u8], offset: &mut usize) -> Result<Self, MqttError> {
        let dup = header.dup;
        let retain = header.retain;
        let topic = read_str(buf, offset)?;

        let qospid = match header.qos {
            QoS::AtMostOnce => QosPid::AtMostOnce,
            QoS::AtLeastOnce => QosPid::AtLeastOnce(Pid::read(buf, offset)?),
            QoS::ExactlyOnce => QosPid::ExactlyOnce(Pid::read(buf, offset)?),
        };

        let properties = PublishProperties::read(buf, offset)?;

        // move offset to payload field
        *offset += 1;

        let payload = &buf[*offset..];

        let publish = Publish {
            dup,
            qospid,
            retain,
            topic,
            properties,
            payload,
        };

        Ok(publish)
    }
    pub fn write(&self, buf: &mut [u8], offset: &mut usize) -> Result<usize, MqttError> {
        check_remaining(buf, offset, self.len())?;

        let dup = self.dup as u8;
        let qos = self.qospid.qos() as u8;
        let retain = self.retain as u8;

        write_u8(buf, offset, 0b0011_0000 | retain | qos << 1 | dup << 3);

        let write_len = write_length(buf, offset, self.len())? + 1;

        write_bytes_with_len(buf, offset, self.topic.as_bytes());

        match self.qospid {
            QosPid::AtMostOnce => {}
            QosPid::AtLeastOnce(pid) => pid.write(buf, offset),
            QosPid::ExactlyOnce(pid) => pid.write(buf, offset),
        }

        match &self.properties {
            Some(properties) => properties.write(buf, offset)?,
            None => write_u8(buf, offset, 0),
        }

        write_bytes(buf, offset, self.payload);

        Ok(write_len)
    }

    fn len(&self) -> usize {
        let mut len = 2 + self.topic.len();
        if self.qospid != QosPid::AtMostOnce {
            len += 2;
        }

        match &self.properties {
            Some(properties) => {
                let properties_len = properties.len();
                let properties_len_len = len_len(properties_len);
                len += properties_len_len + properties_len;
            }
            None => {
                // just 1 byte representing 0 len
                len += 1;
            }
        }

        len += self.payload.len();
        len
    }
}

/// [MQTT5 SPECIFICATION](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901109)
#[derive(Debug, Clone, PartialEq, Default)]
pub struct PublishProperties<'a> {
    /// `1 (0x01)` Byte, Identifier of the Payload Format Indicator.
    ///
    /// Followed by the value of the Payload Format Indicator, either of:
    /// - `0 (0x00)` Byte Indicates that the Will Message is unspecified bytes, which is equivalent to not sending a Payload Format Indicator.
    /// - `1 (0x01)` Byte Indicates that the Will Message is UTF-8 Encoded Character Data.
    pub payload_format_indicator: Option<u8>,

    /// `2 (0x02)` Byte, Identifier of the Message Expiry Interval.
    ///
    /// Followed by the Four Byte Integer representing the Message Expiry Interval.
    /// If present, the Four Byte value is the lifetime of the Application Message in seconds.
    pub message_expiry_interval: Option<u32>,

    /// `3 (0x03)` Identifier of the Content Type.
    ///
    /// Followed by a UTF-8 Encoded String describing the content of the Application Message.
    pub content_type: Option<&'a str>,

    /// `8 (0x08)` Byte, Identifier of the Response Topic.
    ///
    /// Followed by a UTF-8 Encoded String which is used as the Topic Name for a response message.
    pub response_topic: Option<&'a str>,

    /// `9 (0x09)` Byte, Identifier of the Correlation Data.
    ///
    /// Followed by Binary Data.
    /// The Correlation Data is used by the sender of the Request Message to identify which request the Response Message is for when it is received.
    pub correlation_data: Option<&'a [u8]>,

    /// `11 (0x0B)`, Identifier of the Subscription Identifier.
    ///
    /// Followed by a Variable Byte Integer representing the identifier of the subscription.
    pub subscription_identifiers: Vec<usize, 32>,

    /// `35 (0x23)` Byte, Identifier of the Topic Alias.
    ///
    /// Followed by the Two Byte integer representing the Topic Alias value.
    pub topic_alias: Option<u16>,

    /// `38 (0x26)` Byte, Identifier of the User Property.
    ///
    /// Followed by a UTF-8 String Pair.
    pub user_properties: Vec<(&'a str, &'a str), 32>,
}

impl<'a> PublishProperties<'a> {
    pub fn read(buf: &'a [u8], offset: &mut usize) -> Result<Option<Self>, MqttError> {
        let mut properties = Self::default();

        let properties_len = buf[*offset];

        if properties_len == 0 {
            return Ok(None);
        }

        let mut cursor: usize = 1;

        while cursor < properties_len.into() {
            let mut prop_offset = *offset + cursor;
            let prop = buf[prop_offset];

            prop_offset += 1;
            cursor += 1;
            match property(prop)? {
                PropertyType::PayloadFormatIndicator => {
                    properties.payload_format_indicator = Some(buf[prop_offset]);
                    cursor += 1;
                }

                PropertyType::MessageExpiryInterval => {
                    properties.message_expiry_interval = Some(read_u32(buf, prop_offset));
                    cursor += 4;
                }

                PropertyType::TopicAlias => {
                    properties.topic_alias = Some(read_u16(buf, prop_offset));
                    cursor += 2;
                }

                PropertyType::ResponseTopic => {
                    let topic = read_str(buf, &mut prop_offset)?;
                    cursor += 2 + topic.len();
                    properties.response_topic = Some(topic);
                }

                PropertyType::CorrelationData => {
                    let data = read_bytes(buf, &mut prop_offset)?;
                    cursor += 2 + data.len();
                    properties.correlation_data = Some(data);
                }

                PropertyType::UserProperty => {
                    let key = read_str(buf, &mut prop_offset)?;
                    let value = read_str(buf, &mut prop_offset)?;
                    cursor += 2 + key.len() + 2 + value.len();
                    properties
                        .user_properties
                        .push((key, value))
                        .map_err(|_| MqttError::TooManyUserProperties)?;
                }

                PropertyType::SubscriptionIdentifier => {
                    let (id_len, sub_id) = varint_length(buf[prop_offset..].iter())?;

                    if sub_id == 0 {
                        return Err(MqttError::SubscriptionIdentifierZero);
                    }

                    cursor += id_len;

                    properties
                        .subscription_identifiers
                        .push(sub_id)
                        .map_err(|_| MqttError::TooManySubscriptionIdentifiers)?;
                }

                PropertyType::ContentType => {
                    let content_type = read_str(buf, &mut prop_offset)?;
                    cursor += 2 + content_type.len();
                    properties.content_type = Some(content_type);
                }

                prop => return Err(MqttError::InvalidPropertyType(prop)),
            }
        }

        *offset += properties.len();
        Ok(Some(properties))
    }

    pub fn write(&self, buf: &mut [u8], offset: &mut usize) -> Result<(), MqttError> {
        write_length(buf, offset, self.len())?;

        if let Some(payload) = self.payload_format_indicator {
            write_u8(buf, offset, PropertyType::PayloadFormatIndicator as u8);
            write_u8(buf, offset, payload);
        }

        if let Some(expiry_interval) = self.message_expiry_interval {
            write_u8(buf, offset, PropertyType::MessageExpiryInterval as u8);
            write_bytes(buf, offset, &expiry_interval.to_be_bytes());
        }

        if let Some(topic_alias) = self.topic_alias {
            write_u8(buf, offset, PropertyType::TopicAlias as u8);
            write_u16(buf, offset, topic_alias);
        }

        if let Some(response_topic) = &self.response_topic {
            write_u8(buf, offset, PropertyType::ResponseTopic as u8);
            write_bytes_with_len(buf, offset, response_topic.as_bytes());
        }

        if let Some(correlation_data) = self.correlation_data {
            write_u8(buf, offset, PropertyType::CorrelationData as u8);
            write_bytes_with_len(buf, offset, correlation_data);
        }

        for (key, value) in self.user_properties.iter() {
            write_u8(buf, offset, PropertyType::UserProperty as u8);
            write_bytes_with_len(buf, offset, key.as_bytes());
            write_bytes_with_len(buf, offset, value.as_bytes());
        }

        for id in self.subscription_identifiers.iter() {
            write_u8(buf, offset, PropertyType::SubscriptionIdentifier as u8);
            write_remaining_length(buf, offset, *id)?;
        }

        if let Some(content_type) = &self.content_type {
            write_u8(buf, offset, PropertyType::ContentType as u8);
            write_bytes_with_len(buf, offset, content_type.as_bytes());
        }

        Ok(())
    }

    fn len(&self) -> usize {
        let mut len = 0;

        if self.payload_format_indicator.is_some() {
            len += 1 + 1;
        }

        if self.message_expiry_interval.is_some() {
            len += 1 + 4;
        }

        if self.topic_alias.is_some() {
            len += 1 + 2;
        }

        if let Some(topic) = &self.response_topic {
            len += 1 + 2 + topic.len()
        }

        if let Some(data) = &self.correlation_data {
            len += 1 + 2 + data.len()
        }

        for (key, value) in self.user_properties.iter() {
            len += 1 + 2 + key.len() + 2 + value.len();
        }

        for id in self.subscription_identifiers.iter() {
            len += 1 + len_len(*id);
        }

        if let Some(content_type) = &self.content_type {
            len += 1 + 2 + content_type.len()
        }

        len
    }
}

#[cfg(test)]
mod test {
    use super::{Publish, PublishProperties};
    use crate::{control_packet::Header, Pid, QosPid};

    fn sample<'a>() -> Publish<'a> {
        let publish_properties = PublishProperties {
            payload_format_indicator: Some(1),
            message_expiry_interval: Some(4321),
            topic_alias: Some(100),
            response_topic: Some("topic"),
            correlation_data: Some(&[1, 2, 3, 4]),
            user_properties: heapless::Vec::<_, 32>::from_slice(&[("test", "test")]).unwrap(),
            subscription_identifiers: heapless::Vec::<_, 32>::from_slice(&[120, 16_384]).unwrap(),
            content_type: Some("test"),
        };

        Publish {
            dup: false,
            qospid: QosPid::ExactlyOnce(Pid::new(42)),
            retain: false,
            topic: "test",
            properties: Some(publish_properties),
            payload: &[b't', b'e', b's', b't'],
        }
    }

    fn sample_bytes() -> Vec<u8> {
        vec![
            0x34, // payload type
            0x40, // remaining len
            0x00, 0x04, 0x74, 0x65, 0x73, 0x74, // topic name
            0x00, 0x2a, // pkid
            0x33, // properties len
            0x01, 0x01, // payload format indicator
            0x02, 0x00, 0x00, 0x10, 0xe1, // message_expiry_interval
            0x23, 0x00, 0x64, // topic alias
            0x08, 0x00, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, // response topic
            0x09, 0x00, 0x04, 0x01, 0x02, 0x03, 0x04, // correlation_data
            0x26, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, 0x00, 0x04, 0x74, 0x65, 0x73,
            0x74, // user properties
            0x0b, 0x78, // subscription_identifier
            0x0b, 0x80, 0x80, 0x01, // subscription_identifier
            0x03, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, // content_type
            0x74, 0x65, 0x73, 0x74, // payload
        ]
    }

    #[test]
    fn publish_properties_decode() {
        let expected = &sample_bytes()[10..62];
        let mut buf = [0u8; 52];

        sample()
            .properties
            .unwrap()
            .write(&mut buf, &mut 0)
            .unwrap();
        assert_eq!(expected.len(), buf.len());

        assert_eq!(expected, buf);
    }

    #[test]
    fn publish_properties_encode() {
        let bytes = &sample_bytes()[10..62];
        let expected = sample().properties.unwrap();
        let actual = PublishProperties::read(bytes, &mut 0).unwrap().unwrap();
        assert_eq!(expected, actual);
    }

    #[test]
    fn publish_decode() {
        let expected = sample_bytes();
        let mut buf = [0u8; 66];

        sample().write(&mut buf, &mut 0).unwrap();
        assert_eq!(expected.len(), buf.len());

        assert_eq!(expected.as_slice(), buf);
    }

    #[test]
    fn publish_encode() {
        let bytes = &sample_bytes();
        let expected = sample();
        let header = Header::new(0x34).unwrap();
        let actual = Publish::read(header, bytes, &mut 2).unwrap();
        assert_eq!(expected, actual);
    }

    #[test]
    fn publish_encode2() {
        let bytes = [
            0x32, 0x5f, 0x00, 0x47, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2f, 0x6f, 0x6e,
            0x65, 0x6c, 0x69, 0x6e, 0x65, 0x5f, 0x72, 0x63, 0x67, 0x31, 0x39, 0x2f, 0x64, 0x65,
            0x76, 0x69, 0x63, 0x65, 0x2f, 0x6f, 0x6e, 0x65, 0x6c, 0x69, 0x6e, 0x65, 0x5f, 0x73,
            0x77, 0x69, 0x74, 0x63, 0x68, 0x31, 0x2f, 0x67, 0x70, 0x69, 0x6f, 0x2f, 0x64, 0x69,
            0x67, 0x69, 0x74, 0x61, 0x6c, 0x2f, 0x70, 0x6f, 0x65, 0x5f, 0x70, 0x6f, 0x72, 0x74,
            0x37, 0x2f, 0x73, 0x65, 0x74, 0x00, 0x08, 0x05, 0x02, 0x00, 0x00, 0x00, 0x1e, 0x7b,
            0x22, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x3a, 0x22, 0x4f, 0x6e, 0x22, 0x7d,
        ];

        let packet = crate::read_packet(&bytes).unwrap();

        let publish_properties = PublishProperties {
            message_expiry_interval: Some(30),
            ..PublishProperties::default()
        };

        let publish = Publish {
            dup: false,
            qospid: QosPid::AtLeastOnce(Pid::new(8)),
            retain: false,
            topic: "request/oneline_rcg19/device/oneline_switch1/gpio/digital/poe_port7/set",
            properties: Some(publish_properties),
            payload: &[
                0x7b, 0x22, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x3a, 0x22, 0x4f, 0x6e, 0x22, 0x7d,
            ],
        };

        let expected = crate::control_packet::Packet::Publish(publish);

        assert_eq!(expected, packet)
    }
}