use anyhow::Result;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::boxed::Box;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};

pub(crate) use crate::document::replica::Epoch;

use crate::document::ChangeHash;

// Used by the transports that don't authenticate their peers themselves (`tor`).
#[cfg_attr(not(feature = "tor"), allow(dead_code))]
pub(crate) mod challenge;
pub(crate) mod firewall;
pub(crate) mod identity;
#[cfg(feature = "libp2p")]
pub(crate) mod libp2p;
#[cfg(feature = "mycelium")]
pub(crate) mod mycelium;
pub(crate) mod paths;
#[cfg(test)]
pub(crate) mod sim;
pub(crate) mod stats;
#[cfg(feature = "tor")]
pub(crate) mod tor;
pub(crate) mod traffic;
pub(crate) mod wire;

#[cfg(not(any(feature = "mycelium", feature = "libp2p", feature = "tor")))]
compile_error!(
    "continuum needs a transport: build with at least one of the mycelium, libp2p and tor features, e.g. `cargo build --features mycelium`"
);

/// The options of the transports this build has, each checked by its transport when the
/// configuration is loaded (`config::Settings::check`, after the transports enabled):
/// Mycelium's always (its conflicts are errors wherever they come from, as before);
/// libp2p's and Tor's if the node uses them. Options of transports not built aren't
/// checked: they're ignored, and said so (`config::Loaded::unused_transport_options`).
pub(crate) fn check(settings: &crate::config::Settings) -> Result<()> {
    let uses = |name: &str| settings.transports.iter().any(|used| used == name);
    let _ = uses;
    #[cfg(feature = "mycelium")]
    mycelium::check(settings)?;
    #[cfg(feature = "libp2p")]
    if uses("libp2p") {
        libp2p::check(settings)?;
    }
    #[cfg(feature = "tor")]
    if uses("tor") {
        tor::TorTransport::check(settings)?;
    }
    Ok(())
}

/// Starts the transport `name` (one of `config::TRANSPORTS`) as `settings` configure
/// it: its scheme and the transport, or None if `stop` came first (some take a while to
/// start: a supervised Mycelium daemon, arti). `key`: the node key, which some prove
/// themselves.
pub(crate) async fn start(
    name: &str,
    settings: &crate::config::Settings,
    key: &std::sync::Arc<crate::identity::NodeKey>,
    stop: tokio::sync::watch::Receiver<bool>,
) -> Result<Option<(&'static str, std::sync::Arc<dyn Transport>)>> {
    use std::sync::Arc;
    let _ = (key, &stop);
    let started: Option<(&'static str, Arc<dyn Transport>)> = match name {
        #[cfg(feature = "mycelium")]
        "mycelium" => mycelium::start(settings, stop)
            .await?
            .map(|transport| (mycelium::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
        #[cfg(feature = "libp2p")]
        "libp2p" => {
            let _ = stop;
            let config = libp2p::Libp2pConfig::from_settings(settings, false)?;
            Some((
                libp2p::SCHEME,
                Arc::new(libp2p::Libp2pTransport::start(config)?) as Arc<dyn Transport>,
            ))
        }
        #[cfg(feature = "tor")]
        "tor" => tor::TorTransport::from_settings(settings, key, stop)
            .await?
            .map(|transport| (tor::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
        other => anyhow::bail!("transports: no {other} in this build"),
    };
    Ok(started)
}

/// Random identifier of one run of a node (from start to exit).
pub(crate) type RunId = [u8; 16];

/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes
/// on the wire, so 32 of them keep a keepalive within one small packet (a Mycelium
/// chunk: 1300 bytes; see the size test in `mycelium`). There are rarely more heads
/// than nodes that wrote concurrently since the last merging write.
pub(crate) const MAX_LISTED_HEADS: usize = 32;

/// A document's heads, compactly: the heads themselves while there are few, otherwise a
/// digest of them with some of them. Two summaries are equal exactly when the heads are,
/// whatever form each took (nodes with different limits still agree).
#[derive(Debug, Clone)]
pub(crate) enum HeadsSummary {
    /// Sorted.
    Heads(Vec<ChangeHash>),
    Digest {
        /// SHA-256 over the sorted heads.
        digest: [u8; 32],
        /// `MAX_LISTED_HEADS` of the heads, another window of them each time: a receiver
        /// pulls those it lacks, and over a few keepalives every head is listed, instead
        /// of starting the sync over to find out what it lacks.
        sample: Vec<ChangeHash>,
    },
}

/// Where the next sample starts (`HeadsSummary::of`).
static SAMPLE_OFFSET: AtomicUsize = AtomicUsize::new(0);

impl HeadsSummary {
    pub(crate) fn of(mut heads: Vec<ChangeHash>) -> Self {
        heads.sort_unstable();
        if heads.len() <= MAX_LISTED_HEADS {
            return Self::Heads(heads);
        }
        let start = SAMPLE_OFFSET.fetch_add(MAX_LISTED_HEADS, Ordering::Relaxed) % heads.len();
        let sample = heads
            .iter()
            .cycle()
            .skip(start)
            .take(MAX_LISTED_HEADS)
            .copied()
            .collect();
        Self::Digest {
            digest: digest_of(&heads),
            sample,
        }
    }

    /// The digest of the heads, computed if they are listed.
    pub(crate) fn digest(&self) -> [u8; 32] {
        match self {
            Self::Heads(heads) => digest_of(heads),
            Self::Digest { digest, .. } => *digest,
        }
    }

    /// The heads, if all are listed.
    pub(crate) fn heads(&self) -> Option<&[ChangeHash]> {
        match self {
            Self::Heads(heads) => Some(heads),
            Self::Digest { .. } => None,
        }
    }

    /// The heads listed: all of them, or a sample.
    pub(crate) fn listed(&self) -> &[ChangeHash] {
        match self {
            Self::Heads(heads) => heads,
            Self::Digest { sample, .. } => sample,
        }
    }
}

impl PartialEq for HeadsSummary {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::Heads(a), Self::Heads(b)) => a == b,
            _ => self.digest() == other.digest(),
        }
    }
}

impl Eq for HeadsSummary {}

/// SHA-256 over sorted heads.
fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
    let mut hasher = Sha256::new();
    for head in sorted_heads {
        hasher.update(head.0);
    }
    hasher.finalize().into()
}

/// STATUS and STATUS_REPLY. Addressed to one peer: `syncs_with_you` is about the recipient.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct Status {
    /// Only comparable between nodes on the same epoch.
    #[serde(with = "crate::transport::wire::codec::heads")]
    pub(crate) heads: HeadsSummary,
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    /// The sender is ready (same meaning as `SyncEnvelope::ready`).
    pub(crate) ready: bool,
    /// The sender would sync with the recipient: counts it as an active member on its
    /// own epoch.
    pub(crate) syncs_with_you: bool,
    /// The sender is shutting down: its farewell (`keepalive::farewell`). The
    /// recipient holds it dormant at once, rather than after `dormant_after` of silence.
    pub(crate) leaving: bool,
}

/// Cheap to clone: large parts are shared (`bytes::Bytes`).
#[derive(Debug, Clone)]
pub(crate) enum ParsedMessage {
    Hello(HelloRequest),
    HelloReply(HelloReply),
    Sync(SyncEnvelope),
    Status(Status),
    StatusReply(Status),
    Blob(BlobMessage),
    /// Asks the recipient who it is, saying who the sender is (`transport::identity`).
    Who(IdentityProof),
    /// Answers a WHO.
    Me(IdentityProof),
    /// Asks for a PONG with this nonce, over the same path (`transport::paths`).
    Ping(u64),
    /// Answers a PING: its nonce, and when it was answered by the answerer's clock
    /// (`at`), for comparing clocks (`transport::paths`).
    Pong {
        nonce: u64,
        at: crate::clock::UnixMillis,
    },
    /// Asks a member how it is linked to its peers, for a map of the swarm
    /// (`swarm::links`).
    Links(LinksRequest),
    /// Answers a LINKS.
    LinksReply(LinksReply),
}

/// A node's proof that it holds the key of its node ID (`identity`), sending from its
/// transport address to the recipient's: a signature over both addresses
/// (`transport::identity::signed`). WHO and ME end at the identity layer; the rest of
/// the node never sees them.
#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct IdentityProof {
    /// The sender's public node key.
    #[serde(with = "crate::transport::wire::codec::bytes32")]
    pub(crate) key: [u8; 32],
    #[serde(with = "crate::transport::wire::codec::bytes64")]
    pub(crate) signature: [u8; 64],
}

impl std::fmt::Debug for IdentityProof {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("IdentityProof")
            .field("key", &Hex(&self.key))
            .finish_non_exhaustive()
    }
}

/// Content-addressed bytes that any node holding them can serve, in chunks
/// (`swarm::blob`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub(crate) enum BlobId {
    /// An Automerge change, by its hash: e.g. an epoch's snapshot change.
    Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash),
    /// Bytes by their BLAKE3 hash: e.g. the content of a remake.
    Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]),
    /// The changes of a `DeltaSpec`, by its ID (`DeltaSpec::id`). The requests carry
    /// the spec, so any node holding its `to` can serve it.
    Delta(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]),
}

impl std::fmt::Display for BlobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BlobId::Change(hash) => write!(f, "change:{hash}"),
            BlobId::Content(hash) => write!(f, "content:{}", Hex(hash)),
            BlobId::Delta(id) => write!(f, "delta:{}", Hex(id)),
        }
    }
}

/// A blob request or one of its answers. Each is a message of its own, so no message
/// is larger than one chunk.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct BlobMessage {
    /// The sender's epoch (every message carries it).
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    pub(crate) id: BlobId,
    pub(crate) body: BlobBody,
}

#[derive(Clone, Serialize, Deserialize)]
pub(crate) enum BlobBody {
    /// Asks for these chunks; for the manifest if empty. `delta`: the spec of a
    /// `BlobId::Delta`, which the ID alone doesn't say.
    Request {
        chunks: Vec<u32>,
        delta: Option<DeltaSpec>,
    },
    /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter.
    Manifest {
        size: u64,
        chunk_size: u32,
        #[serde(with = "crate::transport::wire::codec::hashes32")]
        hashes: Vec<[u8; 32]>,
    },
    Chunk {
        index: u32,
        #[serde(with = "crate::transport::wire::codec::shared_bytes")]
        data: bytes::Bytes,
    },
    /// The sender doesn't have the blob.
    Missing,
}

/// Sizes instead of contents: logged messages stay short.
impl std::fmt::Debug for BlobBody {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BlobBody::Request { chunks, delta } => f
                .debug_struct("Request")
                .field("chunks", chunks)
                .field("delta", &delta.is_some())
                .finish(),
            BlobBody::Manifest {
                size,
                chunk_size,
                hashes,
            } => f
                .debug_struct("Manifest")
                .field("size", size)
                .field("chunk_size", chunk_size)
                .field("chunks", &hashes.len())
                .finish(),
            BlobBody::Chunk { index, data } => f
                .debug_struct("Chunk")
                .field("index", index)
                .field("bytes", &data.len())
                .finish(),
            BlobBody::Missing => f.write_str("Missing"),
        }
    }
}

/// What a message is, without its contents.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum MessageKind {
    Hello,
    HelloReply,
    Sync,
    Status,
    StatusReply,
    Blob,
    Who,
    Me,
    Ping,
    Pong,
    Links,
    LinksReply,
}

impl ParsedMessage {
    pub(crate) fn kind(&self) -> MessageKind {
        match self {
            ParsedMessage::Hello(_) => MessageKind::Hello,
            ParsedMessage::HelloReply(_) => MessageKind::HelloReply,
            ParsedMessage::Sync(_) => MessageKind::Sync,
            ParsedMessage::Status(_) => MessageKind::Status,
            ParsedMessage::StatusReply(_) => MessageKind::StatusReply,
            ParsedMessage::Blob(_) => MessageKind::Blob,
            ParsedMessage::Who(_) => MessageKind::Who,
            ParsedMessage::Me(_) => MessageKind::Me,
            ParsedMessage::Ping(_) => MessageKind::Ping,
            ParsedMessage::Pong { .. } => MessageKind::Pong,
            ParsedMessage::Links(_) => MessageKind::Links,
            ParsedMessage::LinksReply(_) => MessageKind::LinksReply,
        }
    }

    /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them
    /// but WHO, ME, PING and PONG (which the node never sees).
    pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> {
        match self {
            ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
            ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
            ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
            ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
                Some((status.epoch, status.epoch_seq))
            }
            ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
            ParsedMessage::Links(request) => Some((request.epoch, request.epoch_seq)),
            ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)),
            ParsedMessage::Who(_)
            | ParsedMessage::Me(_)
            | ParsedMessage::Ping(_)
            | ParsedMessage::Pong { .. } => None,
        }
    }
}

/// LINKS: what a member is asked for a map of the swarm (`swarm::links`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct LinksRequest {
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    /// Repeated in the reply: which request it answers.
    pub(crate) nonce: u64,
    /// For this long (seconds), a LINKS_REPLY again whenever the links change (a
    /// subscription; the next LINKS renews it). 0: this reply only, and a subscription
    /// ends (a relayed one runs out instead).
    pub(crate) subscribe_secs: u32,
    /// Members the receiver is to ask on the sender's behalf, which the sender can't
    /// reach, passing their links on (relayed, `swarm::links`); none: the receiver's own.
    pub(crate) of: Vec<String>,
    /// How long relayed links may be held to send several in one LINKS_REPLY, by the
    /// receiver and the hops after it together (milliseconds; `swarm::links`).
    pub(crate) batch_ms: u32,
    /// How many more nodes a relayed request may go through, after the receiver.
    pub(crate) hops: u8,
}

/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct LinksReply {
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    /// The request's.
    pub(crate) nonce: u64,
    /// The sender's own, if asked for; at most `swarm::links::MAX_LINKS`.
    pub(crate) links: Vec<Link>,
    /// Other members' links, passed on (`LinksRequest::of`).
    pub(crate) relayed: Vec<RelayedLinks>,
}

/// A member's links as a LINKS_REPLY passes them on.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RelayedLinks {
    pub(crate) member: String,
    /// How long ago the member reported them (milliseconds).
    pub(crate) age_ms: u32,
    pub(crate) links: Vec<Link>,
}

/// One of a member's links, in a LINKS_REPLY.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Link {
    /// The peer's node ID.
    pub(crate) peer: String,
    /// The transport its messages go over now (the scheme of the path in use), if any.
    pub(crate) transport: Option<String>,
    /// Only through a relay (libp2p).
    pub(crate) relayed: bool,
    /// The path in use is down: none to the peer answers (`paths`), whatever its
    /// connection was.
    pub(crate) down: bool,
    /// The round trip on that path, in microseconds, if measured.
    pub(crate) rtt_us: Option<u32>,
    /// Silent for long (`peers.md`, dormancy).
    pub(crate) dormant: bool,
    /// The peer's other transports (`swarm::links::MAX_OTHERS` at most): where else
    /// it proved itself, those paths up.
    pub(crate) others: Vec<OtherPath>,
    /// How far the peer's clock is ahead of the member's (`paths::ClockOffset`), if
    /// measured: with every member's, a map of the swarm shows whose clock is off.
    pub(crate) clock: Option<paths::ClockOffset>,
}

/// A transport of a link besides the one in use, in a LINKS_REPLY.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct OtherPath {
    /// Its scheme (`mycelium`, `libp2p`, `tor`).
    pub(crate) transport: String,
    /// The round trip of its fastest path, in microseconds, if any is measured.
    pub(crate) rtt_us: Option<u32>,
    /// Only through a relay (libp2p): no direct connection at any of its paths.
    pub(crate) relayed: bool,
}

#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct HelloRequest {
    /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in
    /// another swarm doesn't answer.
    pub(crate) swarm_id: Option<String>,
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    /// Proof of the swarm key (`swarm::auth`); empty without one.
    #[serde(with = "serde_bytes")]
    pub(crate) auth: Vec<u8>,
    /// The sender is ready (same meaning as `SyncEnvelope::ready`).
    pub(crate) ready: bool,
    /// Random per run of the sender; the reply echoes it.
    #[serde(with = "crate::transport::wire::codec::run_id")]
    pub(crate) run_id: RunId,
    /// The handshake's number: the sender numbers the handshakes it starts in a run,
    /// with all peers, from 1. A retry carries the same number, so the recipient resets
    /// its sync session only for a new handshake, not for every resend or late copy.
    pub(crate) handshake: u32,
}

/// A peer a HELLO_REPLY lists: who it is, and where the sender reaches it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct ListedPeer {
    pub(crate) id: String,
    /// As `Transport::addresses_of` gives them; hints for the recipient.
    pub(crate) addresses: Vec<String>,
}

/// Where messages to a peer go now (`Transport::destination`).
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Destination {
    /// Where messages to the peer go: its best path, as paths show it (`Paths::location`;
    /// `stats::PathInfo::address`).
    pub(crate) address: String,
    /// The peer proved itself there; else it is only said to be there, and sends wait
    /// for its proof.
    pub(crate) proven: bool,
    /// The transport's connection to it, if the transport has connections.
    pub(crate) connection: Option<Connection>,
    /// While it is connected only through a relay: how hole punching went.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) punching: Option<Punching>,
    /// Every address it proved itself at, best first, as the paths layer measures them
    /// (`transport::paths`).
    pub(crate) paths: Vec<stats::PathInfo>,
}

/// Hole punching to a peer connected only through a relay (libp2p), once a punch failed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Punching {
    /// Punches that failed in a row.
    pub(crate) failed: usize,
    /// When it is tried again, in seconds from now (0: due).
    pub(crate) next_in_secs: u64,
}

/// How a transport with connections (libp2p) is connected to a peer.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
// Only libp2p and Tor have connections; without both (the cargo features) nothing makes
// one.
#[cfg_attr(not(any(feature = "libp2p", feature = "tor")), allow(dead_code))]
pub(crate) enum Connection {
    /// libp2p's to the peer itself; Tor's always (an onion circuit).
    Direct,
    /// Only through a relay (libp2p).
    Relayed,
}

impl Connection {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Direct => "direct",
            Self::Relayed => "relayed",
        }
    }
}

#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct HelloReply {
    pub(crate) swarm_id: String,
    pub(crate) peers: Vec<ListedPeer>,
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// See `SyncEnvelope::epoch_seq`.
    pub(crate) epoch_seq: u32,
    /// Proof of the swarm key (`swarm::auth`); empty without one.
    #[serde(with = "serde_bytes")]
    pub(crate) auth: Vec<u8>,
    /// The sender is ready.
    pub(crate) ready: bool,
    /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the
    /// recipient (delivered late, or queued while it was down) is ignored.
    #[serde(with = "crate::transport::wire::codec::run_id")]
    pub(crate) run_id: RunId,
    /// How long before this HELLO the sender had last heard from the recipient, if ever
    /// (in milliseconds): tells a restarted node how long it was away.
    pub(crate) away_ms: Option<u64>,
    /// The sender's own run (as its HELLOs carry it): a new one means it restarted.
    #[serde(with = "crate::transport::wire::codec::run_id")]
    pub(crate) sender_run_id: RunId,
}

/// Bytes shown as lowercase hex, e.g. run IDs in logs.
pub(crate) struct Hex<'a>(pub(crate) &'a [u8]);

impl std::fmt::Display for Hex<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.iter().try_for_each(|byte| write!(f, "{byte:02x}"))
    }
}

impl std::fmt::Debug for Hex<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(self, f)
    }
}

/// Whether a handshake carries a swarm-key proof, not the proof: logged messages don't
/// hand out tags, and stay short.
fn auth_shown(auth: &[u8]) -> &'static str {
    if auth.is_empty() { "none" } else { "present" }
}

impl std::fmt::Debug for HelloRequest {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HelloRequest")
            .field("swarm_id", &self.swarm_id)
            .field("epoch", &self.epoch)
            .field("epoch_seq", &self.epoch_seq)
            .field("auth", &auth_shown(&self.auth))
            .field("ready", &self.ready)
            .field("run_id", &Hex(&self.run_id))
            .field("handshake", &self.handshake)
            .finish()
    }
}

impl std::fmt::Debug for HelloReply {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HelloReply")
            .field("swarm_id", &self.swarm_id)
            .field("peers", &self.peers.len())
            .field("epoch", &self.epoch)
            .field("epoch_seq", &self.epoch_seq)
            .field("auth", &auth_shown(&self.auth))
            .field("ready", &self.ready)
            .field("run_id", &Hex(&self.run_id))
            .field("away_ms", &self.away_ms)
            .field("sender_run_id", &Hex(&self.sender_run_id))
            .finish()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct SyncEnvelope {
    pub(crate) swarm_id: String,
    /// Epoch of the document the message was generated from.
    #[serde(with = "crate::transport::wire::codec::epoch")]
    pub(crate) epoch: Epoch,
    /// How many times the sender changed epoch in its current run, when it sent this
    /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
    /// ignores an epoch older than the newest it has seen in the same session.
    pub(crate) epoch_seq: u32,
    pub(crate) body: SyncBody,
    /// The sender was ready when it generated the message: its document is the swarm's,
    /// not one it is still filling (joining, rejoining). Only such a peer can tell a
    /// catching-up node that it has everything; an empty document says nothing else.
    pub(crate) ready: bool,
}

/// What a sync envelope carries.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) enum SyncBody {
    /// An Automerge sync message.
    Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload),
    /// Instead of a message too large to send in one piece: fetch these changes as a
    /// blob (`BlobId::Delta`), then sync on.
    Delta(DeltaSpec),
}

/// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of
/// a document at `to`. Every node holding `to` produces the same bytes for it
/// (`document::changes::delta_bytes`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct DeltaSpec {
    /// Sorted.
    #[serde(with = "crate::transport::wire::codec::change_hashes")]
    pub(crate) from: Vec<ChangeHash>,
    /// Sorted.
    #[serde(with = "crate::transport::wire::codec::change_hashes")]
    pub(crate) to: Vec<ChangeHash>,
}

impl DeltaSpec {
    pub(crate) fn new(mut from: Vec<ChangeHash>, mut to: Vec<ChangeHash>) -> Self {
        from.sort_unstable();
        from.dedup();
        to.sort_unstable();
        to.dedup();
        Self { from, to }
    }

    /// Names the spec as a blob: SHA-256 over both lists.
    pub(crate) fn id(&self) -> [u8; 32] {
        let mut hasher = Sha256::new();
        hasher.update(b"continuum/delta/1");
        for heads in [&self.from, &self.to] {
            hasher.update((heads.len() as u32).to_be_bytes());
            for head in heads {
                hasher.update(head.0);
            }
        }
        hasher.finalize().into()
    }
}

/// A sync message as the document's engine encodes it (`document::sync`): bytes to the
/// transports. Decoded only where it is applied (`document::sync::decode`), after the sender
/// proved to be a member and off the consumer: a large message from anyone else costs
/// nothing but its bytes.
#[derive(Clone)]
pub(crate) struct SyncPayload(pub(crate) bytes::Bytes);

/// Sizes instead of contents: logged messages stay short.
impl std::fmt::Debug for SyncPayload {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "SyncPayload({} bytes)", self.0.len())
    }
}

/// A message serialized for the wire. Cloning is cheap (shared buffer).
#[derive(Clone, Debug)]
pub(crate) struct EncodedMessage {
    pub(crate) peer: String,
    /// Where the transport sends it: `peer` itself for a transport whose identities are
    /// addresses, the peer's address under the identity layer (`transport::identity`).
    pub(crate) address: String,
    pub(crate) body: bytes::Bytes,
    pub(crate) kind: MessageKind,
    /// Bytes the message takes on the wire (without the transport's own framing).
    pub(crate) payload_len: usize,
    /// The message itself, under the paths layer (`transport::paths`): to encode it
    /// again for another transport when a send there fails. None from a transport.
    pub(crate) message: Option<std::sync::Arc<ParsedMessage>>,
}

/// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`).
pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
    + Send
    + Sync
    + 'static;
pub(crate) type MessageHandler = Box<HandlerFn>;

/// A transport identity (or location) as an address, `<scheme>:<it>`, the way paths and
/// the identity layer show it: for a transport's own log lines, so that a peer reads the
/// same in every one.
pub(crate) struct Addressed<T>(pub(crate) &'static str, pub(crate) T);

impl<T: std::fmt::Display> std::fmt::Display for Addressed<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}:{}", self.0, self.1)
    }
}

/// Whether `message` is a PING or PONG. The paths layer times them, so a transport's
/// consumer hands them on as soon as they are read (`hand_on_probe`), not after the
/// messages read before them are handled (which can take a while: sync).
pub(crate) fn is_probe(message: &ParsedMessage) -> bool {
    matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong { .. })
}

/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own; `scheme` is the
/// transport's, for the log.
pub(crate) fn hand_on_probe(
    scheme: &'static str,
    handler: &std::sync::Arc<HandlerFn>,
    sender: String,
    message: ParsedMessage,
    payload_len: usize,
) {
    let handler = std::sync::Arc::clone(handler);
    tokio::spawn(async move {
        if let Err(error) = handler(sender.clone(), message, payload_len).await {
            tracing::warn!(peer = %Addressed(scheme, &sender), error = %format!("{error:#}"), "Failed to process inbound message");
        }
    });
}

#[async_trait]
pub(crate) trait Transport: Send + Sync {
    /// This node's identity on the transport (for Mycelium: its public key).
    fn local_id(&self) -> &str;

    /// Whether `id` can be a node's identity on this transport (for Mycelium: a public
    /// key). Identities other peers name to us (in a HELLO_REPLY) are checked with it.
    fn is_peer_id(&self, id: &str) -> bool;

    /// How long the transport tries to deliver a message before dropping it: after that,
    /// a message sent is delivered or lost for good.
    fn delivery_limit(&self) -> std::time::Duration;

    /// The first wait for an answer to a sync message (`Timing::sync_reply_timeout`): by
    /// default `SYNC_REPLY_TIMEOUT`, longer on a transport whose round trips take
    /// seconds.
    fn sync_reply_timeout(&self) -> std::time::Duration {
        crate::state::SYNC_REPLY_TIMEOUT
    }

    /// Where this node can be reached, as addresses other nodes can use: published in
    /// the member list (`swarm::members`). By default its identity, for a transport
    /// whose identities are addresses (Mycelium: a public key).
    fn local_addresses(&self) -> Vec<String> {
        vec![self.local_id().to_owned()]
    }

    /// For a transport under the identity layer: the transport identity at `target`, an
    /// address of this transport without its scheme, noting where it is if `target` says
    /// more than who (as a libp2p multiaddr ending in the peer ID does). None if
    /// `target` isn't one of this transport's. By default `target` itself, for a
    /// transport whose identities are addresses.
    fn locate(&self, target: &str) -> Option<String> {
        self.is_peer_id(target).then(|| target.to_owned())
    }

    /// For a transport under the identity layer: where the node with transport identity
    /// `target` is, as addresses others can use (`locate` reads them). By default
    /// `target` itself, as for `locate`.
    fn locations(&self, target: &str) -> Vec<String> {
        vec![target.to_owned()]
    }

    /// For a transport under the identity layer: the addresses (`<scheme>:<transport
    /// identity>`, as paths keep them) `peer` proved itself at or is said to be at; this
    /// node's own, for its own node ID. By default its identity.
    fn identities_of(&self, peer: &str) -> Vec<String> {
        vec![peer.to_owned()]
    }

    /// Where `peer` can be reached, as far as this node knows: passed on to others (in
    /// a HELLO_REPLY). By default its identity, as for `local_addresses`.
    fn addresses_of(&self, peer: &str) -> Vec<String> {
        vec![peer.to_owned()]
    }

    /// The transports this node is on, as the schemes of their addresses
    /// (`swarm::topology`), whether or not it has an address on each (one that only
    /// connects out has none). By default the empty scheme, for a transport whose
    /// identities are addresses.
    fn transports(&self) -> Vec<String> {
        vec![String::new()]
    }

    /// Seeds the transport found by itself since last asked (libp2p: nodes on the LAN),
    /// for a node that is joining. By default none.
    fn discovered_seeds(&self) -> Vec<String> {
        Vec::new()
    }

    /// Has the transport tell `found` whenever it finds a seed by itself
    /// (`discovered_seeds`). By default it never does.
    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
        let _ = found;
    }

    /// Has the transport tell `reachable` whenever others may reach this node where they
    /// couldn't before (libp2p: a relay granted it a slot, or AutoNAT found it public).
    /// By default it never does.
    fn on_reachable(&self, reachable: std::sync::Arc<tokio::sync::Notify>) {
        let _ = reachable;
    }

    /// Whether this node has a way to reach `peer` at all: an address of it on one of
    /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`).
    /// By default yes, as for `local_addresses`.
    fn can_reach(&self, peer: &str) -> bool {
        let _ = peer;
        true
    }

    /// Where others say `peer` can be reached (the member list, a HELLO_REPLY): hints,
    /// to be checked before they are relied on. By default nothing to learn, as for
    /// `local_addresses`.
    fn learn_addresses(&self, peer: &str, addresses: &[String]) {
        let _ = (peer, addresses);
    }

    /// The relays listed for the swarm (`swarm::members::relays`), as addresses: under
    /// the identity layer, each transport is given those with its scheme, without it.
    /// By default nothing to learn.
    fn learn_relays(&self, addresses: &[String]) {
        let _ = addresses;
    }

    /// Where messages to `peer` go now. None if nowhere, and by default, for a
    /// transport whose identities are addresses (`peer` is where they go).
    fn destination(&self, peer: &str) -> Option<Destination> {
        let _ = peer;
        None
    }

    /// Where messages to `target` (a transport identity) go now, as an address that
    /// names `target` too: the remote address of the connection in use, say. By default
    /// nowhere but `target` itself (Mycelium).
    fn location(&self, target: &str) -> Option<String> {
        let _ = target;
        None
    }

    /// How this node is connected to `target` (a transport identity), if it is. By
    /// default not at all, for a transport without connections (Mycelium).
    fn connection(&self, target: &str) -> Option<Connection> {
        let _ = target;
        None
    }

    /// How hole punching to `target` went, if a punch failed and it is still connected
    /// only through a relay (libp2p). By default never.
    fn punching(&self, target: &str) -> Option<Punching> {
        let _ = target;
        None
    }

    /// The smoothed round-trip time of the path messages to `peer` go over, if
    /// measured (`transport::paths`). By default none.
    fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
        let _ = peer;
        None
    }

    /// How far `peer`'s clock is ahead of ours, as its PONGs say (`paths::ClockOffset`):
    /// the narrowest bound of its paths. By default unknown.
    fn clock_offset(&self, peer: &str) -> Option<paths::ClockOffset> {
        let _ = peer;
        None
    }

    /// Whether the transport answers PINGs with PONGs itself (the paths layer,
    /// `transport::paths`); if not (a bare transport, in tests), the node does. By
    /// default not.
    fn answers_pings(&self) -> bool {
        false
    }

    /// Figures about the transport itself (`stats`). By default none.
    fn stats(&self) -> stats::TransportStats {
        stats::TransportStats::default()
    }

    /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's
    /// identity, as `local_id` gives it on that node.
    async fn resolve_seed(&self, seed: &str) -> Result<String>;

    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage>;
    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>;

    /// Sends an outbound message payload to a targeted peer on a specific topic.
    async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> {
        let encoded = self.encode(peer, message)?;
        self.send_encoded(&encoded).await
    }

    /// Starts the long-polling inbound event loops, driving messages to the provided handler closure.
    async fn start_consumer(&self, handler: MessageHandler) -> Result<()>;

    /// Asynchronously monitors for fatal network client or sub-process failures.
    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus>;
}

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

    /// Two summaries are equal exactly when the heads are, in any order, whichever form
    /// (listed or digest) each took.
    #[test]
    fn heads_summary_equality_is_set_equality() {
        for_seeds(300, |rng| {
            let n = rng.between(0, 2 * MAX_LISTED_HEADS + 2);
            let heads: Vec<ChangeHash> = (0..n).map(|_| ChangeHash(rng.bytes32())).collect();
            let summary = HeadsSummary::of(heads.clone());
            assert_eq!(summary.heads().is_some(), n <= MAX_LISTED_HEADS);

            let mut shuffled = heads.clone();
            rng.shuffle(&mut shuffled);
            assert_eq!(summary, HeadsSummary::of(shuffled));
            assert_eq!(
                summary,
                HeadsSummary::Digest {
                    digest: summary.digest(),
                    sample: Vec::new(),
                }
            );
            // A sample is some of the heads.
            assert!(summary.listed().iter().all(|head| heads.contains(head)));
            assert_eq!(summary.listed().len(), n.min(MAX_LISTED_HEADS));

            let mut other = heads.clone();
            match rng.below(3) {
                1 if !other.is_empty() => {
                    let i = rng.below(other.len());
                    other.remove(i);
                }
                2 if !other.is_empty() => {
                    let i = rng.below(other.len());
                    other[i] = ChangeHash(rng.bytes32());
                }
                _ => other.push(ChangeHash(rng.bytes32())),
            }
            assert_ne!(summary, HeadsSummary::of(other));
        });
    }

    /// Successive samples cover every head.
    #[test]
    fn samples_rotate_over_all_heads() {
        let heads: Vec<ChangeHash> = (0..100u8).map(|n| ChangeHash([n; 32])).collect();
        let mut seen = std::collections::HashSet::new();
        for _ in 0..heads.len().div_ceil(MAX_LISTED_HEADS) + 1 {
            seen.extend(HeadsSummary::of(heads.clone()).listed().iter().copied());
        }
        // Other tests sample concurrently and move the window too, so allow a few more.
        for _ in 0..100 {
            if seen.len() == heads.len() {
                break;
            }
            seen.extend(HeadsSummary::of(heads.clone()).listed().iter().copied());
        }
        assert_eq!(seen.len(), heads.len());
    }
}