jazz-rs 0.7.0

use std::pin::Pin;

use credo::{Credo, Scope};
use futures::{
    channel::mpsc::{unbounded, UnboundedSender},
    future, Future, FutureExt, Sink, Stream, StreamExt,
};
use jmbl::{ops::OpWithTarget, stream_format::OpSerializer, OpSender, WritableLog};
use litl::{to_newln_sep_stream, Val};
use ridl::{symm_encr::KeySecret, unauth_symm_encr::UnauthEncryptionStream};
use caro::{ObjectID, WriteAccess};
use tracing::{trace, trace_span};
use tracing_futures::Instrument;

use crate::{
    conventions::{claim_to_declare_log_part_of_doc, log_key_secret_name, READ_CONTENT},
    doc::DocID,
    ManagedJMBL,
};

struct AppendEncryptedDiff {
    encryption_stream: UnauthEncryptionStream,
    buffer: Vec<u8>,
    flushing: Option<Pin<Box<dyn Future<Output = ()>>>>,
    content: caro::Node,
    log_id: ObjectID,
    write_access: WriteAccess,
}

impl AppendEncryptedDiff {
    pub fn new(
        log_encr_key: KeySecret,
        content: caro::Node,
        log_id: ObjectID,
        write_access: WriteAccess,
    ) -> Self {
        Self {
            // Null nonce is safe IFF the key is unique to this log
            encryption_stream: UnauthEncryptionStream::new(log_encr_key, [0; 12].into()),
            buffer: Vec::new(),
            flushing: None,
            content,
            log_id,
            write_access,
        }
    }
}

impl Sink<Vec<u8>> for AppendEncryptedDiff {
    type Error = std::io::Error;

    fn poll_ready(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn start_send(
        mut self: std::pin::Pin<&mut Self>,
        mut item: Vec<u8>,
    ) -> Result<(), Self::Error> {
        self.encryption_stream.xor_chunk(&mut item);
        self.buffer.extend_from_slice(&item);
        Ok(())
    }

    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        let flushing = match self.flushing {
            Some(ref mut flushing) => flushing,
            None => {
                if self.buffer.is_empty() {
                    return std::task::Poll::Ready(Ok(()));
                } else {
                    let diff = self.content.diff_for_log_append(
                        self.log_id,
                        &self.write_access,
                        &self.buffer,
                    );

                    self.flushing = Some(
                        {
                            let content = self.content.clone();
                            async move {
                                content.apply_new_diff(diff).await;
                            }
                        }
                        .boxed_local(),
                    );
                    self.flushing.as_mut().unwrap()
                }
            }
        };

        match flushing.poll_unpin(cx) {
            std::task::Poll::Ready(()) => {
                self.flushing = None;
                self.buffer.clear();
                std::task::Poll::Ready(Ok(()))
            }
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }

    fn poll_close(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        self.poll_flush(cx)
    }
}

fn op_to_litl_stream<S: Stream<Item = OpWithTarget>>(src: S) -> impl Stream<Item = Val> {
    src.scan(OpSerializer::default(), |serializer, op| {
        future::ready(Some(serializer.serialize(&op)))
    })
}

pub struct SyncAsyncBridge(UnboundedSender<OpWithTarget>);

impl OpSender for SyncAsyncBridge {
    fn send(&mut self, value: OpWithTarget) -> Result<(), String> {
        self.0
            .unbounded_send(value)
            .map_err(|err| format!("SyncAsyncBridge: send failed: {}", err))
    }

    fn test_get_past_sent_ops(&mut self) -> Option<Vec<OpWithTarget>> {
        unimplemented!()
    }
}

pub fn create_log_as_writable_log(
    doc_scope: Scope,
    content: caro::Node,
    // TODO: make this a weak reference to stop the job when ManagedJMBL is dropped
    managed_jmbl: ManagedJMBL,
) -> (WritableLog, impl Future<Output = ()>) {
    // TODO: add compression
    let (tx_ops, rx_ops) = unbounded();

    let writable_log = WritableLog::new(Box::new(SyncAsyncBridge(tx_ops)));
    let writable_log_id = writable_log.log_id;

    let forward = async move {
        let (log_id, log_write_access) = content.create_log(Option::<()>::None).await;
        let log_encr_key = KeySecret::new_random();

        (*managed_jmbl.0)
            .borrow_mut()
            .jmbl_logs_to_tlpt_logs
            .insert(writable_log_id, log_id);

        let secret_claim_ids = doc_scope
            .entrust_secret_to(
                READ_CONTENT,
                &log_key_secret_name(log_id),
                litl::to_val(&log_encr_key).unwrap(),
            )
            .instrument(trace_span!("reveal_log_encr_key", doc_id = ?DocID(doc_scope.id())))
            .await
            .unwrap();

        trace!(secret_claim_ids = ?secret_claim_ids, "After revealing log secret");

        let log_claim_id = doc_scope
            .make_claim_after(
                claim_to_declare_log_part_of_doc(log_id),
                vec![*secret_claim_ids.last().unwrap()],
            )
            .instrument(trace_span!("declare_log_part_of_doc", doc_id = ?DocID(doc_scope.id())))
            .await
            .unwrap();

        trace!(log_claim_id = ?log_claim_id, doc_id = ?DocID(doc_scope.id()), "After making log claim");

        to_newln_sep_stream(op_to_litl_stream(rx_ops))
            .forward(AppendEncryptedDiff::new(
                log_encr_key,
                content,
                log_id,
                log_write_access,
            ))
            .await
            .unwrap();
    };

    (writable_log, forward)
}