jazz-rs 0.2.8

use std::{convert::Infallible, pin::Pin};

use credo::Credo;
use futures::{
    channel::mpsc::{unbounded, UnboundedSender},
    future, Future, FutureExt, Sink, Stream, StreamExt,
};
use jmbl::{ops::OpWithTarget, stream_format::OpSerializer, OpSender, WritableLog};
use litl::Litl;
use ridl::{symm_encr::KeySecret, unauth_symm_encr::UnauthEncryptionStream};
use tlpt::WriteAccess;

use crate::{
    conventions::{claim_to_declare_log_part_of_doc, claim_to_reveal_log_encr_key},
    ScopedDocID,
};

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

impl AppendEncryptedDiff {
    pub fn new(log_encr_key: KeySecret, content: tlpt::Node, 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,
            write_access,
        }
    }
}

impl Sink<Vec<u8>> for AppendEncryptedDiff {
    type Error = Infallible;

    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.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 = Litl> {
    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(
    scoped_doc_id: ScopedDocID,
    content: tlpt::Node,
    credo: Credo,
) -> (WritableLog, impl Future<Output = ()>) {
    // TODO: add compression
    let (tx_ops, rx_ops) = unbounded();

    let forward = async move {
        let log_write_access = content.create_log();
        let log_encr_key = KeySecret::new_random();
        let group_secret = credo
            .current_group_secret_for(&scoped_doc_id.team)
            .expect("Need to have access to group secret to create managed JMBL");

        let secret_claim_id = {
            credo.make_claim_after_frontier(
                &scoped_doc_id.team,
                claim_to_reveal_log_encr_key(group_secret, log_write_access.id(), &log_encr_key),
            )
        }
        .await
        .unwrap();

        credo
            .make_claim(
                &scoped_doc_id.team,
                claim_to_declare_log_part_of_doc(scoped_doc_id.clone(), log_write_access.id()),
                vec![secret_claim_id],
            )
            .await
            .unwrap();

        op_to_litl_stream(rx_ops)
            .map(|litl| {
                let bytes = litl.to_bytes();
                if bytes.is_empty() {
                    panic!("Empty litl to_bytes {:?}", litl)
                }
                Ok(bytes)
            })
            .forward(AppendEncryptedDiff::new(
                log_encr_key,
                content,
                log_write_access,
            ))
            .await
            .unwrap();
    };

    (WritableLog::new(Box::new(SyncAsyncBridge(tx_ops))), forward)
}