macltok

mod commands;

pub(crate) mod executor;

mod media;

mod mutations;

mod room;

mod session;

mod timeline;

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;

use self::{

    commands::Job,

    executor::{EngineBoxFuture, TaskSet},

    mutations::{Mutation, MutationJob},

    room::{ActorNote, ActorRequest, RoomActor},

    timeline::{SessionContext, Supervisor},

};

use crate::{

    AccountCommand, ClientHandle, Command, ConnectionState, EncryptionCommand, EngineErrorKind,

    EngineFailure, Error, ErrorKind, Event, EventStream, LifecycleCommand, MediaCommand,

    NotificationCommand, RoomCommand, Termination, TimelineCommand,

    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,

}

impl EngineConfig {

    #[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

    }

}

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,

        }

    }

}

#[derive(Debug, ThisError)]

pub enum EngineError {

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

    Unauthenticated,

    #[error("command, event, and media source 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}"),

        }

    }

}

/// tasks catch their own panics, since the sdk's wasm join error drops the payload

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

    upload_reports: mpsc::Receiver<watchers::RoomUpload>,

}

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

    /// one progress forwarder per room that has uploaded, and whether it's subscribed yet

    uploads: HashMap<OwnedRoomId, watch::Receiver<bool>>,

    upload_reports: mpsc::Sender<watchers::RoomUpload>,

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

}

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

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

        {

            return Err(EngineError::ZeroCapacity);

        }

        let sync = Arc::new(

            SyncService::builder(client.clone())

                .with_offline_mode()

                .build()

                .await

                .map_err(EngineError::SyncService)?,

        );

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

            client.clone(),

            config.max_media_bytes,

            config.max_media_sources,

        ));

        let (handle, port, events) = transport_pair(

            client.clone(),

            sync.clone(),

            media.clone(),

            config.command_capacity,

            config.event_capacity,

        );

        client.send_queue().enable_upload_progress(true);

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

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

        let (upload_reports, upload_report_queue) = mpsc::channel(config.event_capacity);

        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,

                upload_reports: upload_report_queue,

            }),

            upload_reports,

            uploads: HashMap::new(),

            verification_handler: None,

            stop,

            stops,

            logouts: Arc::new(watch::Sender::new(Logouts::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.resume_upload_forwarders().await;

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

                }

                Some(note) = self.notes.recv() => self.actor_note(note).await,

                request = self.port.recv(), if !handles_dropped => {

                    // nobody's left to ask, but a logout they asked for still deserves to land

                    let Some(request) = request else {

                        handles_dropped = true;

                        continue;

                    };

                    if let Some(reply) = self.dispatch(request).await {

                        shutdown_reply = Some(reply);

                        break Ok(Termination::ShutDown);

                    }

                }

            }

        };

        if let Err(error) = &ending {

            let _ = events.send(Event::Failed(error.failure())).await;

        }

        self.shutdown().await;

        let _ = events.send(Event::Stopped).await;

        if let Some(reply) = shutdown_reply {

            reply.done(Ok(()));

        }

        // everything else still queued answers `Interrupted` as it drops, but a shutdown did happen

        while let Some(Request { command, reply }) = self.port.try_recv() {

            if matches!(command, Command::Lifecycle(LifecycleCommand::Shutdown)) {

                reply.done(Ok(()));

            }

        }

        ending

    }

    fn start_tasks(&mut self) {

        let queues = self.queues.take().expect("engine tasks start once");

        let (incoming, incoming_verifications) = mpsc::unbounded_channel();

        self.verification_handler = Some(self.verification_handler(incoming));

        let client = &self.client;

        let events = || self.port.event_sender();

        let (notices, notice_queue) = mpsc::unbounded_channel();

        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 12] = [

            (

                "upload tracker",

                Box::pin(watchers::uploads(

                    client.clone(),

                    client.send_queue().subscribe(),

                    queues.upload_reports,

                    events(),

                )),

            ),

            (

                "room subscription watcher",

                Box::pin(watchers::room_subscriptions(

                    self.sync.clone(),

                    self.subscriptions.subscribe(),

                )),

            ),

            (

                "session change watcher",

                Box::pin(session::session(

                    client.clone(),

                    queues.session_changes,

                    queues.persistence,

                    notices,

                    client

                        .device_id()

                        .expect("authentication was checked during construction")

                        .to_owned(),

                    self.logouts.clone(),

                )),

            ),

            (

                "session notice watcher",

                Box::pin(session::session_notices(

                    notice_queue,

                    events(),

                    self.stop.clone(),

                )),

            ),

            (

                "connection state watcher",

                Box::pin(watchers::connection(

                    self.sync.state(),

                    self.sync.clone(),

                    client.clone(),

                    events(),

                    self.port.connection_sender(),

                )),

            ),

            (

                "send queue watcher",

                Box::pin(watchers::send_queue_errors(

                    client.clone(),

                    client.send_queue().subscribe_errors(),

                    self.port.connection_sender().subscribe(),