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::{Completion, RepairController};

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

use crate::{

    Error, ErrorKind, Event, HistoryRepairCommand, RecentHistoryState, 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;

        };

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

        repair.start_recent().await;

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

            reply.done(Ok(()));

        }

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

        if let Some(completion) = repair.shutdown().await {

            self.finish_repair(completion, &mut timelines).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 publication = timelines.room.publish_history().await;

        let mut cancellation = publication.clone();

        if let Err(error) = publication {

            completion.report.failure = Some(error.clone());

            completion.report.phase = crate::HistoryRepairPhase::Paused;

            completion.result = Err(error.clone());

            if completion.recent_state.is_some() {

                completion.recent_state = Some(RecentHistoryState::Failed(error.message));

            }

        }

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

        if let Some(state) = completion.recent_state.clone() {

            if self

                .context

                .events

                .send(Event::RecentHistoryChanged {

                    room_id: self.room_id.clone(),

                    state,

                })

                .await

                .is_err()

            {

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

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

            }

        }

        if self

            .context

            .events

            .publish_repair(completion.report.clone())

            .await

            .is_err()

        {

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

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

        }

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

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

                    }

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

                        repair.execute(command, reply).await;

                    }

                },

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

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

                ErrorKind::QueueFull,

                "thread command queue is full",

            )));

        }

        let Some(opening) = self.opening.get_mut(id) else {

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

                ErrorKind::TimelineNotOpen,

                "timeline is not open",

            )));

        };

        opening.queued.push_back((command, reply));

        self.queued += 1;

    }

    fn open(&mut self, id: TimelineId, reply: Responder, room: &Room) {

        if self.open.contains_key(&id) {

            return reply.done(Ok(()));

        }

        if !self.opening.contains_key(&id) {

            let generation = self.next_generation;

            self.next_generation = self.next_generation.wrapping_add(1);

            let room = room.clone();

            let build_id = id.clone();

            let build = self.builds.spawn(async move {

                let focus = match &build_id {

                    TimelineId::Thread { root_event_id, .. } => TimelineFocus::Thread {

                        root_event_id: root_event_id.clone(),

                    },

                    _ => TimelineFocus::PinnedEvents,

                };

                (build_id, generation, build_timeline(&room, focus).await)

            });

            self.opening.insert(

                id.clone(),

                OpeningThread {