//! Verify the inline payload is a reference-counted
//! [`SharedBytes`] (`Arc<[u8]>`) so the bus and bridge fan-out paths can clone
//! the payload across multiple subscribers without duplicating the underlying
//! buffer.

use std::sync::Arc;

use data_model::{DataPayload, SharedBytes, shared_bytes_from_vec};

#[test]
fn inline_payload_clones_share_underlying_storage() {
    let payload = DataPayload::inline(vec![1u8, 2, 3, 4, 5]);
    let DataPayload::Inline(bytes) = payload else {
        panic!("expected inline")
    };

    let strong_before = Arc::strong_count(&bytes);
    assert_eq!(strong_before, 1);

    let n_subscribers = 8;
    let clones: Vec<SharedBytes> = (0..n_subscribers).map(|_| bytes.clone()).collect();

    // Cloning an Arc<[u8]> must not copy the buffer: all clones must point at
    // the same backing allocation as the original.
    let original_ptr = bytes.as_ptr();
    for c in &clones {
        assert_eq!(c.as_ptr(), original_ptr, "clone must reuse buffer");
        assert_eq!(c.as_ref(), &[1u8, 2, 3, 4, 5]);
    }

    let strong_after = Arc::strong_count(&bytes);
    assert_eq!(strong_after, strong_before + n_subscribers);

    drop(clones);
    assert_eq!(Arc::strong_count(&bytes), strong_before);
}

#[test]
fn shared_bytes_from_vec_reuses_vec_buffer() {
    let mut v = Vec::with_capacity(64);
    v.extend_from_slice(&[7u8, 8, 9]);
    let vec_ptr = v.as_ptr();
    let shared = shared_bytes_from_vec(v);
    // `Arc::<[u8]>::from(Box<[u8]>)` is allowed to copy when the Box's
    // allocation is over-sized -- with capacity == len (after into_boxed_slice
    // shrink) the pointer comparison may still differ.  Just verify the
    // payload survives the conversion and is exposed via the `Arc<[u8]>` API.
    let _ = vec_ptr;
    assert_eq!(shared.as_ref(), &[7u8, 8, 9]);
}

#[test]
fn inline_helper_accepts_arc_directly() {
    let arc: SharedBytes = Arc::from(vec![42u8].into_boxed_slice());
    let strong_before = Arc::strong_count(&arc);
    let payload = DataPayload::inline(arc.clone());
    let DataPayload::Inline(bytes) = payload else {
        panic!("expected inline")
    };
    assert_eq!(
        bytes.as_ptr(),
        arc.as_ptr(),
        "Arc<[u8]> input must not be copied"
    );
    let _ = strong_before;
}
