macltok

use std::collections::{HashMap, VecDeque};

use matrix_sdk::{Room, RoomState};

use matrix_sdk_ui::timeline::{RoomExt, Timeline, TimelineFocus, TimelineReadReceiptTracking};

use ruma::OwnedRoomId;

use tokio::sync::mpsc;

use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, label, resume_unwind};

use super::repair::{Completion, Outcome, RepairController};

use super::timeline::{SessionContext, TaskGuard, TimelineSession};

use crate::{

    Error, ErrorKind, Event, HistoryRepairCommand, RecentHistoryState, TimelineCommand, TimelineId,

    TypingState, transport::Responder,

};

/// a build stuck on the event cache would otherwise hold the room's open forever. a big room on

/// a slow device can take tens of seconds to load from IndexedDB, so this is generous

const BUILD_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);

pub(super) enum ActorRequest {

    /// opening an open room answers with a fresh reset

    Open(Responder),

    Execute(TimelineCommand, Responder),

    Repair(HistoryRepairCommand, Responder),

    Close(Responder),

}

impl ActorRequest {

    pub(super) fn into_reply(self) -> Responder {

        match self {

            Self::Open(reply)

            | Self::Execute(_, reply)

            | Self::Repair(_, reply)

            | Self::Close(reply) => reply,

        }

    }

}

pub(super) enum ActorNote {

    /// stopping on its own, after a leave or a kick. anything sent its way now comes back in `Exited`

    Leaving {

        room_id: OwnedRoomId,

        generation: u64,

    },

    Exited {

        room_id: OwnedRoomId,

        generation: u64,

        /// requests that arrived too late to be served, for the engine to answer or replay

        leftovers: Vec<ActorRequest>,

    },

}

pub(super) struct RoomActor {

    pub(super) room_id: OwnedRoomId,

    pub(super) generation: u64,

    pub(super) room: Room,

    pub(super) requests: mpsc::Receiver<ActorRequest>,

    pub(super) notes: mpsc::UnboundedSender<ActorNote>,

    pub(super) context: SessionContext,

    pub(super) panics: mpsc::UnboundedReceiver<&'static str>,

    pub(super) opening: Vec<Responder>,

}

impl RoomActor {

    pub(super) async fn run(mut self) {

        self.serve().await;

        self.requests.close();

        let mut leftovers = Vec::new();

        while let Ok(request) = self.requests.try_recv() {

            leftovers.push(request);

        }

        let _ = self.notes.send(ActorNote::Exited {

            room_id: self.room_id.clone(),

            generation: self.generation,

            leftovers,

        });

    }

    async fn serve(&mut self) {

        tracing::debug!(room_id = %self.room_id, "building the room timeline");

        let build = build_timeline(

            &self.room,

            TimelineFocus::Live {

                hide_threaded_events: true,

            },

        );

        let timeout = matrix_sdk::sleep::sleep(BUILD_TIMEOUT);

        let timeline = tokio::select! {

            built = build => match built {

                Ok(timeline) => timeline,

                Err(error) => {

                    tracing::warn!(%error, room_id = %self.room_id, "room actor could not build its timeline");

                    let error = Error::with_source(ErrorKind::Timeline, "room timeline build failed", error);

                    return self.refuse_everything(error).await;

                }

            },

            () = timeout => {

                tracing::warn!(room_id = %self.room_id, "room timeline build timed out");

                let error = Error::new(ErrorKind::Timeline, "room timeline build timed out");

                return self.refuse_everything(error).await;

            }

        };

        tracing::debug!(room_id = %self.room_id, "starting the room timeline session");

        let Ok(room_timeline) = TimelineSession::start(

            TimelineId::Room(self.room_id.clone()),

            timeline,

            self.context.clone(),

        )

        .await

        else {

            return;

        };

        let (typing_guard, typing) = self.room.subscribe_to_typing_notifications();

        let mut typing_watcher = TaskGuard(self.context.supervisor.spawn(

            "typing watcher",

            watch_typing(self.room_id.clone(), typing, self.context.events.clone()),

        ));

        let mut timelines = Timelines {

            room: room_timeline,

            threads: Threads::new(self.requests.max_capacity()),

            work: TaskSet::new(),

        };

        let mut repair = RepairController::new(self.room.clone(), self.context.clone());

        tracing::debug!(room_id = %self.room_id, "starting recent history");

        repair.start_recent().await;

        tracing::debug!(room_id = %self.room_id, "room opened");

        for reply in self.opening.drain(..) {

            reply.done(Ok(()));

        }

        let close = self.serve_open(&mut timelines, &mut repair).await;

        let recent_settled = match repair.shutdown().await {

            Some(completion) => {

                let recent = matches!(completion.outcome, Outcome::Recent(_));

                self.finish_repair(completion, &mut timelines).await;

                recent

            }

            None => false,

        };

        if !recent_settled {

            let _ = self

                .context

                .events

                .send(Event::RecentHistoryChanged {

                    room_id: self.room_id.clone(),

                    state: RecentHistoryState::Idle,

                })

                .await;

        }

        if close.is_none() {

            self.requests.close();

            let _ = self.notes.send(ActorNote::Leaving {

                room_id: self.room_id.clone(),

                generation: self.generation,

            });

        }

        // let redactions and receipts that already left finish rather than lie about them

        while timelines.work.join_next().await.is_some() {}

        drop(typing_guard);

        typing_watcher.0.abort();

        let _ = (&mut typing_watcher.0).await;

        let _ = self

            .context

            .events

            .send(Event::TypingChanged(TypingState {

                room_id: self.room_id.clone(),

                users: Vec::new(),

            }))

            .await;

        timelines.threads.close().await;

        timelines.room.close().await;

        if let Some(reply) = close {

            reply.done(Ok(()));

        }

    }

    async fn finish_repair(&self, mut completion: Completion, timelines: &mut Timelines) {

        let mut cancellation = Ok(());

        // pages a failed or cancelled operation already committed still have to be on

        // screen before anyone hears it stopped

        if completion.committed

            && !completion.closing

            && let Err(error) = timelines.room.publish_history().await

        {

            if completion.result.is_ok() {

                completion.unpublished(error.clone());

            }

            cancellation = Err(error);

        }

        let event = match &completion.outcome {

            Outcome::Recent(state) => Event::RecentHistoryChanged {

                room_id: self.room_id.clone(),

                state: state.clone(),

            },

            Outcome::Fill(report) => Event::HistoryRepairChanged(report.clone()),

        };

        let mut resolved = completion.result.clone();

        let events = &self.context.events;

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

            resolved = Err(Error::interrupted());

            cancellation = Err(Error::interrupted());

        }

        let limit = self.context.initial_history_limit;

        if matches!(

            completion.outcome,

            Outcome::Recent(RecentHistoryState::Failed(_))

        ) && limit > 0

            && !timelines.room.has_events().await

        {

            timelines.room.paginate(limit, None);

        }

        completion.resolve(resolved, cancellation);

    }

    /// returns the close request's responder, if a close is why we stopped

    async fn serve_open(

        &mut self,

        timelines: &mut Timelines,

        repair: &mut RepairController,

    ) -> Option<Responder> {

        let mut room_info = self.room.subscribe_info();

        // a leave that landed while the timeline was building won't come round again

        if matches!(self.room.state(), RoomState::Left | RoomState::Banned) {

            return None;

        }

        loop {

            tokio::select! {

                Some(info) = room_info.next() => {

                    if matches!(info.state(), RoomState::Left | RoomState::Banned) {

                        tracing::debug!(room_id = %self.room_id, "room actor stopped because we left the room");

                        return None;

                    }

                }

                request = self.requests.recv() => match request {

                    None => return None,

                    Some(ActorRequest::Close(reply)) => return Some(reply),

                    Some(ActorRequest::Open(reply)) => reply.done(timelines.room.resync().await),

                    Some(ActorRequest::Execute(command, reply)) => {

                        let label = label(&command);

                        tracing::debug!(room_id = %self.room_id, command = %label, "room command");

                        timelines.execute(command, reply, &self.room).await;

                        tracing::debug!(room_id = %self.room_id, command = %label, "room command handed off");

                    }

                    Some(ActorRequest::Repair(command, reply)) => {

                        let label = label(&command);

                        tracing::debug!(room_id = %self.room_id, command = %label, "room repair command");

                        repair.execute(command, reply).await;

                        tracing::debug!(room_id = %self.room_id, command = %label, "room repair command handed off");

                    }

                },

                completion = repair.next_completion(), if repair.is_running() => {

                    if let Some(completion) = completion {

                        self.finish_repair(completion, timelines).await;

                    }

                }

                Some(built) = timelines.threads.builds.join_next(), if !timelines.threads.builds.is_empty() => {

                    timelines.thread_built(built, &self.room_id, &self.context).await;

                }

                Some(label) = self.panics.recv() => panic!("{label} panicked in room {}", self.room_id),

                Some(done) = timelines.work.join_next(), if !timelines.work.is_empty() => {

                    if let Err(error) = done && is_panic(&error) {

                        tracing::warn!(room_id = %self.room_id, %error, "timeline operation panicked");

                    }

                }

            }

        }

    }

    async fn refuse_everything(&mut self, error: Error) {

        for reply in self.opening.drain(..) {

            reply.done(Err(error.clone()));

        }

        self.requests.close();

        while let Some(request) = self.requests.recv().await {

            request.into_reply().done(Err(error.clone()));

        }

    }

}

async fn build_timeline(

    room: &Room,

    focus: TimelineFocus,

) -> Result<Timeline, matrix_sdk_ui::timeline::Error> {

    room.timeline_builder()

        .with_focus(focus)

        .track_read_marker_and_receipts(TimelineReadReceiptTracking::AllEvents)

        .build()

        .await

}

async fn watch_typing(

    room_id: OwnedRoomId,

    mut typing: tokio::sync::broadcast::Receiver<Vec<ruma::OwnedUserId>>,

    events: crate::transport::EventSender,

) {

    loop {

        match typing.recv().await {

            Ok(users) => {

                let state = TypingState {

                    room_id: room_id.clone(),

                    users,

                };

                if events.send(Event::TypingChanged(state)).await.is_err() {

                    return;

                }

            }

            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {

                tracing::warn!(%room_id, skipped, "typing watcher lagged");

            }

            Err(tokio::sync::broadcast::error::RecvError::Closed) => return,

        }

    }

}

struct Timelines {

    room: TimelineSession,

    threads: Threads,

    work: TaskSet<()>,

}

impl Timelines {

    async fn execute(&mut self, command: TimelineCommand, reply: Responder, room: &Room) {

        match command {

            TimelineCommand::OpenThread {

                room_id,

                root_event_id,

            } => self

                .threads

                .open(TimelineId::thread(room_id, root_event_id), reply, room),

            TimelineCommand::CloseThread {

                room_id,

                root_event_id,

            } => {

                let id = TimelineId::thread(room_id, root_event_id);

                reply.done(self.threads.close_one(&id).await);

            }

            TimelineCommand::OpenPinned(room_id) => {

                self.threads.open(TimelineId::Pinned(room_id), reply, room);

            }

            TimelineCommand::ClosePinned(room_id) => {

                reply.done(self.threads.close_one(&TimelineId::Pinned(room_id)).await);

            }

            TimelineCommand::Resync {

                timeline_id: TimelineId::Room(_),

            } => reply.done(self.room.resync().await),

            command => {

                let Some(id) = command

                    .timeline_id()

                    .filter(|id| !matches!(id, TimelineId::Room(_)))

                    .cloned()

                else {

                    return self.room.execute(command, reply, &mut self.work).await;

                };

                if matches!(id, TimelineId::Pinned(_))

                    && matches!(

                        command,

                        TimelineCommand::Send { .. } | TimelineCommand::Paginate { .. }

                    )

                {

                    return reply.done(Err(Error::new(

                        ErrorKind::Invalid,

                        "the pinned timeline can't send or paginate",

                    )));

                }

                if let TimelineCommand::MarkRead {

                    timeline_id,

                    event_id,

                } = &command

                    && !self.threads.open.contains_key(&id)

                {

                    let client = room.client();

                    let (timeline_id, event_id) = (timeline_id.clone(), event_id.clone());

                    self.work.spawn(async move {

                        reply.done(

                            super::commands::mark_read(&client, &timeline_id, event_id).await,

                        );

                    });

                    return;

                }

                match self.threads.route(id, command, reply) {

                    Routed::Open(session, TimelineCommand::Resync { .. }, reply) => {

                        reply.done(session.resync().await);

                    }

                    Routed::Open(session, command, reply) => {

                        session.execute(command, reply, &mut self.work).await;

                    }

                    Routed::Queued => {}

                }

            }

        }

    }

    async fn thread_built(

        &mut self,

        built: Result<Built, JoinError>,

        room_id: &OwnedRoomId,

        context: &SessionContext,

    ) {

        let Some((id, queued, timeline)) = self.threads.built(built) else {

            return;

        };

        let timeline = match timeline {

            Ok(timeline) => timeline,

            Err(error) => {

                tracing::warn!(%error, %room_id, ?id, "timeline build failed");

                let error = Error::with_source(ErrorKind::Timeline, "timeline build failed", error);

                for (_, reply) in queued {

                    reply.done(Err(error.clone()));

                }

                return;

            }

        };

        let Ok(session) = TimelineSession::start(id.clone(), timeline, context.clone()).await

        else {

            return;

        };

        let session = self.threads.open.entry(id).insert_entry(session).into_mut();

        for (command, reply) in queued {

            match command {

                None => reply.done(Ok(())),

                Some(TimelineCommand::Resync { .. }) => reply.done(session.resync().await),

                Some(command) => session.execute(command, reply, &mut self.work).await,

            }

        }

    }

}

type Built = (

    TimelineId,

    u64,

    Result<Timeline, matrix_sdk_ui::timeline::Error>,

);

/// `None` is the open itself, everything else waits behind it

type Queued = VecDeque<(Option<TimelineCommand>, Responder)>;

enum Routed<'a> {

    Open(&'a mut TimelineSession, TimelineCommand, Responder),

    Queued,

}

struct OpeningThread {

    generation: u64,

    build: AbortHandle,

    queued: Queued,

}

/// threads and the pinned timeline, which build off the actor and queue what arrives meanwhile

struct Threads {

    open: HashMap<TimelineId, TimelineSession>,

    opening: HashMap<TimelineId, OpeningThread>,

    builds: TaskSet<Built>,

    next_generation: u64,

    queued: usize,

    capacity: usize,

}

impl Threads {

    fn new(capacity: usize) -> Self {

        Self {

            open: HashMap::new(),

            opening: HashMap::new(),

            builds: TaskSet::new(),

            next_generation: 0,

            queued: 0,

            capacity,

        }

    }

    fn enqueue(&mut self, id: &TimelineId, command: Option<TimelineCommand>, reply: Responder) {

        if self.queued >= self.capacity {