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 thread_list;

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 {}

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

pub(crate) use crate::search::store::SearchStore;

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

pub(crate) enum SearchStore {}

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,

    ThreadListCommand, 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,

    /// what a thread paginates as it opens, and a room if its recent load fails. 0 for neither

    pub initial_history_limit: u16,

    pub max_media_bytes: usize,

    /// media one event context session keeps registered, least recently used dropped first

    pub max_context_media: 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,

    /// `/messages` pages each joined room gets per engine run for search backfill. 0 for none

    pub search_backfill_pages_per_room: u16,

    pub search_backfill_page_size: u16,

    /// the least time between two backfill requests, across every room

    pub search_backfill_interval_ms: u64,

    /// undecryptable messages kept to index once their keys turn up, newest first

    pub max_search_undecrypted: usize,

}

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_context_media(mut self, media: usize) -> Self {

        self.max_context_media = media;

        self

    }

    #[must_use]

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

        self.prepare_to_send = prepare;

        self

    }

    #[must_use]

    pub const fn with_search_backfill_pages_per_room(mut self, pages: u16) -> Self {

        self.search_backfill_pages_per_room = pages;

        self

    }

    #[must_use]

    pub const fn with_search_backfill_page_size(mut self, events: u16) -> Self {

        self.search_backfill_page_size = events;

        self

    }

    #[must_use]

    pub const fn with_search_backfill_interval_ms(mut self, interval_ms: u64) -> Self {

        self.search_backfill_interval_ms = interval_ms;

        self

    }

    #[must_use]

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

        self.max_search_undecrypted = events;

        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_context_media: 1_024,

            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,

            search_backfill_pages_per_room: 10,

            search_backfill_page_size: 100,

            search_backfill_interval_ms: 2_000,

            max_search_undecrypted: 2_000,

        }

    }

}

#[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("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(_) => {

                    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}"),

        }

    }

}

/// tuwunel resends every room in range each time the room list grows, so the sdk's 100 a step

/// had a 399 room account hand most rooms to the event cache four or five times

const ROOM_LIST_GROWING_BATCH: u32 = 500;

/// how long shutdown waits on the mentions and search services before aborting them with the

/// other tasks. a sync response still being processed kept mentions busy for 2 to 3.4s, and the

/// next engine can't start until this one is gone

const SERVICE_STOP_GRACE: std::time::Duration = std::time::Duration::from_millis(500);

async fn stop_within(stop: impl Future<Output = ()>, service: &'static str) {

    if matrix_sdk::timeout::timeout(stop, SERVICE_STOP_GRACE)

        .await

        .is_err()

    {

        tracing::debug!(service, "a service didn't stop in time and is aborted");

    }

}

/// 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>>,

    thread_lists: mpsc::Receiver<Job<ThreadListCommand>>,

    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>,

    contexts: Arc<context::Sessions>,

    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>>,

    thread_lists: mpsc::Sender<Job<ThreadListCommand>>,

    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>>,

    search: crate::search::Handle,

    search_service: Option<crate::search::Service>,

}

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, 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<Result<MentionsStore, crate::mentions::service::Unopened>>,

        search: Option<Result<SearchStore, crate::Error>>,

    ) -> 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_context_media == 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_context_media,

        ));

        let (mentions, mentions_service, mentions_updates) =

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

                client.clone(),

                media.clone(),

                mentions,

                config.clone(),

            )

            .await

            // a missing user id is all it still refuses, and that was turned away above

            .map_err(|_| EngineError::Unauthenticated)?;

        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())

                .with_room_list_growing_batch_size(ROOM_LIST_GROWING_BATCH)

                .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,

        );

        let (search, search_service) = crate::search::Service::new(

            client.clone(),

            media.clone(),

            search,

            port.event_sender(),

            &config,

        );

        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 (thread_lists, thread_list_queue) = mpsc::channel(config.command_capacity);

        let engine = Self {

            contexts: Arc::new(context::Sessions::new(port.event_sender())),

            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()),