macltok

mod commands;

pub(crate) mod context;

pub(crate) mod executor;

mod media;

mod mutations;

mod packs;

mod prepare;

mod repair;

mod room;

mod session;

mod timeline;

mod url_preview;

mod watchers;

use std::{

    collections::{HashMap, VecDeque},

    future::Future,

    panic::AssertUnwindSafe,

    sync::Arc,

};

use futures_util::FutureExt;

use matrix_sdk::{Client, SendOutsideWasm, SessionChange};

use matrix_sdk_ui::sync_service::SyncService;

use ruma::{OwnedDeviceId, OwnedRoomId};

use thiserror::Error as ThisError;

use tokio::sync::{broadcast, mpsc, watch};

pub(crate) use media::MediaService;

pub(crate) use session::Persistence;

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]

pub(crate) use crate::store_lock::StoreLock;

#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]

pub(crate) enum StoreLock {}

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]

pub(crate) use crate::mentions::store::MentionsStore;

#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]

pub(crate) enum MentionsStore {}

use self::{

    commands::Job,

    executor::{EngineBoxFuture, TaskSet},

    mutations::{Mutation, MutationJob},

    packs::PackEdit,

    room::{ActorNote, ActorRequest, RoomActor},

    timeline::{SessionContext, Supervisor},

};

use crate::{

    AccountCommand, ClientHandle, Command, CommandOutput, ConnectionState, EncryptionCommand,

    EngineErrorKind, EngineFailure, Error, ErrorKind, Event, EventStream, ImagePackSource,

    LifecycleCommand, MediaCommand, NotificationCommand, PackCommand, RoomCommand, Termination,

    TimelineCommand,

    normalise::{ROOM_PACKS, ROOM_PREVIEW_URLS},

    transport::{DriverPort, Request, Responder, transport_pair},

};

#[derive(Clone, Debug, Eq, PartialEq)]

#[non_exhaustive]

pub struct EngineConfig {

    pub command_capacity: usize,

    pub event_capacity: usize,

    pub initial_history_limit: u16,

    pub max_media_bytes: usize,

    pub max_media_sources: usize,

    pub max_mentions_bytes: usize,

    pub max_mentions_entries: usize,

    pub max_mention_context_events: usize,

    pub max_mention_candidates: usize,

    /// warm a room up for sending when it's opened or someone starts typing in it

    pub prepare_to_send: bool,

}

impl EngineConfig {

    #[must_use]

    pub const fn with_max_mentions_bytes(mut self, bytes: usize) -> Self {

        self.max_mentions_bytes = bytes;

        self

    }

    #[must_use]

    pub const fn with_max_mentions_entries(mut self, entries: usize) -> Self {

        self.max_mentions_entries = entries;

        self

    }

    #[must_use]

    pub const fn with_max_mention_context_events(mut self, events: usize) -> Self {

        self.max_mention_context_events = events;

        self

    }

    #[must_use]

    pub const fn with_max_mention_candidates(mut self, candidates: usize) -> Self {

        self.max_mention_candidates = candidates;

        self

    }

    #[must_use]

    pub const fn with_command_capacity(mut self, capacity: usize) -> Self {

        self.command_capacity = capacity;

        self

    }

    #[must_use]

    pub const fn with_event_capacity(mut self, capacity: usize) -> Self {

        self.event_capacity = capacity;

        self

    }

    #[must_use]

    pub const fn with_initial_history_limit(mut self, limit: u16) -> Self {

        self.initial_history_limit = limit;

        self

    }

    #[must_use]

    pub const fn with_max_media_bytes(mut self, bytes: usize) -> Self {

        self.max_media_bytes = bytes;

        self

    }

    #[must_use]

    pub const fn with_max_media_sources(mut self, sources: usize) -> Self {

        self.max_media_sources = sources;

        self

    }

    #[must_use]

    pub const fn with_prepare_to_send(mut self, prepare: bool) -> Self {

        self.prepare_to_send = prepare;

        self

    }

}

impl Default for EngineConfig {

    fn default() -> Self {

        Self {

            command_capacity: 64,

            event_capacity: 128,

            initial_history_limit: 30,

            max_media_bytes: 20 * 1024 * 1024,

            max_media_sources: 2_048,

            max_mentions_bytes: 64 * 1024 * 1024,

            max_mentions_entries: 5_000,

            max_mention_context_events: 25_000,

            max_mention_candidates: 5_000,

            prepare_to_send: true,

        }

    }

}

#[derive(Debug, ThisError)]

pub enum EngineError {

    #[error("the Matrix client is not authenticated")]

    Unauthenticated,

    #[error("command, event, media source, and mentions capacities must be nonzero")]

    ZeroCapacity,

    #[error("could not build the sync service")]

    SyncService(#[source] matrix_sdk_ui::sync_service::Error),

    #[error("could not open the mentions service")]

    MentionsStore(#[source] crate::Error),

    #[error("the event receiver has closed")]

    EventReceiverClosed,

    #[error("engine task {task} panicked")]

    TaskPanicked { task: String },

}

impl EngineError {

    fn failure(&self) -> EngineFailure {

        EngineFailure {

            kind: match self {

                Self::EventReceiverClosed => EngineErrorKind::EventReceiverClosed,

                Self::TaskPanicked { .. } => EngineErrorKind::Watcher,

                Self::Unauthenticated

                | Self::ZeroCapacity

                | Self::SyncService(_)

                | Self::MentionsStore(_) => EngineErrorKind::Other,

            },

            message: self.to_string(),

        }

    }

}

#[derive(Clone)]

enum TaskContext {

    /// a panic here costs the caller an `Interrupted` answer and nothing else

    Command(&'static str),

    /// a panic here takes the engine with it

    Watcher(&'static str),

    RoomActor {

        room_id: OwnedRoomId,

        generation: u64,

    },

}

impl TaskContext {

    fn label(&self) -> String {

        match self {

            Self::Command(name) | Self::Watcher(name) => (*name).to_owned(),

            Self::RoomActor {

                room_id,

                generation,

            } => format!("room actor {room_id} generation {generation}"),

        }

    }

}

/// tasks catch their own panics so each comes back with its context, since joining

/// through the sdk's executor gives no task id to look one up by. wasm has nothing to catch

fn panic_message(payload: Box<dyn std::any::Any + Send>) -> String {

    payload

        .downcast_ref::<&str>()

        .map(ToString::to_string)

        .or_else(|| payload.downcast_ref::<String>().cloned())

        .unwrap_or_else(|| "task panicked with a non-string payload".to_owned())

}

struct TaskRegistry {

    tasks: TaskSet<(TaskContext, Option<String>)>,

}

impl TaskRegistry {

    fn spawn(

        &mut self,

        context: TaskContext,

        task: impl Future<Output = ()> + SendOutsideWasm + 'static,

    ) {

        self.tasks.spawn(async move {

            let panic = AssertUnwindSafe(task)

                .catch_unwind()

                .await

                .err()

                .map(panic_message);

            (context, panic)

        });

    }

    fn is_empty(&self) -> bool {

        self.tasks.is_empty()

    }

    async fn join_next(&mut self) -> Option<(TaskContext, Option<String>)> {

        loop {

            if let Ok(done) = self.tasks.join_next().await? {

                return Some(done);

            }

        }

    }

    async fn shutdown(&mut self) {

        self.tasks.shutdown().await;

    }

}

enum RoomSlot {

    Active {

        generation: u64,

        sender: mpsc::Sender<ActorRequest>,

    },

    /// opens and closes that arrive while the actor winds down, replayed in order once it's gone

    Closing {

        generation: u64,

        pending: VecDeque<Deferred>,

    },

}

impl RoomSlot {

    const fn generation(&self) -> u64 {

        match self {

            Self::Active { generation, .. } | Self::Closing { generation, .. } => *generation,

        }

    }

}

enum Deferred {

    Open(Responder),

    Close(Responder),

}

/// logouts in flight, and a token rejection held back because one of them caused it

#[derive(Default)]

pub(super) struct Logouts {

    pub(super) in_flight: usize,

    pub(super) rejected: Option<Option<OwnedDeviceId>>,

}

struct Queues {

    mutations: mpsc::Receiver<MutationJob>,

    notifications: mpsc::Receiver<Job<NotificationCommand>>,

    encryption: mpsc::Receiver<Job<EncryptionCommand>>,

    session_changes: broadcast::Receiver<SessionChange>,

    persistence: Option<Persistence>,

}

/// requests other devices send us, handed from the sdk's event handler to the encryption worker

type IncomingVerifications =

    mpsc::UnboundedSender<matrix_sdk::encryption::verification::VerificationRequest>;

pub struct Engine {

    client: Client,

    sync: Arc<SyncService>,

    port: DriverPort,

    config: EngineConfig,

    media: Arc<MediaService>,

    tasks: TaskRegistry,

    rooms: HashMap<OwnedRoomId, RoomSlot>,

    next_generation: u64,

    notes: mpsc::UnboundedReceiver<ActorNote>,

    note_sender: mpsc::UnboundedSender<ActorNote>,

    /// the rooms sliding sync should follow, applied off the run loop since it touches the store

    subscriptions: watch::Sender<Vec<OwnedRoomId>>,

    mutations: mpsc::Sender<MutationJob>,

    notifications: mpsc::Sender<Job<NotificationCommand>>,

    encryption: mpsc::Sender<Job<EncryptionCommand>>,

    queues: Option<Queues>,

    /// the client can outlive us, so the handler has to come off it when we stop

    verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,

    stop: mpsc::UnboundedSender<Termination>,

    stops: mpsc::UnboundedReceiver<Termination>,

    logouts: Arc<watch::Sender<Logouts>>,

    preparations: prepare::Preparations,

    /// shared with every `ClientHandle`, so the store stays locked while the client is reachable

    _store_lock: Option<Arc<StoreLock>>,

    mentions: crate::mentions::service::Handle,

    mentions_service: Option<crate::mentions::service::Service>,

    mentions_updates: Option<watch::Receiver<crate::MentionsPage>>,

    repair_gate: Arc<tokio::sync::Mutex<()>>,

    repair_shutdown: Arc<std::sync::atomic::AtomicBool>,

}

impl Engine {

    /// the escape hatch for a client you built yourself. `Session` is the usual way in

    pub async fn from_client(

        client: Client,

        config: EngineConfig,

    ) -> Result<(Self, ClientHandle, EventStream), EngineError> {

        let session_changes = client.subscribe_to_session_changes();

        Self::with_persistence(client, config, session_changes, None, None, None).await

    }

    /// `session_changes` is taken when the client is built, since the sdk can refresh

    /// before an engine exists

    pub(crate) async fn with_persistence(

        client: Client,

        config: EngineConfig,

        session_changes: broadcast::Receiver<SessionChange>,

        persistence: Option<Persistence>,

        store_lock: Option<StoreLock>,

        mentions: Option<MentionsStore>,

    ) -> Result<(Self, ClientHandle, EventStream), EngineError> {

        if client.user_id().is_none() {

            return Err(EngineError::Unauthenticated);

        }

        if config.command_capacity == 0

            || config.event_capacity == 0

            || config.max_media_sources == 0

            || config.max_mentions_entries == 0

            || config.max_mention_context_events == 0

            || config.max_mention_candidates == 0

        {

            return Err(EngineError::ZeroCapacity);

        }

        let store_lock = store_lock.map(Arc::new);

        let media = Arc::new(MediaService::new(

            client.clone(),

            config.max_media_bytes,

            config.max_media_sources,

        ));

        let (mentions, mentions_service, mentions_updates) =

            crate::mentions::service::Service::new(

                client.clone(),

                media.clone(),

                mentions,

                config.clone(),

            )

            .await

            .map_err(EngineError::MentionsStore)?;

        let sync = Arc::new(

            SyncService::builder(client.clone())

                .with_offline_mode()

                .with_room_list_extra_required_state(vec![

                    (ROOM_PACKS.into(), "*".to_owned()),

                    (ROOM_PREVIEW_URLS.into(), String::new()),

                ])

                .with_room_list_response_commit_hook(mentions.hook())

                .build()

                .await

                .map_err(EngineError::SyncService)?,

        );

        let (handle, port, events) = transport_pair(

            client.clone(),

            sync.clone(),

            media.clone(),

            store_lock.clone(),

            config.command_capacity,

            config.event_capacity,

        );

        client.send_queue().enable_upload_progress(true);

        client.send_queue().enable_global_upload_progress(true);

        let (note_sender, notes) = mpsc::unbounded_channel();

        let (stop, stops) = mpsc::unbounded_channel();

        let (mutations, mutation_queue) = mpsc::channel(config.command_capacity);

        let (notifications, notification_queue) = mpsc::channel(config.command_capacity);

        let (encryption, encryption_queue) = mpsc::channel(config.command_capacity);

        let engine = Self {

            client,

            sync,

            port,

            config,

            media,

            tasks: TaskRegistry {

                tasks: TaskSet::new(),

            },

            rooms: HashMap::new(),

            next_generation: 0,

            notes,

            note_sender,

            subscriptions: watch::Sender::new(Vec::new()),

            mutations,

            notifications,

            encryption,

            queues: Some(Queues {

                mutations: mutation_queue,

                notifications: notification_queue,

                encryption: encryption_queue,

                session_changes,

                persistence,

            }),

            verification_handler: None,

            preparations: prepare::Preparations::default(),

            stop,

            stops,

            logouts: Arc::new(watch::Sender::new(Logouts::default())),

            _store_lock: store_lock,

            mentions,

            mentions_service: Some(mentions_service),

            mentions_updates: Some(mentions_updates),

            repair_gate: Arc::default(),

            repair_shutdown: Arc::default(),

        };

        Ok((engine, handle, events))

    }

    /// boxed because the engine's future is deep enough to blow the compiler's query depth

    /// in every crate that spawns it, which isn't something a client should have to fix

    pub fn run(self) -> EngineBoxFuture<'static, Result<Termination, EngineError>> {

        Box::pin(self.run_inner())

    }

    async fn run_inner(mut self) -> Result<Termination, EngineError> {

        let events = self.port.event_sender();

        let user_id = self

            .client

            .user_id()

            .expect("authentication was checked during construction")

            .to_owned();

        self.port

            .connection_sender()

            .send_replace(ConnectionState::Connecting);

        for event in [

            Event::Authenticated { user_id },

            Event::ConnectionChanged(ConnectionState::Connecting),

        ] {

            if events.send(event).await.is_err() {

                self.shutdown().await;

                return Err(EngineError::EventReceiverClosed);

            }

        }

        self.start_tasks();

        self.sync.start().await;

        let mut shutdown_reply = None;

        let mut handles_dropped = false;

        let mut logouts = self.logouts.subscribe();

        let ending = loop {

            tokio::select! {

                biased;

                Some(termination) = self.stops.recv() => break Ok(termination),

                () = events.closed() => break Err(EngineError::EventReceiverClosed),

                () = settled(&mut logouts), if handles_dropped => {

                    break Ok(Termination::HandlesDropped);

                }

                Some((context, panic)) = self.tasks.join_next(), if !self.tasks.is_empty() => {

                    let Some(message) = panic else { continue };

                    if let TaskContext::Command(label) = context {

                        tracing::error!(label, panic = %message, "command task panicked");

                        continue;

                    }

                    break Err(EngineError::TaskPanicked { task: context.label() });