gmqtt 0.1.1

use core::convert::TryFrom;
use heapless::{String, Vec};

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

use super::{property, PropertyType};

///The SUBSCRIBE packet is sent from the Client to the Server to create one or more Subscriptions.
///Each Subscription registers a Client’s interest in one or more Topics.
///The Server sends PUBLISH packets to the Client to forward Application Messages that were published to Topics that match these Subscriptions.
///The SUBSCRIBE packet also specifies (for each Subscription) the maximum QoS with which the Server can send Application Messages to the Client.
///
/// [MQTT5 SPECIFICATION](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901161)
#[derive(Debug, Clone, PartialEq)]
pub struct Subscribe<'a> {
    pub pid: Pid,

    /// The Payload of a SUBSCRIBE packet contains a list of Topic Filters indicating the Topics to which the Client wants to subscribe.
    /// The Topic Filters MUST be a UTF-8 Encoded String.
    /// Each Topic Filter is followed by a Subscription Options byte.
    pub filters: Vec<SubscribeFilter, 4>,
    pub properties: Option<SubscribeProperties<'a>>,
}

impl<'a> Subscribe<'a> {
    pub fn new<S: Into<String<128>>>(filter: S, qos: QoS, pid: Pid) -> Result<Self, MqttError> {
        let filter = SubscribeFilter {
            filter: filter.into(),
            qos,
            nolocal: false,
            preserve_retain: false,
            retain_forward_rule: RetainForwardRule::OnEverySubscribe,
        };

        let mut filters = Vec::new();
        filters
            .push(filter)
            .map_err(|_| MqttError::TooManyFilters)?;

        Ok(Subscribe {
            pid,
            filters,
            properties: None,
        })
    }

    pub fn read(buf: &'a [u8], offset: &mut usize) -> Result<Self, MqttError> {
        let pid = Pid::read(buf, offset)?;
        let properties = SubscribeProperties::read(buf, offset)?;

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

        let remaining_len = &buf[*offset..].len();

        let mut filters = heapless::Vec::<SubscribeFilter, 4>::new();
        let mut cursor = 0;
        while cursor < *remaining_len {
            let filter = String::from(read_str(&buf[*offset..], &mut cursor)?);
            let options = buf.last().expect("Buffer cannot be empty");
            cursor += 1;

            let requested_qos = options & 0b0000_0011;

            let nolocal = options >> 2 & 0b0000_0001 != 0;
            let preserve_retain = options >> 3 & 0b0000_0001 != 0;

            let retain_forward_rule = (options >> 4) & 0b0000_0011;
            let retain_forward_rule = match retain_forward_rule {
                0 => RetainForwardRule::OnEverySubscribe,
                1 => RetainForwardRule::OnNewSubscribe,
                2 => RetainForwardRule::Never,
                rule => return Err(MqttError::InvalidRetainForwardRule(rule)),
            };

            filters
                .push(SubscribeFilter {
                    filter,
                    qos: QoS::try_from(requested_qos)?,
                    nolocal,
                    preserve_retain,
                    retain_forward_rule,
                })
                .map_err(|_| MqttError::TooManyFilters)?;
        }

        let subscribe = Subscribe {
            pid,
            filters,
            properties,
        };

        Ok(subscribe)
    }

    pub fn write(&self, buf: &mut [u8], offset: &mut usize) -> Result<usize, MqttError> {
        check_remaining(buf, offset, self.len() + 1)?;
        // Subscribe Command (0x82)
        let header: u8 = 0x82;

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

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

        for filter in &self.filters {
            filter.write(buf, offset);
        }

        Ok(write_len)
    }

    fn len(&self) -> usize {
        let mut len = 2 + self.filters.iter().fold(0, |s, t| s + t.len());

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

        len
    }
}

/// [MQTT5 SPECIFICATION](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901164)
#[derive(Debug, Clone, PartialEq, Default)]
pub struct SubscribeProperties<'a> {
    /// `11 (0x0B)` Byte, Identifier of the Subscription Identifier.
    ///
    /// Followed by a Variable Byte Integer representing the identifier of the subscription.
    pub id: Option<usize>,

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

impl<'a> SubscribeProperties<'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::SubscriptionIdentifier => {
                    let (id_len, sub_id) = varint_length(buf[prop_offset..].iter())?;

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

                    cursor += id_len;

                    properties.id = Some(sub_id);
                }

                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)?;
                }

                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(id) = self.id {
            write_u8(buf, offset, PropertyType::SubscriptionIdentifier as u8);
            write_remaining_length(buf, offset, id)?;
        }

        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());
        }

        Ok(())
    }

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

        if let Some(id) = &self.id {
            len += 1 + len_len(*id);
        }

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

        len
    }
}

///  Subscription filter
#[derive(Debug, Clone, PartialEq)]
pub struct SubscribeFilter {
    pub filter: String<128>,
    pub qos: QoS,
    pub nolocal: bool,
    pub preserve_retain: bool,
    pub retain_forward_rule: RetainForwardRule,
}

impl SubscribeFilter {
    pub fn new<S: Into<String<128>>>(filter: S, qos: QoS) -> SubscribeFilter {
        SubscribeFilter {
            filter: filter.into(),
            qos,
            nolocal: false,
            preserve_retain: false,
            retain_forward_rule: RetainForwardRule::OnEverySubscribe,
        }
    }

    pub fn set_nolocal(&mut self, flag: bool) -> &mut Self {
        self.nolocal = flag;
        self
    }

    pub fn set_preserve_retain(&mut self, flag: bool) -> &mut Self {
        self.preserve_retain = flag;
        self
    }

    pub fn set_retain_forward_rule(&mut self, rule: RetainForwardRule) -> &mut Self {
        self.retain_forward_rule = rule;
        self
    }

    pub fn write(&self, buf: &mut [u8], offset: &mut usize) {
        let mut options = 0;
        options |= self.qos as u8;

        if self.nolocal {
            options |= 1 << 2;
        }

        if self.preserve_retain {
            options |= 1 << 3;
        }

        match self.retain_forward_rule {
            RetainForwardRule::OnEverySubscribe => options |= 0 << 4,
            RetainForwardRule::OnNewSubscribe => options |= 1 << 4,
            RetainForwardRule::Never => options |= 2 << 4,
        }

        write_bytes_with_len(buf, offset, self.filter.as_bytes());
        write_u8(buf, offset, options);
    }

    fn len(&self) -> usize {
        // filter len + filter + options
        2 + self.filter.len() + 1
    }
}

#[derive(Debug, Clone, PartialEq)]
pub enum RetainForwardRule {
    /// Send retained messages at the time of the subscribe
    OnEverySubscribe,

    /// Send retained messages at subscribe only if the subscription does not currently exist
    OnNewSubscribe,

    /// Do not send retained messages at the time of the subscribe
    Never,
}

#[cfg(test)]
mod test {
    use super::{RetainForwardRule, Subscribe, SubscribeFilter, SubscribeProperties};
    use crate::{Pid, QoS};

    fn sample<'a>() -> Subscribe<'a> {
        let subscribe_properties = SubscribeProperties {
            id: Some(100),
            user_properties: heapless::Vec::<_, 8>::from_slice(&[("test", "test")]).unwrap(),
        };

        let mut filter = SubscribeFilter::new("hello", QoS::AtLeastOnce);
        filter
            .set_nolocal(true)
            .set_preserve_retain(true)
            .set_retain_forward_rule(RetainForwardRule::Never);

        Subscribe {
            pid: Pid::new(42),
            filters: heapless::Vec::<_, 4>::from_slice(&[filter]).unwrap(),
            properties: Some(subscribe_properties),
        }
    }

    fn sample_bytes() -> Vec<u8> {
        vec![
            0x82, // packet type
            0x1a, // remaining length
            0x00, 0x2a, // pkid
            0x0f, // properties len
            0x0b, 0x64, // subscription identifier
            0x26, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, 0x00, 0x04, 0x74, 0x65, 0x73,
            0x74, // user properties
            0x00, 0x05, 0x68, 0x65, 0x6c, 0x6c, 0x6f, // filter
            0x2d, // options
        ]
    }

    #[test]
    fn subscribe_properties_decode() {
        let expected = &sample_bytes()[4..20];
        let mut buf = [0u8; 16];

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

        assert_eq!(expected, buf);
    }

    #[test]
    fn subscribe_properties_encode() {
        let bytes = &sample_bytes()[4..20];
        let expected = sample().properties.unwrap();
        let actual = SubscribeProperties::read(bytes, &mut 0).unwrap().unwrap();
        assert_eq!(expected, actual);
    }

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

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

        assert_eq!(expected, buf);
    }

    #[test]
    fn subscribe_encode() {
        let bytes = &sample_bytes();
        let expected = sample();
        let actual = Subscribe::read(bytes, &mut 2).unwrap();
        assert_eq!(expected, actual);
    }

    fn sample2<'a>() -> Subscribe<'a> {
        let filter = SubscribeFilter::new(heapless::String::from("hello/world"), QoS::AtLeastOnce);
        Subscribe {
            pid: Pid::new(42),
            filters: heapless::Vec::<_, 4>::from_slice(&[filter]).unwrap(),
            properties: None,
        }
    }

    fn sample_bytes2() -> Vec<u8> {
        vec![
            0x82, 0x11, 0x00, 0x2a, 0x00, 0x00, 0x0b, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2f, 0x77,
            0x6f, 0x72, 0x6c, 0x64, 0x01,
        ]
    }

    #[test]
    fn subscribe_decode2() {
        let expected = sample_bytes2();
        let mut buf = [0u8; 19];

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

        assert_eq!(expected, buf);
    }

    #[test]
    fn subscribe_encode2() {
        let bytes = &sample_bytes2();
        let expected = sample2();
        let actual = Subscribe::read(bytes, &mut 2).unwrap();
        assert_eq!(expected, actual);
    }
}