macltok

use std::{

    collections::{HashMap, VecDeque},

    sync::Arc,

};

use futures_util::future::{join_all, try_join};

use matrix_sdk::{

    Client, Room, RoomState, deserialized_responses::TimelineEvent as SdkTimelineEvent,

    room::MessagesOptions,

};

use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId, UInt, push::Action};

use serde_json::Value;

use tokio::sync::{Mutex, broadcast::error::RecvError};

use crate::{

    Error, ErrorKind, Event, EventContext, EventContextCloseReason, EventContextCommand,

    EventContextPage, EventContextSession, Sender, TimelineEvent, TimelineItem,

    engine::MediaService,

    normalise::{

        MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,

        normalise_observed, resolve_replies,

    },

    transport::EventSender,

};

const MAX_CONTEXT_LIMIT: u16 = 50;

/// a consumer that never closes its jump views still holds at most this many

const MAX_SESSIONS: usize = 32;

/// open event contexts, oldest first, each owning the media its pages registered

pub(crate) struct Sessions {

    open: Mutex<VecDeque<Session>>,

    events: EventSender,

}

struct Session {

    id: String,

    room_id: OwnedRoomId,

    scope: MediaScope,

}

impl Session {

    /// media ids carry the session, so two owners of one event never share an entry

    fn media_prefix(scope: &MediaScope) -> String {

        format!("{}#", scope.nonce())

    }

}

impl Sessions {

    pub(crate) fn new(events: EventSender) -> Self {

        Self {

            open: Mutex::default(),

            events,

        }

    }

    async fn open(

        &self,

        media: &MediaService,

        room_id: OwnedRoomId,

        scope: MediaScope,

        registrations: Vec<MediaRegistration>,

    ) -> String {

        let id = format!("context-{}", scope.nonce());

        let mut evicted = Vec::new();

        {

            let mut open = self.open.lock().await;

            media.register(&scope, registrations).await;

            open.push_back(Session {

                id: id.clone(),

                room_id,

                scope,

            });

            while open.len() > MAX_SESSIONS {

                if let Some(oldest) = open.pop_front() {

                    media.release(&oldest.scope).await;

                    evicted.push(oldest.id);

                }

            }

        }

        self.announce(evicted, EventContextCloseReason::Evicted)

            .await;

        id

    }

    /// the room and the media prefix the session's pages use

    async fn find(&self, id: &str) -> Result<(OwnedRoomId, String), Error> {

        let open = self.open.lock().await;

        let session = open

            .iter()

            .find(|session| session.id == id)

            .ok_or_else(not_open)?;

        Ok((

            session.room_id.clone(),

            Session::media_prefix(&session.scope),

        ))

    }

    /// under the lock, so a page that lands after its close registers nothing

    async fn register(

        &self,

        media: &MediaService,

        id: &str,

        registrations: Vec<MediaRegistration>,

    ) -> Result<(), Error> {

        let open = self.open.lock().await;

        let session = open

            .iter()

            .find(|session| session.id == id)

            .ok_or_else(not_open)?;

        media.register(&session.scope, registrations).await;

        Ok(())

    }

    async fn close(&self, media: &MediaService, id: &str) -> Result<(), Error> {

        let mut open = self.open.lock().await;

        let index = open

            .iter()

            .position(|session| session.id == id)

            .ok_or_else(not_open)?;

        if let Some(session) = open.remove(index) {

            media.release(&session.scope).await;

        }

        Ok(())

    }

    async fn close_left(&self, media: &MediaService, client: &Client) {

        let mut closed = Vec::new();

        {

            let mut open = self.open.lock().await;

            let mut kept = VecDeque::with_capacity(open.len());

            for session in open.drain(..) {

                let left = client

                    .get_room(&session.room_id)

                    .is_none_or(|room| matches!(room.state(), RoomState::Left | RoomState::Banned));

                if left {

                    media.release(&session.scope).await;

                    closed.push(session.id);

                } else {

                    kept.push_back(session);

                }

            }

            *open = kept;

        }

        self.announce(closed, EventContextCloseReason::RoomLeft)

            .await;

    }

    async fn announce(&self, ids: Vec<String>, reason: EventContextCloseReason) {

        for context_id in ids {

            let event = Event::EventContextClosed {

                context_id,

                reason: reason.clone(),

            };

            if self.events.send(event).await.is_err() {

                return;

            }

        }

    }

    /// the media service is cleared beside this, so nothing is released one by one

    pub(crate) async fn clear(&self) {

        self.open.lock().await.clear();

    }

}

fn not_open() -> Error {

    Error::new(

        ErrorKind::EventContextNotOpen,

        "event context is closed or was never opened",

    )

}

/// leaving a room closes its event contexts

pub(super) async fn close_on_leave(

    client: Client,

    media: Arc<MediaService>,

    sessions: Arc<Sessions>,

) {

    let mut info = client.room_info_notable_update_receiver();

    loop {

        match info.recv().await {

            Ok(_) | Err(RecvError::Lagged(_)) => sessions.close_left(&media, &client).await,

            Err(RecvError::Closed) => return,

        }

    }

}

pub(crate) async fn execute(

    client: Client,

    media: Arc<MediaService>,

    sessions: Arc<Sessions>,

    command: EventContextCommand,

) -> Result<crate::CommandOutput, Error> {

    let current_user_id = client.user_id().map(ToOwned::to_owned);

    let room = |room_id: &OwnedRoomId| client.get_room(room_id).ok_or_else(Error::room_unknown);

    let (context_id, token, limit, backwards) = match command {

        EventContextCommand::Fetch {

            room_id,

            event_id,

            before_limit,

            after_limit,

        } => {

            let scope = media.context_scope();

            let (context, registrations, _) = capture_in(

                room(&room_id)?,

                current_user_id,

                event_id,

                before_limit,

                after_limit,

                &Session::media_prefix(&scope),

            )

            .await?;

            let context_id = sessions.open(&media, room_id, scope, registrations).await;

            return Ok(crate::CommandOutput::EventContext(EventContextSession {

                context_id,

                context,

            }));

        }

        EventContextCommand::Close { context_id } => {

            sessions.close(&media, &context_id).await?;

            return Ok(crate::CommandOutput::Done);

        }

        EventContextCommand::Before {

            context_id,

            token,

            limit,

        } => (context_id, token, limit, true),

        EventContextCommand::After {

            context_id,

            token,

            limit,

        } => (context_id, token, limit, false),

    };

    if limit == 0 {

        return Err(Error::new(

            ErrorKind::Invalid,

            "context page limit must be at least 1",

        ));

    }

    validate_limit(limit)?;

    if token.is_empty() {

        return Err(Error::new(

            ErrorKind::Invalid,

            "context pagination token is empty",

        ));

    }

    let (room_id, media_prefix) = sessions.find(&context_id).await?;

    let room = room(&room_id)?;

    let side = fetch_side(&room, Some(&token), limit, backwards).await?;

    let mut page = Assembly::new(

        &room,

        current_user_id.as_deref(),

        &side.values,

        &side.highlights,

        None,

        &media_prefix,

    )

    .await;

    let events = page.take(0..side.values.len());

    sessions.register(&media, &context_id, page.media).await?;

    Ok(crate::CommandOutput::EventContextPage(EventContextPage {

        events,

        next_token: side.token,

    }))

}

pub(crate) async fn capture(

    room: Room,

    current_user_id: Option<OwnedUserId>,

    event_id: OwnedEventId,

    before_limit: u16,

    after_limit: u16,

) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {

    capture_in(

        room,

        current_user_id,

        event_id,

        before_limit,

        after_limit,

        "",

    )

    .await

}

async fn capture_in(

    room: Room,

    current_user_id: Option<OwnedUserId>,

    event_id: OwnedEventId,

    before_limit: u16,

    after_limit: u16,

    media_prefix: &str,

) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {

    validate_limit(before_limit)?;

    validate_limit(after_limit)?;

    let raw = room

        .event_with_context(&event_id, false, UInt::from(0_u16), None)

        .await

        .map_err(|source| {

            let error = Error::from_sdk("fetching event context failed", source);

            if error

                .server

                .as_ref()

                .is_some_and(|server| server.status == 404)

            {

                Error::new(ErrorKind::EventNotFound, "event was not found")

            } else {

                error

            }

        })?;

    let (target, target_highlight) = raw

        .event

        .as_ref()

        .and_then(raw_event)

        .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event was not found"))?;

    let (before, after) = try_join(

        fetch_side(&room, raw.prev_batch_token.as_deref(), before_limit, true),

        fetch_side(&room, raw.next_batch_token.as_deref(), after_limit, false),

    )

    .await?;

    let target_index = before.values.len();

    let mut values = before.values;

    values.push(target);

    values.extend(after.values);

    let mut highlights = before.highlights;

    highlights.push(target_highlight);

    highlights.extend(after.highlights);

    let mut page = Assembly::new(

        &room,

        current_user_id.as_deref(),

        &values,

        &highlights,

        Some(target_index),

        media_prefix,

    )

    .await;

    let events_before = page.take(0..target_index);

    let event = page

        .take(target_index..target_index + 1)

        .pop()

        .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event could not be read"))?;

    let events_after = page.take(target_index + 1..values.len());

    Ok((

        EventContext {

            events_before,

            event,

            events_after,

            start_token: before.token,

            end_token: after.token,

        },

        page.media,

        values,

    ))

}

struct Side {

    /// in timeline order

    values: Vec<Value>,

    /// beside `values`, from the push actions the sdk worked out as it fetched

    highlights: Vec<bool>,

    token: Option<String>,

}

fn raw_event(event: &SdkTimelineEvent) -> Option<(Value, bool)> {

    let value = serde_json::from_str(event.raw().json().get()).ok()?;

    let highlight = event

        .push_actions()

        .is_some_and(|actions| actions.iter().any(Action::is_highlight));

    Some((value, highlight))

}

async fn fetch_side(

    room: &Room,

    token: Option<&str>,

    limit: u16,

    backwards: bool,

) -> Result<Side, Error> {

    let Some(token) = token else {

        return Ok(Side {

            values: Vec::new(),

            highlights: Vec::new(),

            token: None,

        });

    };

    if limit == 0 {

        return Ok(Side {

            values: Vec::new(),

            highlights: Vec::new(),

            token: Some(token.to_owned()),

        });

    }

    let mut options = if backwards {

        MessagesOptions::backward()

    } else {

        MessagesOptions::forward()

    };

    options = options.from(token);

    options.limit = UInt::from(limit);

    let messages = room

        .messages(options)

        .await

        .map_err(|source| Error::from_sdk("fetching event context events failed", source))?;

    let mut events: Vec<_> = messages.chunk.iter().filter_map(raw_event).collect();

    if backwards {

        events.reverse();

    }

    let (values, highlights) = events.into_iter().unzip();

    Ok(Side {

        values,

        highlights,

        token: messages.end,

    })

}

/// a window of raw events normalised together, so relations inside it land on their targets

struct Assembly {

    items: Vec<(usize, TimelineItem)>,

    media: Vec<MediaRegistration>,

}

impl Assembly {

    async fn new(

        room: &Room,

        current_user_id: Option<&ruma::UserId>,

        values: &[Value],

        highlights: &[bool],

        keep: Option<usize>,

        media_prefix: &str,

    ) -> Self {

        let page = normalise_context(values, keep, current_user_id, room.room_id(), media_prefix);

        let mut items = page.items;

        for (index, item) in &mut items {

            if let TimelineItem::Event(event) = item {

                event.highlight = highlights[*index];

            }

        }

        let mut media = page.media;

        let mut targets: HashMap<OwnedEventId, TimelineEvent> = HashMap::new();

        let mut missing = Vec::new();

        for (_, item) in &items {

            if let TimelineItem::Event(event) = item

                && let Some(event_id) = &event.event_id

            {

                targets.insert(event_id.clone(), (**event).clone());

            }

        }

        for event in events(&mut items) {

            if let Some(reply) = &event.reply_to

                && !targets.contains_key(&reply.event_id)

                && !missing.contains(&reply.event_id)

            {

                missing.push(reply.event_id.clone());

            }

        }

        let fetched = join_all(

            missing

                .into_iter()

                .map(|event_id| fetch_reply(room, current_user_id, event_id, media_prefix)),

        )

        .await;

        for (event, registrations) in fetched.into_iter().flatten() {

            media.extend(registrations);

            if let Some(event_id) = event.event_id.clone() {

                targets.insert(event_id, event);

            }

        }

        let profiles = profiles(

            room,

            items

                .iter()

                .filter_map(|(_, item)| match item {

                    TimelineItem::Event(event) => Some(&event.sender.user_id),

                    _ => None,

                })

                .chain(targets.values().map(|event| &event.sender.user_id)),

        )

        .await;

        apply_profiles(targets.values_mut(), &profiles);

        apply_profiles(events(&mut items), &profiles);

        resolve_replies(events(&mut items), &targets);

        Self { items, media }

    }

    /// the items that came from raw events in `range`

    fn take(&mut self, range: std::ops::Range<usize>) -> Vec<TimelineItem> {

        let (taken, kept) = std::mem::take(&mut self.items)

            .into_iter()

            .partition(|(index, _)| range.contains(index));

        self.items = kept;

        taken.into_iter().map(|(_, item)| item).collect()