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, resume_unwind};

use super::repair::{InstallRequest, RepairController};

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

use crate::{

    Error, ErrorKind, Event, HistoryRepairCommand, TimelineCommand, TimelineId, TypingState,

    transport::Responder,

};

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

        let timeline = match build_timeline(

            &self.room,

            TimelineFocus::Live {

                hide_threaded_events: true,

            },

        )

        .await

        {

            Ok(timeline) => timeline,

            Err(error) => return self.refuse_everything(error).await,

        };

        let Ok(room_timeline) = TimelineSession::start(

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

            timeline,

            self.context.clone(),

        )

        .await

        else {

            return;

        };

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

            reply.done(Ok(()));

        }

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

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

        repair.shutdown().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(()));

        }

    }

    /// 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)) => {

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

                    }

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

                        repair.execute(command, reply).await;

                    }

                },

                install = repair.next_install(), if repair.is_running() => {

                    if let Some(install) = install {

                        timelines.install_history(install, &self.room, &self.context).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: matrix_sdk_ui::timeline::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);

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

        &mut self,

        install: InstallRequest,

        room: &Room,

        context: &SessionContext,

    ) {

        let InstallRequest {

            events,

            prev_batch,

            reply,

        } = install;

        let result = self

            .install_history_inner(events, prev_batch, room, context)

            .await;

        let _ = reply.send(result);

    }

    async fn install_history_inner(

        &mut self,

        events: Vec<matrix_sdk::deserialized_responses::TimelineEvent>,

        prev_batch: Option<String>,

        room: &Room,

        context: &SessionContext,

    ) -> Result<(), Error> {

        let threads_capacity = self.threads.capacity;

        self.threads.close().await;

        self.threads = Threads::new(threads_capacity);

        self.room.pause_for_reseed().await;

        let reseed = async {

            let (cache, _drop_handles) = room.event_cache().await.map_err(|source| {

                Error::with_source(ErrorKind::Timeline, "history event cache failed", source)

            })?;

            cache.reseed(events, prev_batch).await.map_err(|source| {

                Error::with_source(ErrorKind::Timeline, "history cache reseed failed", source)

            })

        }

        .await;

        if let Err(error) = reseed {

            return match self.room.resume_after_failed_reseed().await {

                Ok(()) => Err(error),

                Err(restart_error) => Err(restart_error),

            };

        }

        let timeline = match build_timeline(

            room,

            TimelineFocus::Live {

                hide_threaded_events: true,

            },

        )

        .await

        {

            Ok(timeline) => timeline,

            Err(source) => {

                let error =

                    Error::with_source(ErrorKind::Timeline, "room timeline rebuild failed", source);

                return match self.room.resume_after_failed_reseed().await {

                    Ok(()) => Err(error),

                    Err(restart_error) => Err(restart_error),

                };

            }

        };

        let timeline_id = self.room.id.clone();

        let mut context = context.clone();

        context.initial_history_limit = 0;

        let timeline = match TimelineSession::start(timeline_id, timeline, context).await {

            Ok(timeline) => timeline,

            Err(source) => {

                let error =

                    Error::with_source(ErrorKind::Timeline, "room timeline reset failed", source);

                return match self.room.resume_after_failed_reseed().await {

                    Ok(()) => Err(error),

                    Err(restart_error) => Err(restart_error),

                };

            }

        };

        let previous = std::mem::replace(&mut self.room, timeline);

        drop(previous);

        Ok(())

    }

    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 {