macltok

use std::{collections::HashMap, sync::Arc};

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

use matrix_sdk::{Room, room::MessagesOptions};

use ruma::{OwnedEventId, OwnedUserId, UInt};

use serde_json::Value;

use crate::{

    Error, ErrorKind, EventContext, EventContextCommand, EventContextPage, MediaId, Sender,

    TimelineEvent, TimelineItem,

    engine::MediaService,

    normalise::{

        MediaRegistration, apply_profiles, normalise_context, normalise_observed, resolve_replies,

    },

};

const MAX_CONTEXT_LIMIT: u16 = 50;

pub(crate) async fn execute(

    room: Room,

    current_user_id: Option<OwnedUserId>,

    media: Arc<MediaService>,

    command: EventContextCommand,

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

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

        EventContextCommand::Fetch {

            event_id,

            before_limit,

            after_limit,

            ..

        } => {

            let (context, registrations, _) =

                capture(room, current_user_id, event_id, before_limit, after_limit).await?;

            media.register(registrations).await;

            return Ok(crate::CommandOutput::EventContext(context));

        }

        EventContextCommand::Before { token, limit, .. } => (token, limit, true),

        EventContextCommand::After { token, limit, .. } => (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 side = fetch_side(&room, Some(&token), limit, backwards).await?;

    let mut page = Assembly::new(&room, current_user_id.as_deref(), &side.values, None).await;

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

    media.register(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> {

    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 = raw

        .event

        .as_ref()

        .and_then(|event| serde_json::from_str::<Value>(event.raw().json().get()).ok())

        .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 page = Assembly::new(

        &room,

        current_user_id.as_deref(),

        &values,

        Some(target_index),

    )

    .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>,

    token: Option<String>,

}

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

            token: None,

        });

    };

    if limit == 0 {

        return Ok(Side {

            values: 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 values: Vec<Value> = messages

        .chunk

        .iter()

        .filter_map(|event| serde_json::from_str(event.raw().json().get()).ok())

        .collect();

    if backwards {

        values.reverse();

    }

    Ok(Side {

        values,

        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],

        keep: Option<usize>,

    ) -> Self {

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

        let mut items = page.items;

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

        )

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

    }

}

fn events(items: &mut [(usize, TimelineItem)]) -> impl Iterator<Item = &mut TimelineEvent> {

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

        TimelineItem::Event(event) => Some(&mut **event),

        _ => None,

    })

}

async fn fetch_reply(

    room: &Room,

    current_user_id: Option<&ruma::UserId>,

    event_id: OwnedEventId,

) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {

    let event = room

        .load_or_fetch_event(&event_id, None)

        .await

        .inspect_err(|error| tracing::debug!(%event_id, %error, "fetching a reply target failed"))

        .ok()?;

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

    let identity = format!("{}\n{event_id}", room.room_id());

    match normalise_observed(&value, current_user_id, &identity)? {

        (TimelineItem::Event(event), media) => Some((*event, media)),

        _ => None,

    }

}

/// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the

/// member list once for the room when it is

async fn profiles<'a>(

    room: &Room,

    user_ids: impl Iterator<Item = &'a OwnedUserId>,

) -> HashMap<OwnedUserId, Sender> {

    let mut unique: Vec<&OwnedUserId> = user_ids.collect();

    unique.sort();

    unique.dedup();

    join_all(unique.into_iter().map(|user_id| async move {

        let member = room.get_member(user_id).await.ok().flatten()?;

        Some((

            user_id.clone(),

            Sender {

                user_id: user_id.clone(),

                display_name: member.display_name().map(str::to_owned),

                avatar: member.avatar_url().map(MediaId::plain),

                name_ambiguous: member.name_ambiguous(),

            },

        ))

    }))

    .await

    .into_iter()

    .flatten()

    .collect()

}

fn validate_limit(limit: u16) -> Result<(), Error> {

    if limit > MAX_CONTEXT_LIMIT {

        Err(Error::new(ErrorKind::Invalid, "context limit exceeds 50"))

    } else {

        Ok(())

    }

}