matroskin 0.1.0

//! Length-delimited codec for the WhatsMiner TCP framing protocol.
//!
//! This module implements a [`tokio_util::codec`] compatible codec that handles
//! the WhatsMiner API v3 TCP data framing format. All communication with
//! WhatsMiner devices uses a simple length-prefixed binary protocol:
//!
//! - **Data frames:** `[4-byte LE length][JSON payload]`
//! - **Heartbeat frames:** `[0x00, 0x00, 0x00, 0x00]` (4 zero bytes) — keep-alive
//!
//! The framing protocol does not impose an artificial 8192-byte payload limit;
//! the example buffer size in the vendor documentation is non-normative.
//!
//! - ApiDoc: <https://apidoc.whatsminer.com/#api-TCP_Translate_Protocol-tcp_protocol>
//!
//! # Example
//!
//! ```rust,no_run
//! use futures_util::{SinkExt, StreamExt};
//! use matroskin::actor::codec::{WhatsMinerCodec, WhatsMinerFrame};
//! use tokio::net::TcpStream;
//! use tokio_util::codec::Framed;
//!
//! async fn example() {
//!     let stream = TcpStream::connect("192.168.1.100:4433").await.unwrap();
//!     let mut framed = Framed::new(stream, WhatsMinerCodec);
//!
//!     // Send a heartbeat keep-alive
//!     framed.send(WhatsMinerFrame::Heartbeat).await.unwrap();
//!
//!     // Send a JSON command
//!     framed
//!         .send(WhatsMinerFrame::Data(r#"{"cmd":"get_dev_detail"}"#.into()))
//!         .await
//!         .unwrap();
//!
//!     // Read a response frame
//!     while let Some(frame) = framed.next().await {
//!         match frame.unwrap() {
//!             WhatsMinerFrame::Heartbeat => println!("heartbeat received"),
//!             WhatsMinerFrame::Data(json) => println!("response: {:?}", json),
//!         }
//!     }
//! }
//! ```
//!
//! - Item: [`WhatsMinerFrame`], [`WhatsMinerCodec`]

use bytes::{Buf, BufMut, Bytes, BytesMut};
use tokio_util::codec::{Decoder, Encoder};
use tracing::trace;

use crate::error::{Error, Result};

/// Frame types for the WhatsMiner TCP protocol.
///
/// Each variant corresponds to one of the two valid frame types defined by
/// the WhatsMiner API v3 TCP framing specification:
///
/// - [`Heartbeat`](WhatsMinerFrame::Heartbeat): a zero-length frame (`[0x00; 4]`)
///   used as a keep-alive signal.
/// - [`Data`](WhatsMinerFrame::Data): an opaque length-prefixed payload.
#[derive(Debug, Clone, PartialEq)]
pub enum WhatsMinerFrame {
    Heartbeat,
    Data(Bytes),
}

/// Length-delimited codec for the WhatsMiner TCP protocol.
///
/// The codec only frames opaque bytes. JSON decoding belongs to the response
/// consumer, while raw streaming is handled by the actor's transport owner.
pub struct WhatsMinerCodec;

impl Decoder for WhatsMinerCodec {
    type Item = WhatsMinerFrame;
    type Error = Error;

    /// Attempt to decode a complete [`WhatsMinerFrame`] from the supplied buffer.
    ///
    /// Returns `Ok(None)` if the buffer does not yet contain enough data to
    /// produce a full frame, in which case the codec will wait for more data
    /// to arrive on the stream.
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>> {
        if src.len() < 4 {
            return Ok(None);
        }

        let length = u32::from_le_bytes([src[0], src[1], src[2], src[3]]) as usize;
        if length == 0 {
            src.advance(4);
            trace!("Heartbeat frame decoded");
            return Ok(Some(WhatsMinerFrame::Heartbeat));
        }
        if src.len() - 4 < length {
            return Ok(None);
        }

        src.advance(4);
        let payload = src.split_to(length).freeze();
        trace!(frame_len = length, "Data frame decoded");
        Ok(Some(WhatsMinerFrame::Data(payload)))
    }
}

impl Encoder<WhatsMinerFrame> for WhatsMinerCodec {
    type Error = Error;

    fn encode(&mut self, item: WhatsMinerFrame, dst: &mut BytesMut) -> Result<()> {
        match item {
            WhatsMinerFrame::Heartbeat => dst.put_u32_le(0),
            WhatsMinerFrame::Data(payload) => {
                let len = u32::try_from(payload.len())
                    .map_err(|_| Error::ResponseTooLarge(payload.len()))?;
                dst.reserve(4 + payload.len());
                dst.put_u32_le(len);
                dst.extend_from_slice(&payload);
            }
        }
        Ok(())
    }
}

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

    /// Decode four zero bytes into a [`WhatsMinerFrame::Heartbeat`].
    #[test]
    fn test_decode_heartbeat() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::from(&[0x00, 0x00, 0x00, 0x00][..]);
        let frame = codec.decode(&mut buf).expect("decode should succeed");
        assert_eq!(frame, Some(WhatsMinerFrame::Heartbeat));
        assert!(buf.is_empty(), "buffer should be fully consumed");
    }

    /// Decode a valid JSON data frame.
    #[test]
    fn test_decode_data_frame() {
        let mut codec = WhatsMinerCodec;
        let payload = r#"{"cmd":"get_dev_detail"}"#;
        let payload_bytes = payload.as_bytes();
        let len = payload_bytes.len() as u32;

        let mut buf = BytesMut::with_capacity(4 + payload_bytes.len());
        buf.put_u32_le(len);
        buf.put_slice(payload_bytes);

        let frame = codec.decode(&mut buf).expect("decode should succeed");
        assert_eq!(
            frame,
            Some(WhatsMinerFrame::Data(payload.to_string().into()))
        );
        assert!(buf.is_empty(), "buffer should be fully consumed");
    }

    /// Fewer than 4 bytes in the buffer → decoder returns `None` (waiting for more data).
    #[test]
    fn test_decode_partial_header() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::from(&[0x01, 0x00][..]);
        let frame = codec.decode(&mut buf).expect("decode should succeed");
        assert_eq!(frame, None);
    }

    /// Length header received but not enough payload bytes → decoder returns `None`.
    #[test]
    fn test_decode_partial_payload() {
        let mut codec = WhatsMinerCodec;
        let payload = "hello";

        let mut buf = BytesMut::with_capacity(6);
        buf.put_u32_le(payload.len() as u32);
        buf.put_slice(b"hel"); // only 3 of 5 bytes

        let frame = codec.decode(&mut buf).expect("decode should succeed");
        assert_eq!(frame, None);
    }

    /// Payloads larger than the example C buffer are valid protocol frames.
    #[test]
    fn test_decode_large_payload() {
        let mut codec = WhatsMinerCodec;
        let payload = vec![b'x'; 8193];
        let mut buf = BytesMut::new();
        buf.put_u32_le(payload.len() as u32);
        buf.extend_from_slice(&payload);
        let frame = codec.decode(&mut buf).expect("decode should succeed");
        assert_eq!(frame, Some(WhatsMinerFrame::Data(Bytes::from(payload))));
    }

    /// Encode a [`WhatsMinerFrame::Heartbeat`] → produces `[0x00; 4]`.
    #[test]
    fn test_encode_heartbeat() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::new();
        codec
            .encode(WhatsMinerFrame::Heartbeat, &mut buf)
            .expect("encode should succeed");
        assert_eq!(&buf[..], &[0x00, 0x00, 0x00, 0x00]);
    }

    /// Encode a [`WhatsMinerFrame::Data`] → produces `[len LE][payload bytes]`.
    #[test]
    fn test_encode_data() {
        let mut codec = WhatsMinerCodec;
        let payload = r#"{"cmd":"get_dev_detail"}"#;
        let mut buf = BytesMut::new();
        codec
            .encode(WhatsMinerFrame::Data(payload.to_string().into()), &mut buf)
            .expect("encode should succeed");

        let expected_len = payload.len();
        let mut expected = BytesMut::with_capacity(4 + payload.len());
        expected.put_u32_le(expected_len as u32);
        expected.put_slice(payload.as_bytes());

        assert_eq!(buf, expected);
    }

    /// Encode a frame then decode it → round-trips to the same value.
    #[test]
    fn test_roundtrip() {
        let mut encode_codec = WhatsMinerCodec;
        let mut decode_codec = WhatsMinerCodec;

        // --- Heartbeat round-trip ---
        let mut buf = BytesMut::new();
        encode_codec
            .encode(WhatsMinerFrame::Heartbeat, &mut buf)
            .expect("encode heartbeat");
        let frame = decode_codec
            .decode(&mut buf)
            .expect("decode heartbeat")
            .expect("should produce a frame");
        assert_eq!(frame, WhatsMinerFrame::Heartbeat);

        // --- Data round-trip ---
        let payload = r#"{"STATUS":"S","ip":"192.168.1.100"}"#;
        let mut buf = BytesMut::new();
        encode_codec
            .encode(WhatsMinerFrame::Data(payload.to_string().into()), &mut buf)
            .expect("encode data");
        let frame = decode_codec
            .decode(&mut buf)
            .expect("decode data")
            .expect("should produce a frame");
        assert_eq!(frame, WhatsMinerFrame::Data(payload.to_string().into()));
    }

    /// Split frame arrival: data arrives across multiple partial chunks.
    #[test]
    fn test_split_frame() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::new();

        // First chunk: partial 4-byte prefix (only 2 bytes)
        buf.extend_from_slice(&[5, 0]);
        assert_eq!(codec.decode(&mut buf).unwrap(), None);

        // Second chunk: remainder of length prefix
        buf.extend_from_slice(&[0, 0]);
        assert_eq!(codec.decode(&mut buf).unwrap(), None);

        // Third chunk: partial payload (3 bytes of 5)
        buf.extend_from_slice(b"hel");
        assert_eq!(codec.decode(&mut buf).unwrap(), None);

        // Final chunk: remaining 2 bytes
        buf.extend_from_slice(b"lo");
        let frame = codec.decode(&mut buf).unwrap();
        assert_eq!(
            frame,
            Some(WhatsMinerFrame::Data(Bytes::from_static(b"hello")))
        );
        assert!(buf.is_empty());
    }

    /// Multiple consecutive frames in a single buffer decode sequentially.
    #[test]
    fn test_multiple_frames_in_one_buffer() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::new();

        // Frame 1: Heartbeat
        buf.put_u32_le(0);
        // Frame 2: Data "first"
        buf.put_u32_le(5);
        buf.put_slice(b"first");
        // Frame 3: Heartbeat
        buf.put_u32_le(0);
        // Frame 4: Data "second"
        buf.put_u32_le(6);
        buf.put_slice(b"second");

        assert_eq!(
            codec.decode(&mut buf).unwrap(),
            Some(WhatsMinerFrame::Heartbeat)
        );
        assert_eq!(
            codec.decode(&mut buf).unwrap(),
            Some(WhatsMinerFrame::Data(Bytes::from_static(b"first")))
        );
        assert_eq!(
            codec.decode(&mut buf).unwrap(),
            Some(WhatsMinerFrame::Heartbeat)
        );
        assert_eq!(
            codec.decode(&mut buf).unwrap(),
            Some(WhatsMinerFrame::Data(Bytes::from_static(b"second")))
        );
        assert_eq!(codec.decode(&mut buf).unwrap(), None);
        assert!(buf.is_empty());
    }

    /// Trailing data preservation: unconsumed bytes remain intact for subsequent frames.
    #[test]
    fn test_trailing_data_preservation() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::new();

        // Complete frame followed by 3 trailing bytes of next frame
        buf.put_u32_le(4);
        buf.put_slice(b"ping");
        buf.extend_from_slice(&[0xAA, 0xBB, 0xCC]);

        let frame = codec.decode(&mut buf).unwrap();
        assert_eq!(
            frame,
            Some(WhatsMinerFrame::Data(Bytes::from_static(b"ping")))
        );
        assert_eq!(&buf[..], &[0xAA, 0xBB, 0xCC]);
    }

    /// Codec must not reserve large capacity based on length prefix alone.
    #[test]
    fn test_no_allocation_from_length_only() {
        let mut codec = WhatsMinerCodec;
        let mut buf = BytesMut::with_capacity(16);
        let initial_cap = buf.capacity();
        // 1 GB declared length
        buf.put_u32_le(1_000_000_000);
        let res = codec.decode(&mut buf).unwrap();
        assert_eq!(res, None);
        assert_eq!(buf.capacity(), initial_cap);
    }
}