path: root/src/mycelium.rs

use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use base64::engine::{general_purpose::STANDARD as BASE64, Engine};
use reqwest::StatusCode;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_bytes::{ByteBuf, Bytes};
use std::borrow::Cow;
use std::boxed::Box;
use std::fs::File;
use std::io::Write;
use std::net::TcpListener;
use std::os::unix::io::FromRawFd;
use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;
use thiserror::Error;
use tokio::process::{Child, Command};
use url::Url;

use crate::transport::EncodedMessage;
use crate::transport::Keepalive;
use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
use crate::transport::{MessageHandler, ParsedMessage, Transport};

const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;

const TOPIC_PREFIX: &str = "continuum.v1.";
const TOPIC_HELLO: &str = "continuum.v1.hello";
const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
const TOPIC_SYNC: &str = "continuum.v1.sync";
const TOPIC_PING: &str = "continuum.v1.ping";
const TOPIC_PONG: &str = "continuum.v1.pong";

const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct InboundMessage {
    id: String,
    src_pk: String,
    topic: String,
    payload: String,
}

#[derive(Debug, Serialize)]
struct OutboundMessage {
    dst: Destination,
    topic: String,
    payload: String,
}

#[derive(Debug, Serialize)]
struct Destination {
    pk: String,
}

/// Represents where a 32-byte cryptographic key originates.
#[derive(Debug, Clone)]
pub enum KeySource {
    Raw([u8; 32]),
    Path(PathBuf),
}

/// Atomic structure ensuring private network credentials are always handled together.
#[derive(Debug, Clone)]
pub struct PrivateNetworkConfig {
    pub name: String,
    pub key: KeySource,
}

/// Comprehensive parameters required to spawn or target an instance.
#[derive(Debug, Clone)]
pub struct MyceliumSpawnConfig {
    pub rest_url_opt: Option<Url>,
    pub key_file_opt: Option<PathBuf>,
    pub network_key_file: Option<PathBuf>,
    pub network_name: String,
    pub tcp_port: Option<u16>,
    pub api_port: Option<u16>,
    pub rpc_port: Option<u16>,
    pub discovery_port: Option<u16>,
    pub peers: Vec<String>,
}

impl Default for MyceliumSpawnConfig {
    fn default() -> Self {
        Self {
            rest_url_opt: None,
            key_file_opt: None,
            network_key_file: None,
            network_name: "something".to_string(),
            tcp_port: None,
            api_port: None,
            rpc_port: None,
            discovery_port: None,
            peers: vec!["tls://127.0.0.1:9651".to_string()],
        }
    }
}

pub enum MyceliumRuntime {
    Supervised(Box<SupervisedMycelium>),
    External,
}

pub struct MyceliumSetup {
    pub runtime: MyceliumRuntime,
    pub rest_url: Url,
}

impl MyceliumSetup {
    /// Evaluates the configuration payload and initializes the chosen execution strategy.
    pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
        // TODO warn for all ignored options if rest_url_opt is set
        if let Some(url) = config.rest_url_opt {
            tracing::info!(
                "Configured to interface with an external standalone Mycelium instance at: {}",
                url
            );
            return Ok(Self {
                runtime: MyceliumRuntime::External,
                rest_url: url,
            });
        }

        tracing::info!(
            "Configured to initialize an embedded user-space Mycelium background worker"
        );

        let key = match config.key_file_opt {
            Some(key_file) => KeySource::Path(key_file),
            None => {
                let mut bytes = [0u8; 32];
                getrandom::fill(&mut bytes)?;
                KeySource::Raw(bytes)
            }
        };

        let private_net_config = config
            .network_key_file
            .map(|net_key_path| PrivateNetworkConfig {
                name: config.network_name.clone(),
                key: KeySource::Path(net_key_path),
            });

        let supervised_node = SupervisedMycelium::spawn(
            key,
            private_net_config,
            config.tcp_port,
            config.api_port,
            config.rpc_port,
            config.discovery_port,
            &config.peers,
        )
        .await
        .context("Failed to securely bootstrap embedded Mycelium background worker")?;

        let resolved_url = Url::parse(&format!(
            "http://127.0.0.1:{}",
            supervised_node.config.api_port
        ))?;

        Ok(Self {
            runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)),
            rest_url: resolved_url,
        })
    }

    /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely.
    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
        match &mut self.runtime {
            MyceliumRuntime::Supervised(node) => node.wait_for_exit().await,
            MyceliumRuntime::External => {
                std::future::pending::<std::io::Result<std::process::ExitStatus>>().await
            }
        }
    }
}

#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct MyceliumAdminInfo {
    pub node_ip: String,
    pub node_pubkey: String,
    pub node_start_time: u64,
    pub node_subnet: String,
}

pub async fn fetch_admin_info(rest_url: &Url) -> Result<MyceliumAdminInfo> {
    let url = rest_url.join("/api/v1/admin")?;
    let info = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?
        .get(url)
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;
    Ok(info)
}

#[derive(Debug, Clone)]
pub struct ClusterNodeConfig {
    pub tcp_port: u16,
    pub api_port: u16,
    pub rpc_port: u16,
    pub discovery_port: Option<u16>,
    pub private_network: Option<String>,
}

pub struct SupervisedMycelium {
    child: Child,
    pub config: ClusterNodeConfig,
    _key_file_handle: Option<File>,
    _net_key_file_handle: Option<File>,
    key_file_path: PathBuf,
    net_key_file_path: Option<PathBuf>,
    is_ephemeral_key: bool,
    is_ephemeral_net_key: bool,
}

impl SupervisedMycelium {
    async fn spawn(
        node_key: KeySource,
        private_network: Option<PrivateNetworkConfig>,
        explicit_tcp: Option<u16>,
        explicit_api: Option<u16>,
        explicit_rpc: Option<u16>,
        explicit_discovery_port: Option<u16>,
        peers: &[String],
    ) -> Result<Self> {
        let tcp_port = match explicit_tcp {
            Some(p) => p,
            None => Self::find_available_port().context("Failed to allocate TCP wire port")?,
        };
        let api_port = match explicit_api {
            Some(p) => p,
            None => Self::find_available_port().context("Failed to allocate API port")?,
        };
        let rpc_port = match explicit_rpc {
            Some(p) => p,
            None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?,
        };
        let discovery_port = match explicit_discovery_port {
            Some(0) => Some(
                Self::find_available_port().context("Failed to allocate peer discovery port")?,
            ),
            p => p,
        };

        let config = ClusterNodeConfig {
            tcp_port,
            api_port,
            rpc_port,
            discovery_port,
            private_network: private_network.as_ref().map(|pn| pn.name.clone()),
        };

        let parent_pid = std::process::id();

        let (key_file_handle, key_file_path, node_fd, is_ephemeral_key) =
            Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?;

        let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) =
            match private_network {
                Some(net_cfg) => {
                    let (h, p, fd, eph) =
                        Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
                    (h, Some(p), fd, eph)
                }
                None => (None, None, None, false),
            };

        let api_addr = format!("127.0.0.1:{}", config.api_port);
        let rpc_addr = format!("127.0.0.1:{}", config.rpc_port);

        let mut cmd = Command::new("mycelium");

        let fds_to_inherit: Vec<i32> = [node_fd, net_fd].iter().filter_map(|&x| x).collect();
        if !fds_to_inherit.is_empty() {
            unsafe {
                cmd.pre_exec(move || {
                    for &fd in &fds_to_inherit {
                        libc::fcntl(fd, libc::F_SETFD, 0);
                    }
                    Ok(())
                });
            }
        }

        cmd.arg("--key-file")
            .arg(&key_file_path)
            .arg("--tcp-listen-port")
            .arg(config.tcp_port.to_string())
            .arg("--disable-quic")
            .arg("--api-addr")
            .arg(&api_addr)
            .arg("--jsonrpc-addr")
            .arg(&rpc_addr)
            .arg("--no-tun");

        if let Some(ref discovery_port) = config.discovery_port {
            cmd.arg("--peer-discovery-port")
                .arg(discovery_port.to_string());
        } else {
            cmd.arg("--disable-peer-discovery");
        }

        if let Some(ref net_name) = config.private_network {
            cmd.arg("--network-name").arg(net_name);
        }
        if let Some(ref net_key_path) = net_key_file_path {
            cmd.arg("--network-key-file").arg(net_key_path);
        }

        if !peers.is_empty() {
            cmd.arg("--peers").arg(peers.join(" "));
        }

        cmd.env("RUST_LOG", "info");

        cmd.stdout(Stdio::piped());
        cmd.stderr(Stdio::piped());

        let mut child = cmd
            .spawn()
            .context("Failed to execute mycelium process daemon")?;

        // 1. FIXED: Pass-through streaming for child process STDOUT
        if let Some(mut stdout) = child.stdout.take() {
            tokio::spawn(async move {
                let mut host_stdout = tokio::io::stdout();
                let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await;
            });
        }

        // 2. FIXED: Pass-through streaming for child process STDERR
        if let Some(mut stderr) = child.stderr.take() {
            tokio::spawn(async move {
                let mut host_stderr = tokio::io::stderr();
                let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await;
            });
        }

        let node = Self {
            child,
            config,
            _key_file_handle: key_file_handle,
            _net_key_file_handle: net_key_file_handle,
            key_file_path,
            net_key_file_path,
            is_ephemeral_key,
            is_ephemeral_net_key,
        };

        //let admin_info = node.poll_admin_endpoint().await?;
        //Ok((node, admin_info))
        Ok(node)
    }

    fn resolve_key_strategy(
        desc_tag: &str,
        source: KeySource,
        parent_pid: u32,
    ) -> Result<(Option<File>, PathBuf, Option<i32>, bool)> {
        match source {
            KeySource::Raw(raw_bytes) => {
                let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?;
                let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd));
                Ok((Some(handle), pseudo_path, Some(fd), true))
            }
            KeySource::Path(existing_path) => Ok((None, existing_path, None, false)),
        }
    }

    fn find_available_port() -> Result<u16> {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?;
        Ok(addr.port())
    }

    fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> {
        #[cfg(target_os = "linux")]
        {
            let memfd_name = std::ffi::CString::new(desc_tag)?;
            let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) };
            if fd < 0 {
                return Err(anyhow!(
                    "memfd_create failed natively inside Linux kernel for tag: {}",
                    desc_tag
                ));
            }
            let mut file = unsafe { File::from_raw_fd(fd) };
            file.write_all(&raw_key)?;
            file.flush()?;
            Ok((file, fd))
        }

        #[cfg(not(target_os = "linux"))]
        {
            let mut path = std::env::tempdir();
            path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4()));
            let mut file = File::create(&path)?;
            file.write_all(&raw_key)?;
            file.flush()?;
            Ok((file, -1))
        }
    }

    //async fn poll_admin_endpoint(&mut self) -> Result<MyceliumAdminInfo> {
    //    let client = reqwest::Client::new();
    //    let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port);
    //
    //    for _ in 0..30 {
    //        if let Some(status) = self.child.try_wait()? {
    //            return Err(anyhow!(
    //                "myceliumd process closed prematurely with status: {}",
    //                status
    //            ));
    //        }
    //        if let Ok(resp) = client.get(&url).send().await {
    //            if resp.status().is_success() {
    //                let info: MyceliumAdminInfo = resp.json().await?;
    //                return Ok(info);
    //            }
    //        }
    //        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    //    }
    //    Err(anyhow!(
    //        "Timeout waiting for local mycelium administration socket to respond"
    //    ))
    //}

    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
        self.child.wait().await
    }
}

impl Drop for SupervisedMycelium {
    fn drop(&mut self) {
        let _ = self.child.start_kill();

        #[cfg(not(target_os = "linux"))]
        {
            if self.is_ephemeral_key {
                let _ = std::fs::remove_file(&self.key_file_path);
            }
            if self.is_ephemeral_net_key {
                if let Some(ref path) = self.net_key_file_path {
                    let _ = std::fs::remove_file(path);
                }
            }
        }
    }
}

fn is_pubkey(s: &str) -> bool {
    s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit())
}

pub struct MyceliumTransport {
    // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane
    pub setup: tokio::sync::Mutex<MyceliumSetup>,
    pub client: reqwest::Client,
    pub poll_client: reqwest::Client,
    pub messages_endpoint: Url,
    pubkey_basepath: Url,
    local_id: String,
}

impl MyceliumTransport {
    pub async fn connect(setup: MyceliumSetup) -> Result<Self> {
        let rest_url = setup.rest_url.clone();

        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()?;
        let poll_client = reqwest::Client::builder()
            .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
            .build()?;

        let admin: MyceliumAdminInfo = client
            .get(rest_url.join("/api/v1/admin")?)
            .send()
            .await?
            .error_for_status()?
            .json()
            .await
            .context("Querying local Mycelium identity")?;

        Ok(Self {
            setup: tokio::sync::Mutex::new(setup),
            client,
            poll_client,
            messages_endpoint: rest_url.join("/api/v1/messages")?,
            pubkey_basepath: rest_url.join("/api/v1/pubkey/")?,
            local_id: admin.node_pubkey,
        })
    }

    async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> {
        #[derive(Deserialize)]
        struct PublicKeyResponse {
            #[serde(rename = "publicKey")]
            public_key: String,
        }

        let response: PublicKeyResponse = self
            .client
            .get(self.pubkey_basepath.join(&ip.to_string())?)
            .send()
            .await?
            .error_for_status()?
            .json()
            .await?;
        Ok(response.public_key)
    }
}

#[async_trait]
impl Transport for MyceliumTransport {
    fn local_id(&self) -> &str {
        &self.local_id
    }

    async fn resolve_peer(&self, hint: &str) -> Result<String> {
        if is_pubkey(hint) {
            return Ok(hint.to_owned());
        }

        if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() {
            return self.lookup_pubkey(&ip).await;
        }

        let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0)))
            .await
            .context("DNS lookup timed out")??
            .find_map(|addr| match addr.ip() {
                std::net::IpAddr::V6(ip) => Some(ip),
                _ => None,
            })
            .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?;

        self.lookup_pubkey(&ip).await
    }

    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
        let outbound = format_message(peer, message)?;
        Ok(EncodedMessage {
            peer: peer.to_owned(),
            body: serde_json::to_vec(&outbound)?.into(),
        })
    }

    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
        tracing::debug!(payload_bytes = message.body.len(), "Sending message");
        let response = self
            .client
            .post(self.messages_endpoint.clone())
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .body(message.body.clone())
            .send()
            .await?;

        let status = response.status();
        if !status.is_success() {
            tracing::error!(status = %status, "Mycelium rejected outbound message");
            return Err(TransportError::MyceliumRejected {
                peer: message.peer.clone(),
                status,
            }
            .into());
        }

        // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
        // a message that never arrived.
        if tracing::enabled!(tracing::Level::DEBUG) {
            if let Ok(body) = response
                .bytes()
                .await
                .map_err(anyhow::Error::from)
                .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
            {
                if let Some(id) = body.get("id").and_then(|id| id.as_str()) {
                    tracing::debug!(peer = %message.peer, message_id = id, "Mycelium accepted message");
                }
            }
        }
        Ok(())
    }

    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
        tracing::info!(
            endpoint = self.messages_endpoint.as_str(),
            "Starting Mycelium message consumer"
        );

        loop {
            tracing::trace!("Polling Mycelium inbox");

            let response = match self
                .poll_client
                .get(self.messages_endpoint.clone())
                .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
                .send()
                .await
            {
                Ok(resp) => resp,
                Err(error) => {
                    tracing::warn!(error = %error, "Mycelium inbox poll failed");
                    tokio::time::sleep(Duration::from_millis(250)).await;
                    continue;
                }
            };

            let status = response.status();

            if status == StatusCode::NO_CONTENT {
                tracing::trace!("Mycelium inbox is empty");
                tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
                continue;
            }

            if !status.is_success() {
                tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
                tokio::time::sleep(Duration::from_millis(500)).await;
                continue;
            }

            let raw_bytes = match response.bytes().await {
                Ok(bytes) => bytes,
                Err(error) => {
                    tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    continue;
                }
            };

            let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) {
                Ok(msg) => msg,
                Err(error) => {
                    tracing::error!(
                        error = %error,
                        raw_payload = %String::from_utf8_lossy(&raw_bytes),
                        "Failed to decode inbound Mycelium message structure"
                    );
                    tokio::time::sleep(Duration::from_millis(500)).await;
                    continue;
                }
            };

            tracing::debug!(
                id = %inbound_message.id,
                payload_bytes = raw_bytes.len(),
                "Received inbound Mycelium message"
            );

            match parse_message(&inbound_message) {
                Ok((sender, parsed)) => {
                    if let Err(error) = handler(sender, parsed).await {
                        tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
                    }
                }
                Err(transport_error) => {
                    tracing::error!(
                        error = %transport_error,
                        topic = %inbound_message.topic,
                        "Dropping poisoned packet: parsing failed gracefully"
                    );
                    // No sleep required: we successfully popped a malformed packet and want to process the next one immediately
                }
            }
        }
    }

    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
        let mut guard = self.setup.lock().await;
        guard.wait_for_exit().await
    }
}

fn encode_topic(topic: &str) -> String {
    BASE64.encode(topic.as_bytes())
}

fn decode_topic(topic: &str) -> Result<String, TransportError> {
    let bytes = BASE64.decode(topic)?;
    Ok(String::from_utf8(bytes)?)
}

fn encode_payload(payload: &Vec<u8>) -> String {
    BASE64.encode(payload)
}

fn decode_payload(topic: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
    let bytes = BASE64.decode(payload)?;
    match topic {
        TOPIC_HELLO => {
            let msg = bytes_to_hello_request(&bytes)?;
            Ok(ParsedMessage::Hello(msg))
        }
        TOPIC_HELLO_REPLY => {
            let msg = bytes_to_hello_reply(&bytes)?;
            Ok(ParsedMessage::HelloReply(msg))
        }
        TOPIC_SYNC => {
            let msg = bytes_to_sync_envelope(&bytes)?;
            Ok(ParsedMessage::Sync(msg))
        }
        TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
        TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
        t => Err(TransportError::UnknownTopic(t.to_string())),
    }
}

fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> {
    let sender = message.src_pk.clone();

    if sender.is_empty() {
        return Err(TransportError::NoSender);
    }

    let topic = decode_topic(&message.topic)?;
    let parsed = decode_payload(&topic, &message.payload)?;

    Ok((sender, parsed))
}

fn format_message(
    recipient: &str,
    message: &ParsedMessage,
) -> Result<OutboundMessage, TransportError> {
    let (topic, payload_bytes) = match message {
        ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
        ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
        ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
        ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
        ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
    };
    Ok(OutboundMessage {
        dst: Destination {
            pk: recipient.to_owned(),
        },
        topic: encode_topic(topic),
        payload: encode_payload(&payload_bytes),
    })
}

// =====================================================================
// 1. WIRE SHADOWS
// =====================================================================
//
// rmp_serde encodes structs as arrays. Fields added later are appended at the end and
// marked #[serde(default)], so a message from a node that predates them still decodes
// (an absent epoch is the genesis epoch).

/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch.
#[derive(Serialize, Deserialize, Default)]
struct EpochShadow {
    generation: u64,
    id: ByteBuf,
}

fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow {
    EpochShadow {
        generation: epoch.generation,
        id: ByteBuf::from(epoch.id.to_vec()),
    }
}

fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> {
    if shadow.generation == 0 && shadow.id.is_empty() {
        return Ok(Epoch::GENESIS);
    }
    let id = shadow
        .id
        .as_slice()
        .try_into()
        .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?;
    Ok(Epoch {
        generation: shadow.generation,
        id,
    })
}

#[derive(Serialize, Deserialize)]
struct KeepaliveShadow {
    heads_digest: ByteBuf,
    #[serde(default)]
    epoch: EpochShadow,
}

fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
    rmp_serde::to_vec(&KeepaliveShadow {
        heads_digest: ByteBuf::from(msg.heads_digest.to_vec()),
        epoch: epoch_to_shadow(&msg.epoch),
    })
}

fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
    let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?;
    let heads_digest = shadow
        .heads_digest
        .as_slice()
        .try_into()
        .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?;
    Ok(Keepalive {
        heads_digest,
        epoch: epoch_from_shadow(shadow.epoch)?,
    })
}

#[derive(Serialize, Deserialize)]
pub struct HelloRequestShadow {
    #[serde(default)]
    epoch: EpochShadow,
}

#[derive(Serialize, Deserialize)]
pub struct HelloReplyShadow<'a> {
    // Cow allows zero-copy references when saving, but creates an owned String when loading
    pub cluster_id: Cow<'a, str>,
    #[serde(default)]
    pub peers: Vec<String>,
    #[serde(default)]
    epoch: EpochShadow,
}

#[derive(Serialize, Deserialize)]
pub struct SyncEnvelopeShadow<'a> {
    pub cluster_id: Cow<'a, str>,
    #[serde(with = "automerge_msg_codec")]
    pub message: automerge::sync::Message,
    #[serde(default)]
    epoch: EpochShadow,
}

// =====================================================================
// 2. CUSTOM CODEC
// =====================================================================
mod automerge_msg_codec {
    use super::*;
    use automerge::sync::Message;

    pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let bytes = msg.clone().encode();
        Bytes::new(&bytes).serialize(serializer)
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error>
    where
        D: Deserializer<'de>,
    {
        let byte_buf = ByteBuf::deserialize(deserializer)?;
        Message::decode(&byte_buf)
            .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e)))
    }
}

// =====================================================================
// 3. CLEAN CONVERSION INTERFACE
// =====================================================================

// --- HelloRequest ---
pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
    rmp_serde::to_vec(&HelloRequestShadow {
        epoch: epoch_to_shadow(&msg.epoch),
    })
}
pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
    let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
    Ok(HelloRequest {
        epoch: epoch_from_shadow(shadow.epoch)?,
    })
}

// --- HelloReply ---
pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
    let shadow = HelloReplyShadow {
        cluster_id: Cow::Borrowed(&msg.cluster_id),
        peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
        epoch: epoch_to_shadow(&msg.epoch),
    };
    rmp_serde::to_vec(&shadow)
}
pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
    let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
    Ok(HelloReply {
        cluster_id: shadow.cluster_id.into_owned(),
        peers: shadow.peers,
        epoch: epoch_from_shadow(shadow.epoch)?,
    })
}

// --- SyncEnvelope ---
pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
    let shadow = SyncEnvelopeShadow {
        cluster_id: Cow::Borrowed(&msg.cluster_id),
        message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
        epoch: epoch_to_shadow(&msg.epoch),
    };
    rmp_serde::to_vec(&shadow)
}
pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
    let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
    Ok(SyncEnvelope {
        cluster_id: shadow.cluster_id.into_owned(),
        epoch: epoch_from_shadow(shadow.epoch)?,
        message: shadow.message,
    })
}

#[derive(Error, Debug)]
pub enum TransportError {
    #[error("Mycelium delivered a message without a sender public key")]
    NoSender,

    #[error("Base64 decoding failed: {0}")]
    Base64Decode(#[from] base64::DecodeError),

    #[error("UTF-8 decoding failed: {0}")]
    Utf8Decode(#[from] std::string::FromUtf8Error),

    #[error("MessagePack serialization failed: {0}")]
    MessagePackEncode(#[from] rmp_serde::encode::Error),

    #[error("MessagePack deserialization failed: {0}")]
    MessagePackDecode(#[from] rmp_serde::decode::Error),

    #[error("Unknown topic: {0}")]
    UnknownTopic(String),

    #[error("Invalid payload: {0}")]
    InvalidPayload(String),

    #[error("Network HTTP request failed: {0}")]
    Http(#[from] reqwest::Error),

    #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")]
    MyceliumRejected {
        peer: String,
        status: reqwest::StatusCode,
    },
}