matrix-rust-sdk

#![allow(clippy::large_enum_variant)]

use std::{

    collections::{HashMap, HashSet},

    io::{self, Write, stdout},

    path::{Path, PathBuf},

    sync::Arc,

    time::{Duration, Instant},

};

use clap::Parser;

use color_eyre::Result;

use crossterm::{

    event::{

        self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyEvent, KeyModifiers,

    },

    execute,

};

use futures_util::{StreamExt as _, pin_mut};

use imbl::Vector;

use layout::Flex;

use matrix_sdk::{

    AuthSession, Client, SqliteCryptoStore, SqliteEventCacheStore, SqliteStateStore,

    ThreadingSupport,

    authentication::matrix::MatrixSession,

    config::StoreConfig,

    encryption::{BackupDownloadStrategy, EncryptionSettings},

    reqwest::Url,

    ruma::{OwnedRoomId, api::client::room::create_room::v3::Request as CreateRoomRequest},

    search_index::{SearchIndexGuard, SearchIndexStoreKind},

};

use matrix_sdk_base::{RoomStateFilter, event_cache::store::EventCacheStoreLockGuard};

use matrix_sdk_common::{cross_process_lock::CrossProcessLockConfig, locks::Mutex};

use matrix_sdk_ui::{

    Timeline as SdkTimeline,

    room_list_service::{self, State, filters::new_filter_non_left},

    sync_service::SyncService,

    timeline::{RoomExt as _, TimelineFocus, TimelineItem},

};

use ratatui::{DefaultTerminal, prelude::*, style::palette::tailwind, widgets::*};

use throbber_widgets_tui::{Throbber, ThrobberState};

use tokio::{

    spawn,

    sync::mpsc::{Receiver, Sender, channel, error::TryRecvError},

    task::JoinHandle,

    time::timeout,

};

use tracing::{debug, error, warn};

use tracing_subscriber::EnvFilter;

use widgets::{

    recovery::create_centered_throbber_area, room_view::RoomView, settings::SettingsView,

};

use crate::widgets::{

    create_room::CreateRoomView,

    help::HelpView,

    room_list::{ExtraRoomInfo, RoomInfos, RoomList, Rooms},

    search::{

        indexing::{IndexingMessage, IndexingView},

        searching::SearchingView,

    },

    status::Status,

};

mod widgets;

const HEADER_BG: Color = tailwind::BLUE.c950;

const NORMAL_ROW_COLOR: Color = tailwind::SLATE.c950;

const ALT_ROW_COLOR: Color = tailwind::SLATE.c900;

const SELECTED_STYLE_FG: Color = tailwind::BLUE.c300;

const TEXT_COLOR: Color = tailwind::SLATE.c200;

type Timelines = Arc<Mutex<HashMap<OwnedRoomId, Timeline>>>;

#[derive(Debug, Parser)]

struct Cli {

    /// The homeserver the client should connect to.

    server_name: String,

    /// The path where session specific data should be stored.

    #[clap(default_value = "/tmp/")]

    session_path: PathBuf,

    /// Set the proxy that should be used for the connection.

    #[clap(short, long, env = "PROXY")]

    proxy: Option<Url>,

    /// Whether to *not* reload the `pos`ition sliding sync token from disk at

    /// start or not, for the room list sliding sync.

    ///

    /// Set to false by default (i.e. reload the position from disk).

    #[clap(short, long, default_value_t = false)]

    dont_share_pos: bool,

}

#[derive(Default)]

pub enum GlobalMode {

    /// The default mode, no popout screen is opened.

    #[default]

    Default,

    /// Mode where we have opened the help screen.

    Help,

    /// Mode where we have opened the settings screen.

    Settings { view: SettingsView },

    /// Mode where we are shutting our tasks down and exiting multiverse.

    Exiting { shutdown_task: JoinHandle<()> },

    /// Mode where we have opened the create room screen

    CreateRoom { view: CreateRoomView },

    /// Mode where we have opened the search screen

    Searching { view: SearchingView, is_global: bool },

    /// Mode where we have opened the indexing screen

    Indexing { view: IndexingView },

}

/// Helper function to create a centered rect using up certain percentage of the

/// available rect `r`

fn popup_area(area: Rect, percent_x: u16, percent_y: u16) -> Rect {

    let vertical = Layout::vertical([Constraint::Percentage(percent_y)]).flex(Flex::Center);

    let horizontal = Layout::horizontal([Constraint::Percentage(percent_x)]).flex(Flex::Center);

    let [area] = vertical.areas(area);

    let [area] = horizontal.areas(area);

    area

}

#[tokio::main]

async fn main() -> Result<()> {

    let cli = Cli::parse();

    let file_writer = tracing_appender::rolling::hourly(&cli.session_path, "logs-");

    tracing_subscriber::fmt()

        .with_env_filter(EnvFilter::from_default_env())

        .with_ansi(false)

        .with_writer(file_writer)

        .init();

    color_eyre::install()?;

    let share_pos = !cli.dont_share_pos;

    let client = configure_client(cli).await?;

    // Watch for errors in background jobs.

    spawn({

        let client = client.clone();

        async move {

            let mut recv = client.task_monitor().subscribe();

            while let Ok(report) = recv.recv().await {

                error!(name = report.task.name, ?report.reason, "A background task has crashed!");

            }

        }

    });

    let event_cache = client.event_cache();

    event_cache.config_mut().experimental_auto_backpagination = true;

    event_cache.subscribe()?;

    let terminal = ratatui::init();

    execute!(stdout(), EnableMouseCapture)?;

    let mut app = App::new(client, share_pos).await?;

    app.run(terminal).await

}

pub struct Timeline {

    timeline: Arc<SdkTimeline>,

    items: Arc<Mutex<Vector<Arc<TimelineItem>>>>,

    task: JoinHandle<()>,

}

#[derive(Default)]

pub struct AppState {

    /// What popup are we showing that is covering the majority of the screen,

    /// mainly used for help and settings screens.

    global_mode: GlobalMode,

    /// State for a global throbber.

    throbber_state: ThrobberState,

}

struct App {

    /// Reference to the main SDK client.

    client: Client,

    /// The sync service used for synchronizing events.

    sync_service: Arc<SyncService>,

    /// Timelines data structures for each room.

    timelines: Timelines,

    /// The room list widget on the left-hand side of the screen.

    room_list: RoomList,

    /// A view displaying the contents of the selected room, the widget on the

    /// right-hand side of the screen.

    room_view: RoomView,

    /// Task listening to room list service changes, and spawning timelines.

    listen_task: JoinHandle<()>,

    /// Task that is indexing events for search.

    indexing_task: JoinHandle<()>,

    /// Receiver that notifies when indexing is complete.

    indexing_receiver: Receiver<(bool, IndexingMessage)>,

    /// The status widget at the bottom of the screen.

    status: Status,

    state: AppState,

    last_tick: Instant,

}

impl App {

    const TICK_RATE: Duration = Duration::from_millis(250);

    async fn new(client: Client, share_pos: bool) -> Result<Self> {

        let sync_service =

            Arc::new(SyncService::builder(client.clone()).with_share_pos(share_pos).build().await?);

        let rooms = Rooms::default();

        let room_infos = RoomInfos::default();

        let timelines = Timelines::default();

        let room_list_service = sync_service.room_list_service();

        let all_rooms = room_list_service.all_rooms().await?;

        let listen_task = spawn(Self::listen_task(

            rooms.clone(),

            room_infos.clone(),

            timelines.clone(),

            all_rooms,

        ));

        // This will sync (with encryption) until an error happens or the program is

        // stopped.

        sync_service.start().await;

        let status = Status::new();

        let room_list =

            RoomList::new(client.clone(), rooms, room_infos, sync_service.clone(), status.handle());

        let room_view = RoomView::new(client.clone(), timelines.clone(), status.handle());

        let (indexing_sender, indexing_receiver) = channel::<(bool, IndexingMessage)>(1024);

        let indexing_task =

            spawn(App::indexing_task(client.clone(), indexing_sender, sync_service.clone()));

        let indexing_view = IndexingView::new();

        Ok(Self {

            sync_service,

            timelines,

            room_list,

            room_view,

            client,

            listen_task,

            indexing_task,

            indexing_receiver,

            status,

            state: AppState {

                global_mode: GlobalMode::Indexing { view: indexing_view },

                ..Default::default()

            },

            last_tick: Instant::now(),

        })

    }

    async fn listen_task(

        rooms: Rooms,

        room_infos: RoomInfos,

        timelines: Timelines,

        all_rooms: room_list_service::RoomList,

    ) {

        let (stream, entries_controller) = all_rooms.entries_with_dynamic_adapters(50_000);

        entries_controller.set_filter(Box::new(new_filter_non_left()));

        pin_mut!(stream);

        let mut previous_rooms = HashSet::new();

        while let Some(diffs) = stream.next().await {

            let all_rooms = {

                // Apply the diffs to the list of room entries.

                let mut rooms = rooms.lock();

                for diff in diffs {

                    diff.apply(&mut rooms);

                }

                // Collect rooms early to release the room entries list lock.

                (*rooms).clone()

            };

            let mut new_rooms = HashMap::new();

            let mut new_timelines = Vec::new();

            // Update all the room info for all rooms.

            for room in all_rooms.iter() {

                let raw_name = room.name();

                let display_name =

                    room.cached_display_name().map(|display_name| display_name.to_string());

                let is_dm = room

                    .is_direct()

                    .await

                    .map_err(|err| {

                        warn!("couldn't figure whether a room is a DM or not: {err}");

                    })

                    .ok();

                room_infos.lock().insert(

                    room.room_id().to_owned(),

                    ExtraRoomInfo { raw_name, display_name, is_dm },

                );

            }

            // Initialize all the new rooms.

            for room in

                all_rooms.into_iter().filter(|room| !previous_rooms.contains(room.room_id()))

            {

                // Initialize the timeline.

                let Ok(timeline) = room

                    .timeline_builder()

                    .with_focus(TimelineFocus::Live { hide_threaded_events: true })

                    .build()

                    .await

                else {

                    error!("error when creating default timeline");

                    continue;

                };

                // Save the timeline in the cache.

                let (items, stream) = timeline.subscribe().await;

                let items = Arc::new(Mutex::new(items));

                // Spawn a timeline task that will listen to all the timeline item changes.

                let i = items.clone();

                let timeline_task = spawn(async move {

                    pin_mut!(stream);

                    let items = i;

                    while let Some(diffs) = stream.next().await {

                        let mut items = items.lock();

                        for diff in diffs {

                            diff.apply(&mut items);

                        }

                    }

                });

                new_timelines.push((

                    room.room_id().to_owned(),

                    Timeline { timeline: Arc::new(timeline), items, task: timeline_task },

                ));

                // Save the room list service room in the cache.

                new_rooms.insert(room.room_id().to_owned(), room);

            }

            previous_rooms.extend(new_rooms.into_keys());

            timelines.lock().extend(new_timelines);

        }

    }

    async fn wait_for_room_sync(

        update_sender: &Sender<(bool, IndexingMessage)>,

        sync_service: Arc<SyncService>,

    ) {

        let mut sync_subscriber = sync_service.room_list_service().state();

        // Spin until there are rooms to index

        while let Some(state) = sync_subscriber.next().await {

            match state {

                State::Running => return,

                State::Terminated { from: _prev } => {

                    while let Err(e) =

                        update_sender.send((true, IndexingMessage::Progress(0))).await

                    {

                        debug!("Failed to send final message, trying again: {e:?}");

                    }

                    return;

                }

                _ => {

                    debug!("Sync service not running. Waiting to start indexing. {state:?}");

                }

            }

        }

    }

    async fn index_event_cache(

        client: &Client,

        update_sender: &Sender<(bool, IndexingMessage)>,

        store: &EventCacheStoreLockGuard,

        search_index_guard: &mut SearchIndexGuard<'_>,

        mut count: usize,

    ) -> Result<usize, ()> {

        for room in client.rooms_filtered(RoomStateFilter::JOINED.union(RoomStateFilter::LEFT)) {

            let room_id = room.room_id();

            let maybe_room_cache = room.event_cache().await;

            let Ok((room_cache, _drop_handles)) = maybe_room_cache else {

                warn!("Failed to get RoomEventCache: {maybe_room_cache:?}");

                continue;

            };

            let redaction_rules = room.clone_info().room_version_rules_or_default().redaction;

            let maybe_timeline_events = store.get_room_events(room_id, None, None).await;

            let Ok(timeline_events) = maybe_timeline_events else {

                warn!("Failed to get room's events: {maybe_timeline_events:?}");

                continue;

            };

            let no_of_events = timeline_events.len();

            if let Err(err) = search_index_guard

                .bulk_handle_timeline_event(

                    timeline_events.clone().into_iter(),

                    &room_cache,

                    room_id,

                    &redaction_rules,

                )

                .await

            {

                error!("Failed to handle event for indexing: {err}");

                let mut error = Some(err);

                while let Some(err) = error.take() {

                    if let Err(e) =

                        update_sender.send((true, IndexingMessage::Error(err.to_string()))).await

                    {

                        debug!("Failed to send final error message, trying again: {e:?}");

                    }

                }

                return Err(());

            }

            count += no_of_events;

            let _ = update_sender.send((false, IndexingMessage::Progress(count))).await;

        }

        Ok(count)

    }

    async fn index_from_server(

        client: &Client,

        update_sender: &Sender<(bool, IndexingMessage)>,

        search_index_guard: &mut SearchIndexGuard<'_>,

        mut count: usize,

    ) -> Result<usize, ()> {

        let batch_size = 25;

        let mut rooms = client.rooms_filtered(RoomStateFilter::JOINED);

        let mut idx = 0;

        while !rooms.is_empty() {

            let room = &rooms[idx];

            let room_id = room.room_id();

            let maybe_room_cache = room.event_cache().await;

            let Ok((room_cache, _drop_handles)) = maybe_room_cache else {

                warn!("Failed to get RoomEventCache: {maybe_room_cache:?}");

                idx = (idx + 1) % rooms.len();

                continue;

            };

            let redaction_rules = room.clone_info().room_version_rules_or_default().redaction;

            let Ok(pagination) = room_cache.pagination().run_backwards_until(batch_size).await

            else {

                error!("Failed to backpaginate {room_id}");

                idx = (idx + 1) % rooms.len();

                continue;

            };

            let no_of_events = pagination.events.len();

            if let Err(err) = search_index_guard

                .bulk_handle_timeline_event(

                    pagination.events.clone().into_iter(),

                    &room_cache,

                    room_id,

                    &redaction_rules,

                )

                .await

            {

                warn!("Failed to handle event for indexing: {err}");

                let mut error = Some(err);

                while let Some(err) = error.take() {

                    if let Err(e) =

                        update_sender.send((true, IndexingMessage::Error(err.to_string()))).await

                    {

                        debug!("Failed to send final error message, trying again: {e:?}");

                    }

                }

                return Err(());

            }

            count += no_of_events;

            let _ = update_sender.send((false, IndexingMessage::Progress(count))).await;

            if pagination.reached_start {

                rooms.remove(idx);

                let len = rooms.len();

                if len > 0 {

                    idx %= len;

                }

            } else {

                idx = (idx + 1) % rooms.len();

            }

        }