jazz-rs 0.5.0

use futures::{channel::mpsc::Receiver, future, Stream, StreamExt, TryStreamExt};
use jmbl::{
    ops::OpWithTarget,
    stream_format::{OpDeserializer, StreamFormatReadError},
};
use litl::{ReadNewlnSepStreamError, Val};
use ridl::{symm_encr::KeySecret, unauth_symm_encr::UnauthEncryptionStream};
use thiserror::Error;
use tlpt::Diff;
use tracing::trace;

struct DecryptingDiffReader {
    receiver: Receiver<Diff>,
    decryption_stream: UnauthEncryptionStream,
}

impl DecryptingDiffReader {
    pub fn new(receiver: Receiver<Diff>, log_encr_key: KeySecret) -> Self {
        Self {
            receiver,
            // Null nonce is safe IFF the key is unique to this log
            decryption_stream: UnauthEncryptionStream::new(log_encr_key, [0; 12].into()),
        }
    }
}

// this should make it possible to use the DecryptingReader as an async reader
impl Stream for DecryptingDiffReader {
    type Item = Result<Vec<u8>, futures::io::Error>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        match self.receiver.poll_next_unpin(cx) {
            std::task::Poll::Ready(Some(diff)) => match diff {
                Diff::Log(log_diff) => {
                    let mut append = log_diff.append;
                    self.decryption_stream.xor_chunk(&mut append);
                    std::task::Poll::Ready(Some(Ok(append)))
                }
                _ => panic!("Expected log diff"),
            },
            std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }
}

fn op_stream<S: Stream<Item = Result<Val, ReadNewlnSepStreamError>>>(
    values: S,
) -> impl Stream<Item = Result<OpWithTarget, LogStreamError>> {
    values.scan(OpDeserializer::new(), |deserializer, value| {
        future::ready(Some(
            value
                .map_err(Into::into)
                .and_then(|litl| deserializer.deserialize(litl).map_err(Into::into)),
        ))
    })
}

pub fn log_stream(
    receiver: Receiver<Diff>,
    log_encr_key: KeySecret,
) -> impl Stream<Item = Result<OpWithTarget, LogStreamError>> {
    // TODO: add decompression
    op_stream(litl::read_newln_sep_stream(
        DecryptingDiffReader::new(receiver, log_encr_key).into_async_read(),
    ))
    .inspect_ok(|op| {
        let age = ti64::now().0 - op.op.time.0;
        trace!(age_ms = age, "Op age");
    })
}

#[derive(Error, Debug)]
pub enum LogStreamError {
    #[error(transparent)]
    StreamFormatReadError(#[from] StreamFormatReadError),
    #[error(transparent)]
    ReadError(#[from] ReadNewlnSepStreamError),
}