//! M5: Real iceoryx2 production sink.
//!
//! One publish-subscribe service per configured stream plus a dedicated
//! heartbeat stream. All samples are fixed-size `[u8]` slices with an
//! in-band header so consumers can read key/value directly from shm.
//!
//! Binary layout per sample (all values little-endian):
//!   [0..8)   seq         u64 — monotonically increasing per-stream
//!   [8..16)  src_ts_us   u64 — upstream Hazelcast event timestamp (µs epoch)
//!   [16..24) ts_us       u64 — local publish timestamp (µs epoch)
//!   [24..72) key         [u8; 48] — NUL-padded map key (UTF-8)
//!   [72..76) len         u32 — actual value length (≤ max_value_kib × 1024)
//!   [76..)   value       [u8; MAX_VALUE] — opaque value bytes
//!
//! Heartbeat uses the same layout with seq=0 and key="heartbeat".
//!
//! Constraint compliance:
//!   C1 — `#![forbid(unsafe_code)]` at crate root: no unsafe here.
//!   C2 — no async: iceoryx2 API is synchronous.
//!   C5 — iceoryx2 0.9.1 already pinned in Cargo.toml.
//!   C8 — fixed-size samples, pre-allocated publishers.

use crate::config::Config;
use crate::metrics::MetricsSnapshot;
use crate::sink::{PlaneSink, SinkError};
use iceoryx2::prelude::*;

/// Alias for thread-safe IPC service type.
use log::{info, warn};
use std::collections::HashMap;

// ── Binary layout constants ────────────────────────────────────────────────

/// Byte offset at which the value payload begins.
const HEADER_BYTES: usize = 76; // seq(8) + src_ts_us(8) + ts_us(8) + key(48) + len(4)
const KEY_LEN: usize = 48;

// iceoryx2 Publisher<Service=ipc_threadsafe::Service, Payload=[u8], UserHeader=()>
type PublisherIpc = iceoryx2::port::publisher::Publisher<ipc_threadsafe::Service, [u8], ()>;

// ── Per-stream publisher state ──────────────────────────────────────────────

struct StreamPublisher {
    /// The iceoryx2 publisher port.
    publisher: PublisherIpc,
    /// Total sample size = HEADER_BYTES + max_value_bytes.
    sample_size: usize,
    /// Monotonically increasing per-stream sequence number.
    seq: u64,
}

// ── Iceoryx2Sink ────────────────────────────────────────────────────────────

/// The production `PlaneSink` that publishes into iceoryx2 services.
///
/// Keyed by the **short stream name** from config (e.g. `"eigen_scan"`).
/// The full iceoryx2 service name is `{prefix}/{short_name}`.
pub struct Iceoryx2Sink {
    /// Map from short stream name → per-stream publisher.
    streams: HashMap<String, StreamPublisher>,
    /// Heartbeat publisher (best-effort; `None` if creation failed).
    heartbeat: Option<StreamPublisher>,
    /// The iceoryx2 node — held for the sink's lifetime.
    _node: iceoryx2::node::Node<ipc_threadsafe::Service>,
    /// Cached from config for use in `publish_and_notify`.
    max_value_bytes: usize,
}

type NodeIpc = iceoryx2::node::Node<ipc_threadsafe::Service>;

impl Iceoryx2Sink {
    /// Create the sink, opening or creating iceoryx2 services for every
    /// configured stream plus the heartbeat stream.
    ///
    /// Returns `SinkError` if any iceoryx2 service cannot be created or
    /// opened. Per §5.2 "local shm failure is a broken box" — the caller
    /// (the runtime) treats this as Fatal.
    pub fn new(config: &Config) -> Result<Self, SinkError> {
        let max_value_bytes = config.plane.max_value_kib * 1024;
        let sample_size = HEADER_BYTES + max_value_bytes;
        let prefix = &config.plane.service_prefix;

        let node_name = NodeName::new(&format!("hzbridge_{}", config.hz.cluster))
            .map_err(|e| SinkError::new(format!("node name: {e}")))?;
        let node = NodeBuilder::new()
            .name(&node_name)
            .create::<ipc_threadsafe::Service>()
            .map_err(|e| SinkError::new(format!("node create: {e:?}")))?;

        let mut streams = HashMap::new();
        for sc in &config.streams {
            let full_name = format!("{}/{}", prefix, sc.stream);
            let pub_ = Self::open_service(&node, &full_name, sample_size)?;
            info!(
                "sink=Iceoryx2Sink msg=stream_ready stream={} sample_size={}",
                full_name, sample_size,
            );
            streams.insert(sc.stream.clone(), pub_);
        }

        // Heartbeat — best-effort; a failure here does not abort the bridge.
        let hb_name = format!("{}/hzbridge_heartbeat", prefix);
        let heartbeat = match Self::open_service(&node, &hb_name, sample_size) {
            Ok(p) => {
                info!("sink=Iceoryx2Sink msg=heartbeat_ready stream={}", hb_name);
                Some(p)
            }
            Err(e) => {
                warn!("sink=Iceoryx2Sink msg=heartbeat_failed error={e}");
                None
            }
        };

        Ok(Self {
            streams,
            heartbeat,
            _node: node,
            max_value_bytes,
        })
    }

    /// Create or open one publish-subscribe service and return a publisher.
    fn open_service(
        node: &NodeIpc,
        service_name: &str,
        sample_size: usize,
    ) -> Result<StreamPublisher, SinkError> {
        let name = ServiceName::new(service_name)
            .map_err(|e| SinkError::new(format!("service name '{service_name}': {e}")))?;
        let service = node
            .service_builder(&name)
            .publish_subscribe::<[u8]>()
            .open_or_create()
            .map_err(|e| SinkError::new(format!("open_or_create '{service_name}': {e:?}")))?;
        let publisher = service
            .publisher_builder()
            .initial_max_slice_len(sample_size)
            .max_loaned_samples(2)
            .backpressure_strategy(BackpressureStrategy::DiscardData)
            .create()
            .map_err(|e| SinkError::new(format!("publisher '{service_name}': {e:?}")))?;
        Ok(StreamPublisher {
            publisher,
            sample_size,
            seq: 0,
        })
    }

    /// Write a value into a loaned iceoryx2 sample and send it.
    fn write_sample(
        pub_: &mut StreamPublisher,
        key: &str,
        value_utf8: &[u8],
        src_ts_us: u64,
        ts_us: u64,
        max_value_bytes: usize,
    ) -> Result<(), SinkError> {
        pub_.seq = pub_.seq.wrapping_add(1);
        let mut sample = pub_
            .publisher
            .loan_slice(pub_.sample_size)
            .map_err(|e| SinkError::new(format!("loan_slice: {e:?}")))?;

        let buf = sample.payload_mut();
        let seq = pub_.seq;

        // seq
        buf[0..8].copy_from_slice(&seq.to_le_bytes());
        // src_ts_us
        buf[8..16].copy_from_slice(&src_ts_us.to_le_bytes());
        // ts_us
        buf[16..24].copy_from_slice(&ts_us.to_le_bytes());
        // key (NUL-padded, truncated to KEY_LEN)
        let key_bytes = key.as_bytes();
        let copy_len = key_bytes.len().min(KEY_LEN);
        buf[24..24 + copy_len].copy_from_slice(&key_bytes[..copy_len]);
        if copy_len < KEY_LEN {
            buf[24 + copy_len..72].fill(0);
        }
        // len
        let value_len = value_utf8.len().min(max_value_bytes);
        buf[72..76].copy_from_slice(&(value_len as u32).to_le_bytes());
        // value
        let vs = HEADER_BYTES;
        buf[vs..vs + value_len].copy_from_slice(&value_utf8[..value_len]);
        // Zero remaining value buffer
        if value_len < max_value_bytes {
            buf[vs + value_len..vs + max_value_bytes].fill(0);
        }

        sample.send().map_err(|e| SinkError::new(format!("send: {e:?}")))?;
        Ok(())
    }

    fn now_us() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or(std::time::Duration::ZERO)
            .as_micros() as u64
    }

    /// Encode a `MetricsSnapshot` as a space-separated key=value text blob.
    fn encode_heartbeat(snapshot: &MetricsSnapshot) -> Vec<u8> {
        format!(
            "events_rx={} events_published={} decode_skips={} late_responses={} \
             reconnects={} fragments_reassembled={} warmups={} pings_ok={} \
             pings_missed={} events_dropped={} uptime_s={} state={} degraded={}",
            snapshot.events_rx,
            snapshot.events_published,
            snapshot.decode_skips,
            snapshot.late_responses,
            snapshot.reconnects,
            snapshot.fragments_reassembled,
            snapshot.warmups,
            snapshot.pings_ok,
            snapshot.pings_missed,
            snapshot.events_dropped,
            snapshot.uptime_s,
            snapshot.state,
            snapshot.degraded,
        )
        .into_bytes()
    }
}

impl PlaneSink for Iceoryx2Sink {
    fn publish(
        &mut self,
        stream: &str,
        key: &str,
        value_utf8: &[u8],
        src_ts_us: u64,
    ) -> Result<(), SinkError> {
        let ts_us = Self::now_us();
        let pub_ = self.streams.get_mut(stream).ok_or_else(|| {
            SinkError::new(format!("unknown stream: {stream}"))
        })?;
        Self::write_sample(pub_, key, value_utf8, src_ts_us, ts_us, self.max_value_bytes)
    }

    fn heartbeat(&mut self, snapshot: &MetricsSnapshot) -> Result<(), SinkError> {
        let Some(pub_) = self.heartbeat.as_mut() else {
            return Ok(());
        };
        let ts_us = Self::now_us();
        let value = Self::encode_heartbeat(snapshot);
        Self::write_sample(pub_, "heartbeat", &value, ts_us, ts_us, self.max_value_bytes)
    }
}

// ── Tests ──────────────────────────────────────────────────────────────────

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

    #[test]
    fn sample_layout_offsets_are_correct() {
        let mut buf = vec![0u8; HEADER_BYTES + 256];
        // seq = 42
        buf[0..8].copy_from_slice(&42u64.to_le_bytes());
        // src_ts_us = 1000
        buf[8..16].copy_from_slice(&1000u64.to_le_bytes());
        // ts_us = 2000
        buf[16..24].copy_from_slice(&2000u64.to_le_bytes());
        // key = "BTCUSDT"
        let k = b"BTCUSDT";
        buf[24..24 + k.len()].copy_from_slice(k);
        // len = 6
        buf[72..76].copy_from_slice(&6u32.to_le_bytes());
        // value = "123456"
        buf[76..82].copy_from_slice(b"123456");

        assert_eq!(u64::from_le_bytes(buf[0..8].try_into().unwrap()), 42);
        assert_eq!(u64::from_le_bytes(buf[8..16].try_into().unwrap()), 1000);
        assert_eq!(u64::from_le_bytes(buf[16..24].try_into().unwrap()), 2000);
        assert_eq!(&buf[24..31], b"BTCUSDT");
        assert_eq!(u32::from_le_bytes(buf[72..76].try_into().unwrap()), 6);
        assert_eq!(&buf[76..82], b"123456");
    }

    #[test]
    fn header_byte_count_is_76() {
        assert_eq!(HEADER_BYTES, 76);
    }

    #[test]
    fn heartbeat_encoding_contains_all_fields() {
        let snap = MetricsSnapshot {
            events_rx: 100,
            events_published: 95,
            decode_skips: 2,
            late_responses: 0,
            reconnects: 1,
            fragments_reassembled: 0,
            warmups: 3,
            pings_ok: 50,
            pings_missed: 1,
            events_dropped: 0,
            uptime_s: 3600,
            state: "Streaming".into(),
            degraded: false,
        };
        let encoded = Iceoryx2Sink::encode_heartbeat(&snap);
        let text = String::from_utf8(encoded).unwrap();
        assert!(text.contains("events_rx=100"));
        assert!(text.contains("events_published=95"));
        assert!(text.contains("decode_skips=2"));
        assert!(text.contains("state=Streaming"));
        assert!(text.contains("degraded=false"));
        assert!(text.contains("uptime_s=3600"));
    }

    #[test]
    fn key_truncation_fits_48_bytes() {
        let long_key = "a".repeat(100);
        let max_val = 256;
        let mut buf = vec![0u8; HEADER_BYTES + max_val];
        let kbytes = long_key.as_bytes();
        let copy_len = kbytes.len().min(KEY_LEN);
        buf[24..24 + copy_len].copy_from_slice(&kbytes[..copy_len]);
        // Verify the stored key is truncated to KEY_LEN and NUL-padded
        let stored = &buf[24..72];
        assert_eq!(&stored[..KEY_LEN - 1], &[b'a'; 47]);
        assert_eq!(stored[KEY_LEN - 1], b'a');
    }

    #[test]
    fn value_truncation_fits_max_value() {
        let max_val = 256;
        let big_value = vec![0xFFu8; 500];
        let value_len = big_value.len().min(max_val);
        assert_eq!(value_len, max_val);
        assert_eq!(value_len, 256);
    }
}