use anyhow::Result;
use async_trait::async_trait;
use automerge::ChangeHash;
use automerge::sync::Message;
use sha2::{Digest, Sha256};
use std::boxed::Box;
use std::future::Future;
use std::pin::Pin;

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

pub mod firewall;
pub mod mycelium;

/// Most heads listed in a keepalive; more are sent as a digest. 16 heads are about
/// 520 bytes, well within one Mycelium chunk (1300 bytes).
pub const MAX_LISTED_HEADS: usize = 16;

/// A document's heads, compactly: the heads themselves while there are few, otherwise a
/// digest of them. Either way, two summaries are equal exactly when the heads are (the
/// number of heads decides the form, so equal heads always take the same one).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HeadsSummary {
    /// Sorted.
    Heads(Vec<ChangeHash>),
    /// SHA-256 over the sorted heads.
    Digest([u8; 32]),
}

impl HeadsSummary {
    pub fn of(mut heads: Vec<ChangeHash>) -> Self {
        heads.sort_unstable();
        if heads.len() <= MAX_LISTED_HEADS {
            return Self::Heads(heads);
        }
        let mut hasher = Sha256::new();
        for head in &heads {
            hasher.update(head.0);
        }
        Self::Digest(hasher.finalize().into())
    }

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

/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
#[derive(Debug, Clone)]
pub struct Keepalive {
    /// Only comparable between nodes on the same epoch.
    pub heads: HeadsSummary,
    pub epoch: Epoch,
    /// The sender is ready (same meaning as `SyncEnvelope::ready`).
    pub ready: bool,
    /// The sender would sync with the recipient: counts it as an active member on its
    /// own epoch.
    pub syncs_with_you: bool,
}

#[derive(Debug)]
pub enum ParsedMessage {
    Hello(HelloRequest),
    HelloReply(HelloReply),
    Sync(SyncEnvelope),
    Ping(Keepalive),
    Pong(Keepalive),
}

impl ParsedMessage {
    /// Every message carries the sender's epoch.
    pub fn epoch(&self) -> Epoch {
        match self {
            ParsedMessage::Hello(hello) => hello.epoch,
            ParsedMessage::HelloReply(reply) => reply.epoch,
            ParsedMessage::Sync(envelope) => envelope.epoch,
            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
        }
    }
}

#[derive(Debug, Clone)]
pub struct HelloRequest {
    pub epoch: Epoch,
    /// Proof of the swarm key (`swarm::auth`); empty without one.
    pub auth: Vec<u8>,
}

#[derive(Debug, Clone)]
pub struct HelloReply {
    pub swarm_id: String,
    pub peers: Vec<String>,
    pub epoch: Epoch,
    /// Proof of the swarm key (`swarm::auth`); empty without one.
    pub auth: Vec<u8>,
}

#[derive(Debug, Clone)]
pub struct SyncEnvelope {
    pub swarm_id: String,
    /// Epoch of the document the message was generated from.
    pub epoch: Epoch,
    pub message: Message,
    /// 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 ready: bool,
}

/// A message serialized for the wire. Cloning is cheap (shared buffer).
#[derive(Clone, Debug)]
pub struct EncodedMessage {
    pub peer: String,
    pub body: bytes::Bytes,
}

pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
    + Send
    + Sync
    + 'static;
pub type MessageHandler = Box<HandlerFn>;

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

    /// 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>;
}