use anyhow::Result;
use async_trait::async_trait;
use automerge::sync::Message;
use std::boxed::Box;
use std::future::Future;
use std::pin::Pin;

pub use crate::replica::Epoch;

pub type HeadsDigest = [u8; 32];

#[derive(Debug, Clone)]
pub struct Keepalive {
    pub heads_digest: HeadsDigest,
    /// The digest is only comparable between nodes on the same epoch.
    pub epoch: Epoch,
}

#[derive(Debug)]
pub enum ParsedMessage {
    Hello(HelloRequest),
    HelloReply(HelloReply),
    Sync(SyncEnvelope),
    Ping(Keepalive),
    Pong(Keepalive),
    /// Asks a peer for its document of a newer epoch (late or new nodes).
    SnapshotRequest(SnapshotRequest),
    SnapshotReply(SnapshotReply),
}

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,
            ParsedMessage::SnapshotRequest(request) => request.epoch,
            ParsedMessage::SnapshotReply(reply) => reply.epoch,
        }
    }
}

#[derive(Debug, Clone)]
pub struct HelloRequest {
    pub epoch: Epoch,
}

#[derive(Debug, Clone)]
pub struct HelloReply {
    pub cluster_id: String,
    pub peers: Vec<String>,
    pub epoch: Epoch,
}

#[derive(Debug, Clone)]
pub struct SyncEnvelope {
    pub cluster_id: String,
    /// Epoch of the document the message was generated from.
    pub epoch: Epoch,
    pub message: Message,
}

#[derive(Debug, Clone)]
pub struct SnapshotRequest {
    /// The requester's epoch.
    pub epoch: Epoch,
    /// The epoch whose document is wanted.
    pub wanted: Epoch,
    /// Byte offset of the wanted chunk in the saved document.
    pub offset: u64,
}

/// One chunk of the sender's saved document, or a refusal.
#[derive(Clone)]
pub struct SnapshotReply {
    /// The sender's epoch; equals `wanted` unless refused.
    pub epoch: Epoch,
    pub wanted: Epoch,
    /// SHA-256 of the whole saved document; identifies the copy chunks come from.
    pub digest: [u8; 32],
    /// Length of the whole saved document.
    pub total: u64,
    pub offset: u64,
    pub data: Vec<u8>,
    /// Non-empty if the sender won't serve the document, with the reason.
    pub refused: String,
}

impl std::fmt::Debug for SnapshotReply {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SnapshotReply")
            .field("epoch", &self.epoch)
            .field("wanted", &self.wanted)
            .field("total", &self.total)
            .field("offset", &self.offset)
            .field("data_bytes", &self.data.len())
            .field("refused", &self.refused)
            .finish()
    }
}

/// 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 user-supplied peer hint (e.g. a seed) to a peer identity.
    async fn resolve_peer(&self, hint: &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>;
}