macltok

//! a browser-side slice of macltok, login through to a live sync, exported for

//! wasm-bindgen so both the worker tests and the size measurement reach the same code

use std::{pin::pin, sync::Arc};

use futures_channel::{mpsc, oneshot};

use futures_util::{

    StreamExt,

    future::{Either, select},

};

use gloo_timers::future::TimeoutFuture;

use macltok::{

    ClientHandle, EncryptedSessionStore, EngineConfig, Error, Event, EventStream, MessageContent,

    PasswordLogin, RoomMembership, SendState, Session, StoreConfig, TimelineContent, TimelineDiff,

    TimelineId, TimelineItem, imbl::Vector,

};

use wasm_bindgen::prelude::*;

const PHASE_TIMEOUT_MS: f64 = 30_000.0;

/// logs in, then proves a real sliding sync by finding the seeded room, reading the

/// seeded message out of its timeline, and seeing a message of its own come back from

/// the server

#[wasm_bindgen]

pub async fn smoke(

    homeserver: String,

    username: String,

    password: String,

    store_name: String,

    room_name: String,

    seeded: String,

) -> Result<(), JsError> {

    let sessions = Arc::new(EncryptedSessionStore::indexeddb(

        StoreConfig::indexeddb(store_name.clone(), "macltok-web"),

    ));

    let login = PasswordLogin::new(homeserver, username, password).allow_insecure_loopback(true);

    let session = Session::login_password(login, sessions).await?;

    let (engine, handle, events) = session.build(EngineConfig::default()).await?;

    let (panicked, panics) = mpsc::unbounded();

    macltok::on_panic(move |message| {

        let _ = panicked.unbounded_send(message.to_owned());

    });

    let mut events = Watched { events, panics };

    let (done, running) = oneshot::channel();

    wasm_bindgen_futures::spawn_local(async move {

        let _ = done.send(engine.run().await);

    });

    let mut phase = Phase::new("the seeded room to sync");

    let (_never, mut idle) = oneshot::channel();

    let room_id = loop {

        let rooms = handle.rooms().await;

        if let Some(room) = rooms

            .into_iter()

            .find(|room| room.membership == RoomMembership::Joined && room.name == room_name)

        {

            break room.room_id;

        }

        while !matches!(

            phase.next(&mut events, &mut idle).await?,

            Event::RoomsChanged

        ) {}

    };

    let timeline_id = TimelineId::from(room_id.clone());

    let mut phase = Phase::new("the seeded message in the opened timeline");

    let mut opening = command(

        &handle,

        |handle| async move { handle.open_room(room_id).await },

    );

    let mut items = None;

    while !contains(&items, &seeded, false) {

        let event = phase.next(&mut events, &mut opening).await?;

        track(&timeline_id, event, &mut items)?;

    }

    let sent = format!("{seeded}, answered from {store_name}");

    let mut phase = Phase::new("the sent message to come back from the server");

    let content = MessageContent::text(sent.clone());

    let target = timeline_id.clone();

    let mut sending = command(&handle, |handle| async move {

        handle.send_message(target, content).await

    });

    while !contains(&items, &sent, true) {

        let event = phase.next(&mut events, &mut sending).await?;

        track(&timeline_id, event, &mut items)?;

    }

    handle.shutdown().await?;

    while events.events.recv().await.is_some() {}

    running

        .await

        .map_err(|_| JsError::new("engine task vanished"))??;

    Ok(())

}

/// answers only arrive while the event stream is drained, so commands run beside it

fn command<F>(

    handle: &ClientHandle,

    run: impl FnOnce(ClientHandle) -> F,

) -> oneshot::Receiver<Result<(), Error>>

where

    F: Future<Output = Result<(), Error>> + 'static,

{

    let (answer, answered) = oneshot::channel();

    let pending = run(handle.clone());

    wasm_bindgen_futures::spawn_local(async move {

        let _ = answer.send(pending.await);

    });

    answered

}

/// a panicking engine task goes silent rather than stopping, and a phase would otherwise wait

/// out its whole timeout

struct Watched {

    events: EventStream,

    panics: mpsc::UnboundedReceiver<String>,

}

struct Phase {

    what: &'static str,

    deadline: f64,

}

impl Phase {

    fn new(what: &'static str) -> Self {

        Self {

            what,

            deadline: js_sys::Date::now() + PHASE_TIMEOUT_MS,

        }

    }

    async fn next(

        &mut self,

        watched: &mut Watched,

        pending: &mut oneshot::Receiver<Result<(), Error>>,

    ) -> Result<Event, JsError> {

        let left = (self.deadline - js_sys::Date::now()).max(0.0);

        let timeout = TimeoutFuture::new(left as u32);

        let panicked = watched.panics.next();

        let outcome = match select(pin!(watched.events.recv()), select(timeout, panicked)).await {

            Either::Left((event, _)) => Some(event),

            Either::Right((Either::Left(_), _)) => None,

            Either::Right((Either::Right((message, _)), _)) => {

                return Err(JsError::new(&format!(

                    "panicked while waiting for {}, {}",

                    self.what,

                    message.unwrap_or_default()

                )));

            }

        };

        if let Ok(Some(Err(error))) = pending.try_recv() {

            return Err(JsError::new(&format!("{} failed, {error}", self.what)));

        }

        match outcome {

            Some(Some(Event::Stopped) | None) => Err(JsError::new(&format!(

                "engine stopped while waiting for {}",

                self.what

            ))),

            Some(Some(Event::SessionSaveFailed { message })) => Err(JsError::new(&format!(

                "saving the session failed while waiting for {}, {message}",

                self.what

            ))),

            Some(Some(event)) => Ok(event),

            None => Err(JsError::new(&format!(

                "timed out after {}s waiting for {}",

                PHASE_TIMEOUT_MS / 1000.0,

                self.what

            ))),

        }

    }

}

fn track(

    timeline_id: &TimelineId,

    event: Event,

    items: &mut Option<Vector<TimelineItem>>,

) -> Result<(), JsError> {

    match event {

        Event::TimelineReset {

            timeline_id: id,

            items: snapshot,

        } if &id == timeline_id => *items = Some(snapshot),

        Event::TimelineChanged {

            timeline_id: id,

            diffs,

        } if &id == timeline_id => {

            if let Some(items) = items {

                TimelineDiff::try_apply_all(diffs, items)

                    .map_err(|error| JsError::new(&format!("timeline diff misfit, {error:?}")))?;

            }

        }

        _ => {}

    }

    Ok(())

}

fn contains(items: &Option<Vector<TimelineItem>>, body: &str, remote_echo: bool) -> bool {

    items.iter().flatten().any(|item| {

        let TimelineItem::Event(event) = item else {

            return false;

        };

        let TimelineContent::Message(message) = &event.content else {

            return false;

        };

        message.body == body

            && (!remote_echo || (event.event_id.is_some() && event.send_state == SendState::Sent))

    })

}