p2panda 0.6.1

// SPDX-License-Identifier: MIT OR Apache-2.0

use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};

use futures_util::{Stream, StreamExt, ready};
use p2panda_core::cbor::{DecodeError, EncodeError, decode_cbor, encode_cbor};
use p2panda_core::timestamp::{HybridTimestamp, LamportTimestamp, Timestamp};
use p2panda_core::{Signature, SigningKey, Topic, VerifyingKey};
use p2panda_net::gossip::{GossipHandle, GossipSubscription};
use pin_project::pin_project;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tracing::warn;

use crate::forge::{Forge, OperationForge};

/// Message specification version to create and encode messages for ephemeral streams.
const MESSAGE_VERSION: u64 = 1;

/// Message being disseminated to other nodes via gossip.
///
/// They can be seen as a wrapper around the application's message payloads, providing integrity
/// and provenance guarantees, plus making sure each message is unique with the help of a
/// timestamp.
///
/// Messages are represented as a tuple and are CBOR-encoded as follows:
///
/// ```plain
/// (
///    version[u64],
///    verifying_key[32 bytes],
///    signature[64 bytes],
///    timestamp[u64],
///    lamport_timestamp[u64],
///    body[bytes],
/// )
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
struct WrappedMessage<M> {
    version: u64,
    verifying_key: VerifyingKey,
    signature: Signature,
    timestamp: HybridTimestamp,
    body: M,
}

impl<M> WrappedMessage<M>
where
    M: Serialize + for<'a> Deserialize<'a>,
{
    pub fn new(
        body: M,
        timestamp: HybridTimestamp,
        signing_key: &SigningKey,
    ) -> Result<Self, EncodeError> {
        let verifying_key = signing_key.verifying_key();
        let signature = Self::sign(signing_key, verifying_key, timestamp, &body)?;

        Ok(Self {
            version: MESSAGE_VERSION,
            verifying_key,
            signature,
            timestamp,
            body,
        })
    }

    pub fn from_bytes(bytes: &[u8]) -> Result<Self, WrappedMessageError> {
        // Attempt deserializing message tuple. This fails if the encoding is invalid.
        let (version, verifying_key, signature, timestamp, logical, body): (
            u64,
            VerifyingKey,
            Signature,
            Timestamp,
            LamportTimestamp,
            M,
        ) = decode_cbor(bytes)?;

        let timestamp = HybridTimestamp::from_parts(timestamp, logical);

        // Check supported message version.
        if version != MESSAGE_VERSION {
            return Err(WrappedMessageError::UnsupportedVersion(version));
        }

        let message = Self {
            version,
            verifying_key,
            signature,
            timestamp,
            body,
        };

        // Check message integrity and provenance.
        message.verify()?;

        Ok(message)
    }

    pub fn to_bytes(&self) -> Result<Vec<u8>, EncodeError> {
        let (timestamp, logical) = self.timestamp.to_parts();
        let message = (
            self.version,
            self.verifying_key,
            self.signature,
            timestamp,
            logical,
            &self.body,
        );
        let bytes = encode_cbor(&message)?;
        Ok(bytes)
    }

    pub fn verify(&self) -> Result<(), WrappedMessageError> {
        let (timestamp, logical) = self.timestamp.to_parts();
        let message = (
            self.version,
            self.verifying_key,
            timestamp,
            logical,
            &self.body,
        );

        // Treat encoding error for verifying the signature as an "invalid signature" since
        // the data came from a remote (potentially malicious) node.
        let bytes = encode_cbor(&message).map_err(|_| WrappedMessageError::InvalidSignature)?;

        if !self.verifying_key.verify(&bytes, &self.signature) {
            return Err(WrappedMessageError::InvalidSignature);
        }

        Ok(())
    }

    fn sign(
        signing_key: &SigningKey,
        verifying_key: VerifyingKey,
        timestamp: HybridTimestamp,
        body: &M,
    ) -> Result<Signature, EncodeError> {
        let (timestamp, logical) = timestamp.to_parts();
        let message = (MESSAGE_VERSION, verifying_key, timestamp, logical, body);
        let bytes = encode_cbor(&message)?;
        Ok(signing_key.sign(&bytes))
    }
}

#[derive(Debug, Error)]
enum WrappedMessageError {
    #[error("unsupported message version {0}")]
    UnsupportedVersion(u64),

    #[error("invalid message encoding: {0}")]
    InvalidEncoding(#[from] DecodeError),

    #[error("invalid message signature")]
    InvalidSignature,
}

/// Message coming from an ephemeral stream subscription.
///
/// Ephemeral messages are verified (for integrity and provenance) but not persisted on a system
/// layer. They contain the application's message payloads as well as public key of the author and
/// timestamp at which they were sent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EphemeralMessage<M> {
    topic: Topic,
    inner: WrappedMessage<M>,
}

impl<M> EphemeralMessage<M>
where
    M: Serialize + for<'a> Deserialize<'a>,
{
    /// Associated topic.
    pub fn topic(&self) -> Topic {
        self.topic
    }

    /// Verified author.
    pub fn author(&self) -> VerifyingKey {
        self.inner.verifying_key
    }

    /// Timestamp when this operation was created.
    ///
    /// Microseconds since the UNIX epoch based on system time.
    pub fn timestamp(&self) -> u64 {
        // Only return the wall-clock time to the user as this is the interesting bit, the logical
        // lamport timestamp helps internally with keeping messages unique.
        let (timestamp, _logical) = self.inner.timestamp.to_parts();
        timestamp.into()
    }

    /// Application message payload.
    pub fn body(&self) -> &M {
        &self.inner.body
    }
}

/// Returns publish and subscribe halfs of an ephemeral messaging stream for a given topic.
pub(crate) fn ephemeral_stream<M>(
    topic: Topic,
    forge: OperationForge,
    handle: GossipHandle,
) -> (EphemeralStreamPublisher<M>, EphemeralStreamSubscription<M>) {
    let subscription = handle.subscribe();

    let tx = EphemeralStreamPublisher {
        topic,
        forge,
        inner: handle,
        timestamp: Arc::new(Mutex::new(HybridTimestamp::now())),
        _marker: PhantomData,
    };

    let rx = EphemeralStreamSubscription {
        topic,
        inner: subscription,
        _marker: PhantomData,
    };

    (tx, rx)
}

/// Publish messages into an ephemeral topic stream.
///
/// Any message type `M` can be published as long as it can be encoded into bytes by implementing
/// serde's [`Serialize`] and [`Deserialize`] traits.
///
/// Only currently reachable and subscribed peers will receive published messages.
///
/// ## Example
///
/// ```rust
/// # use p2panda_core::Topic;
/// # use serde::{Serialize, Deserialize};
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// # let node = p2panda::builder().spawn().await?;
/// #
/// let now_playing = Topic::random();
///
/// #[derive(Clone, Debug, Serialize, Deserialize)]
/// struct PlaylistItem {
///     artist: String,
///     song_name: String,
/// }
///
/// let (tx, _rx) = node.ephemeral_stream::<PlaylistItem>(now_playing).await?;
///
/// tx.publish(PlaylistItem {
///     artist: "Richard Cheese".into(),
///     song_name: "Panda".into(),
/// }).await?;
/// #
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Debug)]
pub struct EphemeralStreamPublisher<M> {
    topic: Topic,
    forge: OperationForge,
    inner: GossipHandle,
    timestamp: Arc<Mutex<HybridTimestamp>>,
    _marker: PhantomData<M>,
}

impl<M> EphemeralStreamPublisher<M>
where
    M: Serialize + for<'a> Deserialize<'a>,
{
    /// Associated topic.
    pub fn topic(&self) -> Topic {
        self.topic
    }

    /// Publish a message into an ephemeral topic stream.
    ///
    /// Only currently reachable and subscribed peers will receive published messages.
    pub async fn publish(&self, message: M) -> Result<(), EphemeralPublishError> {
        // The PlumTree implementation for the gossip overlay used by p2panda-net ignores duplicate
        // messages to avoid flooding the network. This can lead to surprises by the users as they
        // expect messages to still arrive, not noticing it's because of a duplicate payload.
        //
        // To help with this we're using a microsecond-precision timestamp + lamport logical clock
        // serving somewhat as a nonce, making sure that every message is guaranteed to be unique.
        let timestamp = {
            let mut timestamp = self
                .timestamp
                .lock()
                .expect("lock poisoned by another thread");
            *timestamp = timestamp.increment();
            *timestamp
        };

        let bytes = {
            let wrapped = WrappedMessage::new(message, timestamp, self.forge.signing_key())?;
            wrapped.to_bytes()?
        };

        self.inner
            .publish(bytes)
            .await
            .map_err(|_err| EphemeralPublishError::BrokenChannel)?;

        Ok(())
    }
}

/// Error occurred when publishing a message to ephemeral topic stream.
#[derive(Debug, Error)]
pub enum EphemeralPublishError {
    /// If this error occurs probably something is wrong with the system.
    #[error("critical encoding error: {0}")]
    Encode(#[from] EncodeError),

    /// Broken / closed communication channel with the internal gossip actor in `p2panda-net`. This
    /// can be due to the actor crashing.
    ///
    /// Users may re-attempt sending the message in case the actor restarted later.
    #[error("error in internal gossip actor occurred")]
    BrokenChannel,
}

/// Subscription to messages arriving from an ephemeral topic stream.
///
/// ## Example
///
/// ```no_run
/// use futures_util::StreamExt;
/// use p2panda_core::Topic;
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// # let node = p2panda::spawn().await?;
/// let topic = Topic::random();
///
/// let (_tx, mut rx) = node.ephemeral_stream::<String>(topic).await?;
///
/// while let Some(stream_event) = rx.next().await {
///     // .. react to received messages
/// }
/// #
/// # Ok(())
/// # }
/// ```
#[pin_project]
pub struct EphemeralStreamSubscription<M> {
    topic: Topic,
    #[pin]
    inner: GossipSubscription,
    _marker: PhantomData<M>,
}

impl<M> EphemeralStreamSubscription<M>
where
    M: Serialize + for<'a> Deserialize<'a>,
{
    /// Associated topic.
    pub fn topic(&self) -> Topic {
        self.topic
    }
}

impl<M> Stream for EphemeralStreamSubscription<M>
where
    M: Serialize + for<'a> Deserialize<'a>,
{
    type Item = EphemeralMessage<M>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match ready!(self.inner.poll_next_unpin(cx)) {
            // Check encoding & supported version and signature during deserialisation.
            Some(Ok(bytes)) => match WrappedMessage::from_bytes(&bytes) {
                Ok(wrapped) => Poll::Ready(Some(EphemeralMessage {
                    topic: self.topic,
                    inner: wrapped,
                })),
                Err(err) => {
                    // Don't bother users with invalid wrapped messages as this type is not public.
                    // Instead we log a warning, in case this reveals a buggy implementation, etc.
                    warn!("invalid ephemeral message received: {err}");
                    Poll::Pending
                }
            },
            // Ignore internal broadcast channel error, this only indicates that the channel
            // dropped a message which we can't do much about on this layer anymore. In the future
            // we want to remove this error type altogether.
            //
            // Related issue: https://github.com/p2panda/p2panda/issues/959
            Some(Err(_)) => Poll::Pending,
            // Internal stream seized.
            None => Poll::Ready(None),
        }
    }
}

#[cfg(test)]
mod tests {
    use p2panda_core::SigningKey;
    use p2panda_core::timestamp::HybridTimestamp;

    use super::WrappedMessage;

    #[test]
    fn encoding() {
        let signing_key = SigningKey::generate();
        let timestamp = HybridTimestamp::now();

        let message_1 = WrappedMessage::new(
            "This message is signed!".to_string(),
            timestamp,
            &signing_key,
        )
        .unwrap();

        let bytes = message_1.to_bytes().unwrap();
        let message_2 = WrappedMessage::from_bytes(&bytes).unwrap();

        assert_eq!(message_1, message_2);
    }

    #[test]
    fn signatures() {
        let signing_key = SigningKey::generate();
        let timestamp = HybridTimestamp::now();

        let message = WrappedMessage::new(
            "This message is signed!".to_string(),
            timestamp,
            &signing_key,
        )
        .unwrap();

        assert!(message.verify().is_ok());
    }
}