Skip to main content

sedsnet/
packet.rs

1//! Telemetry packet core type and formatting helpers.
2//!
3//! [`Packet`] is the main payload-bearing type. It:
4//! - holds sender, endpoints, timestamp and raw payload bytes,
5//! - validates payload sizes and encodings against the schema from `message_meta`,
6//! - supports pretty printing (header + decoded values) for debugging/logging,
7//! - uses [`SmallPayload`] internally to keep small messages on the stack.
8
9use crate::config::{StandardSmallPayload, runtime_device_identifier, runtime_string_precision};
10use crate::queue::ByteCost;
11use crate::{
12    MessageClass, MessageDataType, MessageElement, TelemetryError, TelemetryResult,
13    config::{DataEndpoint, DataType},
14    data_type_size, get_data_type, get_message_name, message_meta,
15    router::LeBytes,
16};
17use crate::{impl_data_as_prim, impl_from_prim_slices, impl_ledecode_auto};
18use alloc::{string::String, string::ToString, sync::Arc, vec, vec::Vec};
19use core::any::TypeId;
20use core::fmt::{Formatter, Write};
21use core::sync::atomic::{AtomicU32, Ordering};
22// ============================================================================
23// Constants
24// ============================================================================
25
26/// Threshold (in ms since boot/epoch) above which timestamps are treated as
27/// Unix epoch time rather than an uptime counter.
28///
29/// Anything smaller is formatted as an uptime-style duration; larger values
30/// are formatted as a UTC date-time.
31const EPOCH_MS_THRESHOLD: u64 = 1_000_000_000_000; // clearly not an uptime counter
32
33/// Default starting capacity for human-readable strings.
34const DEFAULT_STRING_CAPACITY: usize = 96;
35
36/// Process-local packet nonce generator used to keep same-timestamp packets
37/// distinct for recent-ID dedupe even when their logical payload matches.
38static PACKET_NONCE_COUNTER: AtomicU32 = AtomicU32::new(0xA5C3_1F27);
39
40#[inline]
41fn next_packet_nonce() -> u16 {
42    let mut x = PACKET_NONCE_COUNTER.fetch_add(0x9E37_79B9, Ordering::Relaxed);
43    x ^= x << 13;
44    x ^= x >> 17;
45    x ^= x << 5;
46    (x as u16) | 1
47}
48
49// ============================================================================
50// Packet
51// ============================================================================
52
53/// Payload-bearing packet (safe, validated, shareable).
54///
55/// This is the primary data structure passed around inside the crate and
56/// across FFI boundaries (via views / wrappers).
57#[derive(Clone, Debug, PartialEq, Eq)]
58pub struct Packet {
59    /// Logical message type (schema selector).
60    ty: DataType,
61
62    /// Size of the payload in bytes.
63    ///
64    /// This is cached and must match `payload.len()`. [`Packet::validate`]
65    /// checks the invariant.
66    data_size: usize,
67
68    /// Logical sender identifier (e.g. device or subsystem name).
69    sender: Arc<str>,
70
71    /// Destination endpoints for this message.
72    endpoints: Arc<[DataEndpoint]>,
73
74    /// Timestamp in milliseconds.
75    ///
76    /// - If `< EPOCH_MS_THRESHOLD`, treated as an uptime counter and formatted
77    ///   like `12m 34s 567ms`.
78    /// - If `>= EPOCH_MS_THRESHOLD`, treated as Unix epoch ms and formatted
79    ///   as `YYYY-MM-DD HH:MM:SS.mmmZ`.
80    timestamp: u64,
81
82    /// Small per-packet nonce carried on the wire and folded into packet ID
83    /// calculation so otherwise-identical packets emitted in the same
84    /// millisecond do not collapse in recent-ID dedupe.
85    nonce: u16,
86
87    /// Raw payload bytes, stored via [`SmallPayload`] for small/large optimization.
88    payload: StandardSmallPayload,
89
90    /// Inline wire-format shape used when this packet came from a migration-safe frame.
91    wire_shape: Option<MessageElement>,
92
93    /// Explicit destination sender hashes frozen into the wire-format contract.
94    wire_target_senders: Option<Arc<[u64]>>,
95}
96
97// ============================================================================
98// Internal helpers for validation / formatting
99// ============================================================================
100
101/// Effective element width (in bytes) for the given message data type.
102///
103/// For numeric/bool types this is the true width of one element.
104/// For string/hex we return 1 to treat the payload as a byte stream when
105/// checking dynamic length multiples.
106#[inline]
107const fn element_width(dt: MessageDataType) -> usize {
108    match dt {
109        MessageDataType::UInt8 | MessageDataType::Int8 | MessageDataType::Bool => 1,
110        MessageDataType::UInt16 | MessageDataType::Int16 => 2,
111        MessageDataType::UInt32 | MessageDataType::Int32 | MessageDataType::Float32 => 4,
112        MessageDataType::UInt64 | MessageDataType::Int64 | MessageDataType::Float64 => 8,
113        MessageDataType::UInt128 | MessageDataType::Int128 => 16,
114        // For String/Hex we treat width as 1 (byte granularity) when checking dynamic multiples.
115        MessageDataType::String | MessageDataType::Binary => 1,
116        MessageDataType::NoData => 0,
117    }
118}
119
120// Simple 64-bit hash used for packet IDs (no_std friendly).
121/// Not cryptographic; just a decent mix for deduplication.
122/// Takes an initial hash value and a byte slice, returns the updated hash.
123#[inline]
124pub fn hash_bytes_u64(mut h: u64, bytes: &[u8]) -> u64 {
125    // Arbitrary odd constant (not FNV, but good enough for dedupe).
126    const PRIME: u64 = 0x9E37_79B1;
127    for &b in bytes {
128        h ^= b as u64;
129        h = h.wrapping_mul(PRIME);
130        // a tiny extra mix
131        h ^= h >> 27;
132    }
133    h
134}
135
136/// Stable compact source address derived from a logical sender name.
137///
138/// Network-master address assignment can override the discovery mapping at the
139/// routing layer, but standalone packets need a deterministic address so the
140/// canonical wire header never has to carry the sender string.
141#[inline]
142pub(crate) fn sender_address_u32(sender: &str) -> u32 {
143    // Unresolved wire senders retain their numeric identity through repacking
144    // and receipts. Hashing the alias text would invent a different sender.
145    if let Some(address) = sender
146        .strip_prefix("@addr:")
147        .and_then(|s| s.parse::<u32>().ok())
148    {
149        return address;
150    }
151    let raw = hash_bytes_u64(0xA6D3_8C21_4B7F_19E5, sender.as_bytes()) as u32;
152    if raw == 0 { 1 } else { raw }
153}
154
155/// Validate the payload of a dynamic-length message.
156///
157/// - For `String`: trims trailing NULs for validation and ensures UTF-8 (if non-empty).
158/// - For `Hex`: no additional validation.
159/// - For numerics/bool: ensures the length is a multiple of the element width.
160#[inline]
161fn validate_dynamic_len_and_content_for_element(
162    element: MessageElement,
163    bytes: &[u8],
164) -> TelemetryResult<()> {
165    match element.data_type() {
166        MessageDataType::String => {
167            let end = bytes
168                .iter()
169                .rposition(|&b| b != 0)
170                .map(|i| i + 1)
171                .unwrap_or(0);
172            if end > 0 {
173                core::str::from_utf8(&bytes[..end]).map_err(|_| TelemetryError::InvalidUtf8)?;
174            }
175            Ok(())
176        }
177        MessageDataType::Binary => Ok(()),
178        dt => {
179            let w = element_width(dt);
180            if w == 0 || !bytes.len().is_multiple_of(w) {
181                return Err(TelemetryError::SizeMismatch {
182                    expected: w,
183                    got: bytes.len(),
184                });
185            }
186            Ok(())
187        }
188    }
189}
190
191// ============================================================================
192// Decode bytes trait
193// ============================================================================
194
195trait LeDecode: Sized {
196    const WIDTH: usize;
197    fn from_le(slice: &[u8]) -> Self;
198}
199
200impl_ledecode_auto!(f32);
201impl_ledecode_auto!(f64);
202
203impl_ledecode_auto!(u16);
204impl_ledecode_auto!(u32);
205impl_ledecode_auto!(u64);
206impl_ledecode_auto!(u128);
207
208impl_ledecode_auto!(i16);
209impl_ledecode_auto!(i32);
210impl_ledecode_auto!(i64);
211impl_ledecode_auto!(i128);
212
213// special 1-byte cases
214impl_ledecode_auto!(u8);
215impl_ledecode_auto!(i8);
216
217// ============================================================================
218// Packet impl
219// ============================================================================
220
221impl Packet {
222    /// Validate `payload` against the provided schema element.
223    ///
224    /// This helper is shared by normal packet construction and by
225    /// `new_with_wire_contract(...)`. The latter matters because a packet that
226    /// was already packed may need to keep using the element shape it was
227    /// written with even if the local runtime registry has since changed.
228    ///
229    /// # Parameters
230    /// - `element`: Effective schema element to validate against. This may come
231    ///   from the current runtime registry or from the inline wire contract.
232    /// - `payload`: Raw payload bytes to validate.
233    ///
234    /// # Returns
235    /// - `Ok(())` when the payload is compatible with `element`.
236    /// - `Err(TelemetryError)` when the payload length or encoding does not
237    ///   match the required shape.
238    #[inline]
239    pub(crate) fn validate_payload_against_element(
240        element: MessageElement,
241        payload: &[u8],
242    ) -> TelemetryResult<()> {
243        match element {
244            MessageElement::Static(need, dt, _) => {
245                let need = need * data_type_size(dt);
246                if payload.len() != need {
247                    return Err(TelemetryError::SizeMismatch {
248                        expected: need,
249                        got: payload.len(),
250                    });
251                }
252                Ok(())
253            }
254            MessageElement::Dynamic(_, _) => {
255                validate_dynamic_len_and_content_for_element(element, payload)
256            }
257        }
258    }
259
260    /// Test helper for constructing frozen wire-contract metadata from borrowed inputs.
261    ///
262    /// This constructor is used by unpacking paths that may carry a
263    /// migration-safe wire contract. The contract can supply:
264    ///
265    /// - `wire_shape`: an inline `MessageElement` that keeps payload decoding
266    ///   stable even if the local runtime schema changed after the packet was
267    ///   packed.
268    /// - `wire_target_senders`: a frozen list of destination-holder sender
269    ///   hashes that routers and relays use to keep in-flight forwarding bound
270    ///   to the originally intended remote holders.
271    ///
272    /// # Parameters
273    /// - `ty`: Logical message type ID carried by the frame.
274    /// - `endpoints`: Logical destination endpoints from the endpoint bitmap.
275    /// - `sender`: Logical sender string decoded from the frame.
276    /// - `timestamp`: Wire timestamp in milliseconds.
277    /// - `payload`: Payload bytes after any decompression.
278    /// - `wire_shape`: Optional inline payload shape carried by the wire
279    ///   contract. When present, validation uses this value instead of the
280    ///   current runtime schema entry.
281    /// - `wire_target_senders`: Frozen destination-holder hashes carried by the
282    ///   wire contract. These are routing metadata, not application payload.
283    ///
284    /// # Returns
285    /// - `Ok(Packet)` when all packet invariants and payload validation pass.
286    /// - `Err(TelemetryError)` when endpoints are empty or the payload is
287    ///   incompatible with the effective schema element.
288    #[inline]
289    #[allow(clippy::too_many_arguments)]
290    #[cfg(test)]
291    pub(crate) fn new_with_wire_contract(
292        ty: DataType,
293        endpoints: &[DataEndpoint],
294        sender: &str,
295        timestamp: u64,
296        nonce: u16,
297        payload: Arc<[u8]>,
298        wire_shape: Option<MessageElement>,
299        wire_target_senders: Arc<[u64]>,
300    ) -> TelemetryResult<Self> {
301        if endpoints.is_empty() {
302            return Err(TelemetryError::EmptyEndpoints);
303        }
304
305        let element = wire_shape.unwrap_or_else(|| message_meta(ty).element);
306        Self::validate_payload_against_element(element, &payload)?;
307
308        Ok(Self {
309            ty,
310            data_size: payload.len(),
311            sender: sender.into(),
312            endpoints: Arc::<[DataEndpoint]>::from(endpoints),
313            timestamp,
314            nonce,
315            payload: StandardSmallPayload::from_arc(payload),
316            wire_shape,
317            wire_target_senders: (!wire_target_senders.is_empty()).then_some(wire_target_senders),
318        })
319    }
320
321    /// Preserve already-owned wire metadata rather than allocating a second copy.
322    #[inline]
323    #[allow(clippy::too_many_arguments)]
324    pub(crate) fn new_with_owned_wire_contract(
325        ty: DataType,
326        endpoints: Arc<[DataEndpoint]>,
327        sender: Arc<str>,
328        timestamp: u64,
329        nonce: u16,
330        payload: StandardSmallPayload,
331        wire_shape: Option<MessageElement>,
332        wire_target_senders: Option<Arc<[u64]>>,
333    ) -> TelemetryResult<Self> {
334        if endpoints.is_empty() {
335            return Err(TelemetryError::EmptyEndpoints);
336        }
337        let element = wire_shape.unwrap_or_else(|| message_meta(ty).element);
338        Self::validate_payload_against_element(element, &payload)?;
339        Ok(Self {
340            ty,
341            data_size: payload.len(),
342            sender,
343            endpoints,
344            timestamp,
345            nonce,
346            payload,
347            wire_shape,
348            wire_target_senders,
349        })
350    }
351
352    /// Create a packet from a raw payload, validating against `message_meta(ty)`.
353    ///
354    /// Checks:
355    /// - `endpoints` is non-empty.
356    /// - For static element count:
357    ///   - `payload.len() == element_count * data_type_size(get_data_type(ty))`.
358    /// - For dynamic:
359    ///   - Length and encoding are validated by [`validate_dynamic_len_and_content`].
360    /// # Arguments
361    /// - `ty`: logical message type (schema selector).
362    /// - `endpoints`: destination endpoint list (must be non-empty).
363    /// - `sender`: logical sender identifier (e.g. device or subsystem name).
364    /// - `timestamp`: timestamp in milliseconds.
365    /// - `payload`: raw payload bytes.
366    /// # Returns
367    /// - `Ok(Packet)` if validation passes.
368    /// - `Err(TelemetryError)` if validation fails.
369    /// # Errors
370    /// - [`TelemetryError::EmptyEndpoints`] if `endpoints` is empty.
371    /// - [`TelemetryError::SizeMismatch`] if the payload size does not match
372    ///   the expected size for static element counts, or is not a multiple
373    ///   of the element width for dynamic types.
374    /// - [`TelemetryError::InvalidUtf8`] if the payload is a string
375    ///   type and is not valid UTF-8 .
376    pub fn new(
377        ty: DataType,
378        endpoints: &[DataEndpoint],
379        sender: &str,
380        timestamp: u64,
381        payload: Arc<[u8]>,
382    ) -> TelemetryResult<Self> {
383        Self::new_with_nonce(
384            ty,
385            endpoints,
386            sender,
387            timestamp,
388            next_packet_nonce(),
389            payload,
390        )
391    }
392
393    /// Build a local publication while retaining owned sender/schema metadata
394    /// and an already-inline payload. Wire identity and validation are identical.
395    pub(crate) fn new_local(
396        ty: DataType,
397        endpoints: Arc<[DataEndpoint]>,
398        sender: Arc<str>,
399        timestamp: u64,
400        payload: StandardSmallPayload,
401    ) -> TelemetryResult<Self> {
402        Self::new_with_owned_wire_contract(
403            ty,
404            endpoints,
405            sender,
406            timestamp,
407            next_packet_nonce(),
408            payload,
409            None,
410            None,
411        )
412    }
413
414    /// Create a packet with an explicit nonce value.
415    ///
416    /// This is mainly useful when callers need stable byte-for-byte wire output
417    /// across repeated constructions, such as deterministic tests or
418    /// precomputed fixtures.
419    pub fn new_with_nonce(
420        ty: DataType,
421        endpoints: &[DataEndpoint],
422        sender: &str,
423        timestamp: u64,
424        nonce: u16,
425        payload: Arc<[u8]>,
426    ) -> TelemetryResult<Self> {
427        Self::new_with_owned_wire_contract(
428            ty,
429            Arc::from(endpoints),
430            Arc::from(sender),
431            timestamp,
432            nonce,
433            StandardSmallPayload::from_arc(payload),
434            None,
435            None,
436        )
437    }
438
439    /// Resolve the schema element this packet should use for validation,
440    /// formatting, and typed payload access.
441    ///
442    /// Normal locally-built packets use the current runtime schema entry for
443    /// `self.ty`. Packets decoded from migration-safe frames may instead carry
444    /// `self.wire_shape`, which takes precedence so an in-flight payload can
445    /// still be interpreted after runtime schema churn.
446    #[inline]
447    fn effective_element(&self) -> MessageElement {
448        self.wire_shape
449            .unwrap_or_else(|| message_meta(self.ty).element)
450    }
451
452    /// Return the effective primitive payload kind for this packet.
453    ///
454    /// This is derived from `effective_element()` so typed decode helpers keep
455    /// following any inline wire shape when one is present.
456    #[inline]
457    fn effective_data_type(&self) -> MessageDataType {
458        self.effective_element().data_type()
459    }
460
461    /// Return the effective message class for this packet.
462    ///
463    /// Like `effective_data_type()`, this honors inline wire-shape metadata so
464    /// formatting remains consistent for in-flight packets across schema
465    /// changes.
466    #[inline]
467    fn effective_message_class(&self) -> MessageClass {
468        self.effective_element().message_type()
469    }
470
471    /// Compute a stable 64-bit identifier for this packet.
472    ///
473    /// This is *not* packed – it is derived locally from:
474    /// - sender
475    /// - logical type (DataType)
476    /// - endpoints
477    /// - timestamp
478    /// - payload bytes
479    ///
480    /// Identical packets on different boards/links will compute the same ID,
481    /// so the Router/Relay can use it to drop duplicates.
482    #[inline]
483    pub fn packet_id(&self) -> u64 {
484        hash_bytes_u64(self.packet_id_header(), self.payload())
485    }
486
487    #[cfg(feature = "latest-value-egress")]
488    pub(crate) fn scoped_packet_id(&self) -> TelemetryResult<u64> {
489        let header = self.packet_id_header();
490        #[cfg(feature = "compact-packet-store")]
491        {
492            self.payload
493                .with_slice(|bytes| hash_bytes_u64(header, bytes))
494        }
495        #[cfg(not(feature = "compact-packet-store"))]
496        {
497            Ok(hash_bytes_u64(header, self.payload()))
498        }
499    }
500
501    #[inline]
502    fn packet_id_header(&self) -> u64 {
503        // Seed with an arbitrary non-zero constant.
504        let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
505
506        // Compact source address. Sender names are discovery/config metadata,
507        // not packet-header identity.
508        h = hash_bytes_u64(h, &sender_address_u32(self.sender.as_ref()).to_le_bytes());
509
510        // Wire identity must not change when an incomplete schema is learned.
511        h = hash_bytes_u64(h, &self.ty.0.to_le_bytes());
512
513        // The wire carries an endpoint set, in ascending ID order. Input
514        // declaration order and duplicate destinations must not change the
515        // receipt ID. The common decoded/sorted path needs no normalization.
516        if self
517            .endpoints
518            .windows(2)
519            .all(|pair| pair[0].as_u32() < pair[1].as_u32())
520        {
521            for ep in self.endpoints.iter() {
522                h = hash_bytes_u64(h, &ep.0.to_le_bytes());
523            }
524        } else {
525            for ep in crate::wire_format::unique_sorted_endpoints(&self.endpoints) {
526                h = hash_bytes_u64(h, &ep.0.to_le_bytes());
527            }
528        }
529
530        // Timestamp + data_size as bytes
531        h = hash_bytes_u64(h, &self.timestamp.to_le_bytes());
532        h = hash_bytes_u64(h, &self.nonce.to_le_bytes());
533        h = hash_bytes_u64(h, &(self.data_size as u64).to_le_bytes());
534
535        h
536    }
537
538    /// Generic helper: decode the payload as a Vec<T> in little-endian,
539    /// after checking the runtime data kind and size.
540    #[inline]
541    fn _as_le_bytes<T>(&self, expected_kind: MessageDataType) -> TelemetryResult<Vec<T>>
542    where
543        T: LeDecode,
544    {
545        self.ensure_kind(expected_kind)?;
546
547        let bytes: &[u8] = self.payload();
548        let width = T::WIDTH;
549
550        if !bytes.len().is_multiple_of(width) {
551            // Packet should already be validated; if not, surface a size error.
552            return Err(TelemetryError::SizeMismatch {
553                expected: (bytes.len() / width) * width,
554                got: bytes.len(),
555            });
556        }
557
558        let count = bytes.len() / width;
559        let mut out = Vec::with_capacity(count);
560
561        for chunk in bytes.chunks_exact(width) {
562            out.push(T::from_le(chunk));
563        }
564
565        Ok(out)
566    }
567
568    /// Validate basic invariants:
569    ///
570    /// - `endpoints` is non-empty.
571    /// - `payload.len() == data_size`.
572    /// - For static element count:
573    ///   - `data_size == element_count * data_type_size(get_data_type(ty))`.
574    /// - For dynamic:
575    ///   - Length and encoding are validated by [`validate_dynamic_len_and_content`].
576    /// # Returns
577    /// - `Ok(())` if validation passes.
578    /// - `Err(TelemetryError)` if validation fails.
579    pub fn validate(&self) -> TelemetryResult<()> {
580        if self.endpoints.is_empty() {
581            return Err(TelemetryError::EmptyEndpoints);
582        }
583        if self.payload.len() != self.data_size {
584            return Err(TelemetryError::SizeMismatch {
585                expected: self.data_size,
586                got: self.payload.len(),
587            });
588        }
589
590        Self::validate_payload_against_element(self.effective_element(), &self.payload)
591    }
592
593    /* ---- Getters ---- */
594    /// Get the message data type.
595    /// This is the logical schema selector.
596    #[inline]
597    pub fn data_type(&self) -> DataType {
598        self.ty
599    }
600
601    /// Get the sender identifier.
602    /// This is typically a device or subsystem name.
603    #[inline]
604    pub fn sender(&self) -> &str {
605        &self.sender
606    }
607
608    /// Get the destination endpoints for this message.
609    #[inline]
610    pub fn endpoints(&self) -> &[DataEndpoint] {
611        &self.endpoints
612    }
613
614    /// Get the timestamp in milliseconds.
615    #[inline]
616    pub fn timestamp(&self) -> u64 {
617        self.timestamp
618    }
619
620    /// Get the packet nonce.
621    #[inline]
622    pub fn nonce(&self) -> u16 {
623        self.nonce
624    }
625
626    /// Get the payload size in bytes.
627    #[inline]
628    pub fn data_size(&self) -> usize {
629        self.data_size
630    }
631
632    /// Get the raw payload bytes.
633    #[inline]
634    pub fn payload(&self) -> &[u8] {
635        &self.payload
636    }
637
638    /// Read one payload field/range without caching or retaining a full slice.
639    pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
640        #[cfg(feature = "compact-packet-store")]
641        {
642            self.payload.read_range(offset, out)
643        }
644        #[cfg(not(feature = "compact-packet-store"))]
645        {
646            let end = offset
647                .checked_add(out.len())
648                .filter(|&n| n <= self.data_size)
649                .ok_or(TelemetryError::Unpack("payload range out of bounds"))?;
650            out.copy_from_slice(&self.payload()[offset..end]);
651            Ok(())
652        }
653    }
654
655    /// Return the optional inline wire shape preserved from unpacking.
656    ///
657    /// This is crate-visible because packing and routing need to keep the
658    /// contract intact when a packet is forwarded again.
659    #[inline]
660    pub(crate) fn wire_shape(&self) -> Option<MessageElement> {
661        self.wire_shape
662    }
663
664    /// Return the frozen destination-holder hashes preserved from the wire contract.
665    ///
666    /// Routers and relays use this list to avoid delivering an in-flight packet
667    /// to newly-learned holders that were not part of the original delivery
668    /// contract when the packet was packed.
669    #[inline]
670    pub(crate) fn wire_target_senders(&self) -> &[u64] {
671        self.wire_target_senders.as_deref().unwrap_or(&[])
672    }
673
674    /// Override the packet nonce while keeping the rest of the packet intact.
675    #[inline]
676    pub fn with_nonce(mut self, nonce: u16) -> Self {
677        self.nonce = nonce;
678        self
679    }
680
681    /// Header-only string (no decoded data).
682    ///
683    /// Example:
684    /// `Type: FOO, Data Size: 8, Sender: dev0, Endpoints: [EP_A, EP_B], Timestamp: 1234 (1s 234ms)`
685    /// # Returns
686    /// - Human-readable string with header fields.
687    pub fn header_string(&self) -> String {
688        let mut out = String::with_capacity(DEFAULT_STRING_CAPACITY);
689
690        let _ = write!(
691            &mut out,
692            "Type: {}, Data Size: {}, Sender: {}, Endpoints: [",
693            get_message_name(self.ty),
694            self.data_size,
695            self.sender.as_ref(),
696        );
697        for (i, ep) in self.endpoints.iter().enumerate() {
698            if i != 0 {
699                out.push_str(", ");
700            }
701            #[cfg(feature = "std")]
702            {
703                let endpoint_name = ep.as_str();
704                out.push_str(endpoint_name.as_ref());
705            }
706            #[cfg(not(feature = "std"))]
707            out.push_str(ep.as_str());
708        }
709        out.push_str("], Timestamp: ");
710        let _ = write!(&mut out, "{}", self.timestamp);
711
712        out.push_str(" (");
713        append_human_time(&mut out, self.timestamp);
714        out.push(')');
715        out
716    }
717
718    /// Borrow the payload as UTF-8 without trailing NULs (no allocation).
719    ///
720    /// Returns `None` if the message `DataType` is not a `String` type or if
721    /// the payload is not valid UTF-8 (after trimming trailing NUL).
722    /// # Returns
723    /// - `Some(&str)` if the payload is a valid UTF-8 string.
724    /// - `None` otherwise.
725    #[inline]
726    pub fn data_as_utf8_ref(&self) -> Option<&str> {
727        if self.effective_data_type() != MessageDataType::String {
728            return None;
729        }
730        let bytes = &self.payload;
731        let end = bytes.iter().rposition(|&b| b != 0).map(|i| i + 1)?;
732        core::str::from_utf8(&bytes[..end]).ok()
733    }
734
735    /// Helper: append decoded numeric/float elements to `s`.
736    ///
737    /// - Uses `LeBytes::from_le_slice` with fixed-width chunks.
738    /// - Floats (`f32`/`f64`) are formatted with a fixed precision
739    ///   the current runtime string precision.
740    #[inline]
741    fn data_to_string<T>(&self, s: &mut String)
742    where
743        T: LeBytes + core::fmt::Display + 'static,
744    {
745        let mut first = true;
746
747        for chunk in self.payload.chunks(T::WIDTH) {
748            if chunk.len() != T::WIDTH {
749                break;
750            }
751            if !first {
752                s.push_str(", ");
753            }
754            first = false;
755
756            let v = T::from_le_slice(chunk);
757
758            // If this is a float type, use precision; otherwise, default formatting.
759            if TypeId::of::<T>() == TypeId::of::<f32>() || TypeId::of::<T>() == TypeId::of::<f64>()
760            {
761                // `{:.*}` = "use this precision argument"
762                let _ = write!(s, "{:.*}", runtime_string_precision(), v);
763            } else {
764                let _ = write!(s, "{v}");
765            }
766        }
767    }
768
769    /// Full pretty string including decoded data portion.
770    ///
771    /// - String payloads are rendered as `"..."`
772    /// - Numeric/bool payloads are rendered as comma-separated values
773    /// - Hex payloads are delegated to [`Packet::to_hex_string`]
774    /// # Returns
775    /// - Human-readable string with header and decoded data.
776    pub fn as_string(&self) -> String {
777        let mut s = String::from("{");
778        s.push_str(&self.header_string());
779
780        if self.payload.is_empty() {
781            s.push_str(", Data: (<NoData>)}");
782            return s;
783        }
784
785        match self.effective_message_class() {
786            MessageClass::Data => {
787                s.push_str(", Data: (");
788            }
789            MessageClass::Error => {
790                s.push_str(", Error: (");
791            }
792            MessageClass::Warning => {
793                s.push_str(", Warning: (");
794            }
795        }
796
797        if let Some(msg) = self.data_as_utf8_ref() {
798            s.push('"');
799            s.push_str(msg);
800            s.push_str("\")}");
801            return s;
802        }
803
804        match self.effective_data_type() {
805            MessageDataType::Float64 => {
806                self.data_to_string::<f64>(&mut s);
807            }
808            MessageDataType::Float32 => {
809                self.data_to_string::<f32>(&mut s);
810            }
811            MessageDataType::UInt128 => {
812                self.data_to_string::<u128>(&mut s);
813            }
814            MessageDataType::UInt64 => {
815                self.data_to_string::<u64>(&mut s);
816            }
817            MessageDataType::UInt32 => {
818                self.data_to_string::<u32>(&mut s);
819            }
820            MessageDataType::UInt16 => {
821                self.data_to_string::<u16>(&mut s);
822            }
823            MessageDataType::UInt8 => {
824                self.data_to_string::<u8>(&mut s);
825            }
826            MessageDataType::Int128 => {
827                self.data_to_string::<i128>(&mut s);
828            }
829            MessageDataType::Int64 => {
830                self.data_to_string::<i64>(&mut s);
831            }
832            MessageDataType::Int32 => {
833                self.data_to_string::<i32>(&mut s);
834            }
835            MessageDataType::Int16 => {
836                self.data_to_string::<i16>(&mut s);
837            }
838            MessageDataType::Int8 => {
839                self.data_to_string::<i8>(&mut s);
840            }
841            MessageDataType::Bool => {
842                // Interpret any nonzero as true.
843                let mut it = self.payload.iter().peekable();
844                while let Some(b) = it.next() {
845                    let _ = write!(s, "{}", *b != 0);
846                    if it.peek().is_some() {
847                        s.push_str(", ");
848                    }
849                }
850            }
851            MessageDataType::String => {
852                // Already handled above via `data_as_utf8_ref`.
853            }
854            MessageDataType::Binary => return self.to_hex_string(),
855            MessageDataType::NoData => {
856                s.push_str("<no data>");
857            }
858        }
859
860        s.push_str(")}");
861        s
862    }
863
864    /// Hex dump variant of [`Packet::as_string`].
865    ///
866    /// Produces:
867    ///
868    /// `Type: ..., Data Size: ..., ..., Timestamp: ... (...), Data (hex): 0xNN 0xNN ...`
869    /// # Returns
870    /// - Human-readable string with header and hex-formatted data.
871    pub fn to_hex_string(&self) -> String {
872        // Header first.
873        let mut s = self.header_string();
874        s.push_str(", Data (hex):");
875
876        if !self.payload.is_empty() {
877            // Reserve roughly 5 chars per byte: " 0xNN".
878            s.reserve(self.payload.len().saturating_mul(5));
879            for &b in self.payload.iter() {
880                let _ = write!(&mut s, " 0x{:02x}", b);
881            }
882        }
883        s
884    }
885
886    // =========================================================================
887    // Typed data accessors
888    // =========================================================================
889
890    /// Ensure this packet's element type matches `expected`.
891    #[inline]
892    fn ensure_kind(&self, expected: MessageDataType) -> TelemetryResult<()> {
893        let dt = self.effective_data_type();
894        if dt != expected {
895            return Err(TelemetryError::TypeMismatch {
896                expected: data_type_size(expected),
897                got: data_type_size(dt),
898            });
899        }
900        Ok(())
901    }
902
903    impl_data_as_prim! {
904        data_as_f32,   f32,   MessageDataType::Float32;
905        data_as_f64,   f64,   MessageDataType::Float64;
906
907        data_as_u8,    u8,    MessageDataType::UInt8;
908        data_as_u16,   u16,   MessageDataType::UInt16;
909        data_as_u32,   u32,   MessageDataType::UInt32;
910        data_as_u64,   u64,   MessageDataType::UInt64;
911        data_as_u128,  u128,  MessageDataType::UInt128;
912
913        data_as_i8,    i8,    MessageDataType::Int8;
914        data_as_i16,   i16,   MessageDataType::Int16;
915        data_as_i32,   i32,   MessageDataType::Int32;
916        data_as_i64,   i64,   MessageDataType::Int64;
917        data_as_i128,  i128,  MessageDataType::Int128;
918    }
919
920    /// Decode payload as `bool`s. Any non-zero byte is treated as `true`.
921    #[inline]
922    pub fn data_as_bool(&self) -> TelemetryResult<Vec<bool>> {
923        self.ensure_kind(MessageDataType::Bool)?;
924        Ok(self.payload.iter().map(|&b| b != 0).collect())
925    }
926
927    /// Decode payload as a string (for String type).
928    #[inline]
929    pub fn data_as_string(&self) -> TelemetryResult<String> {
930        self.ensure_kind(MessageDataType::String)?;
931
932        let bytes = &self.payload;
933        let end = bytes
934            .iter()
935            .rposition(|&b| b != 0)
936            .map(|i| i + 1)
937            .unwrap_or(0);
938
939        if end == 0 {
940            return Ok(String::new());
941        }
942
943        let s = core::str::from_utf8(&bytes[..end])
944            .map_err(|_| TelemetryError::InvalidUtf8)?
945            .to_string();
946        Ok(s)
947    }
948
949    /// Decode payload as raw bytes (for Binary type).
950    #[inline]
951    pub fn data_as_binary(&self) -> TelemetryResult<Vec<u8>> {
952        self.ensure_kind(MessageDataType::Binary)?;
953        Ok(self.payload.to_vec())
954    }
955
956    /// Internal helper: build a packet from a slice of primitive values
957    /// encoded as little-endian, using the given sender.
958    ///
959    /// Works for all numeric types (integer and float) as long as the schema's
960    /// element width matches `T`'s width. Not used for String/Binary/Bool.
961    fn from_prim_le_slice_with_sender<T>(
962        ty: DataType,
963        values: &[T],
964        endpoints: &[DataEndpoint],
965        sender: &str,
966        timestamp: u64,
967    ) -> TelemetryResult<Self>
968    where
969        T: Copy + 'static,
970    {
971        let dt = get_data_type(ty);
972
973        // Only allow numeric-ish types here; String/Binary/Bool are handled elsewhere.
974
975        if dt == MessageDataType::Bool
976            || dt == MessageDataType::String
977            || dt == MessageDataType::Binary
978        {
979            // For these, use dedicated constructors (bool / string / binary).
980            return Err(TelemetryError::BadArg);
981        }
982        let element_size = data_type_size(dt);
983
984        // Ensure T's width matches what the schema expects.
985        if element_size != size_of::<T>() {
986            return Err(TelemetryError::TypeMismatch {
987                expected: element_size,
988                got: size_of::<T>(),
989            });
990        }
991
992        let total_bytes = values.len() * element_size;
993
994        // If the schema has a static element count, enforce it up front.
995        if let MessageElement::Static(exact, _, _) = message_meta(ty).element {
996            let exact_bytes = exact * element_size;
997            if total_bytes != exact_bytes {
998                return Err(TelemetryError::SizeMismatch {
999                    expected: exact_bytes,
1000                    got: total_bytes,
1001                });
1002            }
1003        }
1004
1005        let mut bytes = vec![0u8; total_bytes];
1006        unsafe { bytes.set_len(total_bytes) };
1007
1008        for (i, v) in values.iter().copied().enumerate() {
1009            let offset = i * element_size;
1010            let dst = &mut bytes[offset..offset + element_size];
1011
1012            // Copy the raw memory of `v` into the buffer.
1013            unsafe {
1014                core::ptr::copy_nonoverlapping(
1015                    &v as *const T as *const u8,
1016                    dst.as_mut_ptr(),
1017                    element_size,
1018                );
1019            }
1020
1021            // Normalize to little-endian on big-endian targets.
1022            // On little-endian this block is compiled out.
1023            #[cfg(target_endian = "big")]
1024            {
1025                dst.reverse();
1026            }
1027        }
1028
1029        Self::new(ty, endpoints, sender, timestamp, Arc::<[u8]>::from(bytes))
1030    }
1031
1032    /// Same as `from_prim_le_slice_with_sender` but uses the runtime device identifier.
1033    /// as the sender (mirrors `from_u8_slice`, `from_f32_slice`).
1034    #[inline]
1035    pub fn from_prim_le_slice<T>(
1036        ty: DataType,
1037        values: &[T],
1038        endpoints: &[DataEndpoint],
1039        timestamp: u64,
1040    ) -> TelemetryResult<Self>
1041    where
1042        T: Copy + 'static,
1043    {
1044        let sender = runtime_device_identifier();
1045        Self::from_prim_le_slice_with_sender(ty, values, endpoints, &sender, timestamp)
1046    }
1047
1048    // -------------------------------------------------------------------------
1049    // Convenience wrappers for all numeric types
1050    // -------------------------------------------------------------------------
1051    impl_from_prim_slices! {
1052        from_u8_slice,   u8;
1053        from_u16_slice,  u16;
1054        from_i8_slice,   i8;
1055        from_i16_slice,  i16;
1056        from_u32_slice,  u32;
1057        from_i32_slice,  i32;
1058        from_u64_slice,  u64;
1059        from_i64_slice,  i64;
1060        from_u128_slice, u128;
1061        from_i128_slice, i128;
1062        from_f32_slice,  f32;
1063        from_f64_slice,  f64;
1064    }
1065
1066    /// Builds a packet with an empty payload for types whose schema allows zero bytes.
1067    #[inline]
1068    pub fn from_no_data(
1069        ty: DataType,
1070        endpoints: &[DataEndpoint],
1071        timestamp: u64,
1072    ) -> TelemetryResult<Self> {
1073        let meta = message_meta(ty);
1074        match meta.element {
1075            MessageElement::Static(need, _, _) => {
1076                if need != 0 {
1077                    return Err(TelemetryError::SizeMismatch {
1078                        expected: need,
1079                        got: 0,
1080                    });
1081                }
1082            }
1083            MessageElement::Dynamic(_, _) => {
1084                // Dynamic with zero-length payload is OK.
1085            }
1086        }
1087
1088        let sender = runtime_device_identifier();
1089        Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from([]))
1090    }
1091
1092    /// Bool constructor: encodes each bool as a single byte (0 / 1).
1093    #[inline]
1094    pub fn from_bool_slice(
1095        ty: DataType,
1096        values: &[bool],
1097        endpoints: &[DataEndpoint],
1098        timestamp: u64,
1099    ) -> TelemetryResult<Self> {
1100        if get_data_type(ty) != MessageDataType::Bool {
1101            return Err(TelemetryError::TypeMismatch {
1102                expected: data_type_size(get_data_type(ty)),
1103                got: size_of::<bool>(),
1104            });
1105        }
1106
1107        let total_bytes = values.len();
1108        if let MessageElement::Static(exact, _, _) = message_meta(ty).element
1109            && total_bytes != exact
1110        {
1111            return Err(TelemetryError::SizeMismatch {
1112                expected: exact,
1113                got: total_bytes,
1114            });
1115        }
1116
1117        let mut bytes = Vec::with_capacity(total_bytes);
1118        bytes.extend(values.iter().map(|b| if *b { 1u8 } else { 0u8 }));
1119
1120        let sender = runtime_device_identifier();
1121        Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from(bytes))
1122    }
1123
1124    /// String constructor (dynamic length). Trailing NULs are not added;
1125    /// `new()` + `validate_dynamic_len_and_content` will do UTF-8 validation.
1126    #[inline]
1127    pub fn from_str_slice(
1128        ty: DataType,
1129        s: &str,
1130        endpoints: &[DataEndpoint],
1131        timestamp: u64,
1132    ) -> TelemetryResult<Self> {
1133        if get_data_type(ty) != MessageDataType::String {
1134            return Err(TelemetryError::TypeMismatch {
1135                expected: data_type_size(get_data_type(ty)),
1136                got: 1,
1137            });
1138        }
1139
1140        let bytes: Arc<[u8]> = Arc::from(s.as_bytes());
1141        let sender = runtime_device_identifier();
1142        Self::new(ty, endpoints, &sender, timestamp, bytes)
1143    }
1144}
1145
1146// ============================================================================
1147// Time formatting (no_std-friendly, UTC or uptime)
1148// ============================================================================
1149
1150#[inline]
1151fn div_mod_u64(n: u64, d: u64) -> (u64, u64) {
1152    (n / d, n % d)
1153}
1154
1155// Howard Hinnant–style civil-from-days (proleptic Gregorian).
1156fn civil_from_days(mut z: i64) -> (i32, u32, u32) {
1157    // epoch (1970-01-01) has days=0
1158    z += 719_468; // shift to civil base
1159    let era = (if z >= 0 { z } else { z - 146_096 }) / 146_097;
1160    let doe = z - era * 146_097; // [0, 146096]
1161    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365; // [0, 399]
1162    let y = (yoe as i32) + era as i32 * 400;
1163    let doy = (doe - (365 * yoe + yoe / 4 - yoe / 100)) as i32; // [0, 365]
1164    let mp = (5 * doy + 2) / 153; // [0, 11]
1165    let d = doy - (153 * mp + 2) / 5 + 1; // [1, 31]
1166    let m = mp + if mp < 10 { 3 } else { -9 }; // [1, 12]
1167    let y = y + (m <= 2) as i32; // year
1168    (y, m as u32, d as u32)
1169}
1170
1171/// Append a human-readable timestamp to `out`, either uptime (`hh:mm:ss.mmm`)
1172/// or UTC epoch like `YYYY-MM-DD HH:MM:SS.mmmZ`, depending on threshold.
1173fn append_human_time(out: &mut String, total_ms: u64) {
1174    if total_ms >= EPOCH_MS_THRESHOLD {
1175        // Unix epoch path.
1176        let (secs, sub_ms) = div_mod_u64(total_ms, 1_000);
1177        let days = (secs / 86_400) as i64;
1178        let sod = (secs % 86_400) as u32; // seconds of day
1179        let (year, month, day) = civil_from_days(days);
1180        let hour = sod / 3_600;
1181        let min = (sod % 3_600) / 60;
1182        let sec = sod % 60;
1183        let _ = Write::write_fmt(
1184            out,
1185            format_args!(
1186                "{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}Z",
1187                year, month, day, hour, min, sec, sub_ms as u32
1188            ),
1189        );
1190    } else {
1191        // Uptime-style duration.
1192        let hours = total_ms / 3_600_000;
1193        let minutes = (total_ms % 3_600_000) / 60_000;
1194        let seconds = (total_ms % 60_000) / 1_000;
1195        let milliseconds = total_ms % 1_000;
1196        if hours > 0 {
1197            let _ = Write::write_fmt(
1198                out,
1199                format_args!("{hours}h {minutes:02}m {seconds:02}s {milliseconds:03}ms"),
1200            );
1201        } else if minutes > 0 {
1202            let _ = Write::write_fmt(
1203                out,
1204                format_args!("{minutes}m {seconds:02}s {milliseconds:03}ms"),
1205            );
1206        } else {
1207            let _ = Write::write_fmt(out, format_args!("{seconds}s {milliseconds:03}ms"));
1208        }
1209    }
1210}
1211
1212// ============================================================================
1213// Display impl
1214// ============================================================================
1215
1216impl core::fmt::Display for Packet {
1217    #[inline]
1218    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
1219        f.write_str(&Packet::as_string(self))
1220    }
1221}
1222
1223impl ByteCost for Packet {
1224    #[cfg(feature = "compact-packet-store")]
1225    fn park_for_queue(&mut self) -> TelemetryResult<()> {
1226        self.payload.park_for_queue()
1227    }
1228    #[inline]
1229    fn byte_cost(&self) -> usize {
1230        size_of::<Self>()
1231            + self.sender.len()
1232            + self.endpoints.len() * size_of::<DataEndpoint>()
1233            + self.payload.byte_cost()
1234    }
1235}
1236
1237/// A consumed packet with separately stored payload. No original full payload or
1238/// decompressed cache remains. Legacy full-slice APIs are deliberately unavailable.
1239#[cfg(feature = "compact-packet-store")]
1240pub struct StoredPacket {
1241    header: Packet,
1242    payload: crate::packet_store::StoredBytes,
1243}
1244#[cfg(feature = "compact-packet-store")]
1245impl StoredPacket {
1246    pub fn data_type(&self) -> DataType {
1247        self.header.data_type()
1248    }
1249    pub fn sender(&self) -> &str {
1250        self.header.sender()
1251    }
1252    pub fn data_size(&self) -> usize {
1253        self.payload.len()
1254    }
1255    pub fn stored_payload_size(&self) -> usize {
1256        self.payload.stored_len()
1257    }
1258    pub fn payload_is_compressed(&self) -> bool {
1259        self.payload.is_compressed()
1260    }
1261    pub fn pin_payload(&self) -> TelemetryResult<crate::packet_store::PinnedBytes> {
1262        self.payload.pin()
1263    }
1264    pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
1265        self.payload.read_range(offset, out)
1266    }
1267    #[cfg(feature = "compact-packet-compression")]
1268    pub fn read_payload_range_bounded(
1269        &self,
1270        codec: &mut crate::packet_store::BoundedPacketCodec,
1271        offset: usize,
1272        out: &mut [u8],
1273    ) -> TelemetryResult<()> {
1274        codec.read_range(&self.payload, offset, out)
1275    }
1276    #[cfg(feature = "compact-packet-compression")]
1277    pub fn read_payload_range_with_codec(
1278        &self,
1279        codec: &mut crate::packet_store::PacketCodec,
1280        offset: usize,
1281        out: &mut [u8],
1282        scratch: &mut [u8],
1283    ) -> TelemetryResult<()> {
1284        codec.read_range(&self.payload, offset, out, scratch)
1285    }
1286}
1287#[cfg(feature = "compact-packet-store")]
1288impl Packet {
1289    pub fn into_stored(
1290        mut self,
1291        store: &Arc<crate::packet_store::PacketStore>,
1292    ) -> TelemetryResult<StoredPacket> {
1293        let payload = store.store(self.payload())?;
1294        self.payload = StandardSmallPayload::new(&[]);
1295        Ok(StoredPacket {
1296            header: self,
1297            payload,
1298        })
1299    }
1300    #[cfg(feature = "compact-packet-compression")]
1301    pub fn into_stored_bounded(
1302        mut self,
1303        store: &Arc<crate::packet_store::PacketStore>,
1304        codec: &mut crate::packet_store::BoundedPacketCodec,
1305    ) -> TelemetryResult<StoredPacket> {
1306        let payload = codec.store(store, self.payload())?;
1307        self.payload = StandardSmallPayload::new(&[]);
1308        Ok(StoredPacket {
1309            header: self,
1310            payload,
1311        })
1312    }
1313    #[cfg(feature = "compact-packet-compression")]
1314    pub fn into_stored_compressed(
1315        mut self,
1316        store: &Arc<crate::packet_store::PacketStore>,
1317        codec: &mut crate::packet_store::PacketCodec,
1318        scratch: &mut [u8],
1319    ) -> TelemetryResult<StoredPacket> {
1320        let payload = codec.store(store, self.payload(), scratch)?;
1321        self.payload = StandardSmallPayload::new(&[]);
1322        Ok(StoredPacket {
1323            header: self,
1324            payload,
1325        })
1326    }
1327}