grapevine/src/api/client_server/message.rs

use std::{
    collections::{BTreeMap, HashSet},
    sync::Arc,
};

use ruma::{
    api::client::{
        error::ErrorKind,
        message::{get_message_events, send_message_event},
    },
    events::{StateEventType, TimelineEventType},
    uint, RoomId, UInt, UserId,
};

use crate::{
    service::{pdu::PduBuilder, rooms::timeline::PduCount},
    services, utils,
    utils::filter::{load_limit, CompiledRoomEventFilter},
    Ar, Error, PduEvent, Ra, Result,
};

/// # `PUT /_matrix/client/r0/rooms/{roomId}/send/{eventType}/{txnId}`
///
/// Send a message event into the room.
///
/// - Is a NOOP if the txn id was already used before and returns the same event
///   id again
/// - The only requirement for the content is that it has to be valid json
/// - Tries to send the event into the room, auth rules will determine if it is
///   allowed
pub(crate) async fn send_message_event_route(
    body: Ar<send_message_event::v3::Request>,
) -> Result<Ra<send_message_event::v3::Response>> {
    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
    let sender_device = body.sender_device.as_deref();

    let mutex_state = Arc::clone(
        services()
            .globals
            .roomid_mutex_state
            .write()
            .await
            .entry(body.room_id.clone())
            .or_default(),
    );
    let state_lock = mutex_state.lock().await;

    // Forbid m.room.encrypted if encryption is disabled
    if TimelineEventType::RoomEncrypted == body.event_type.to_string().into()
        && !services().globals.allow_encryption()
    {
        return Err(Error::BadRequest(
            ErrorKind::forbidden(),
            "Encryption has been disabled",
        ));
    }

    // Check if this is a new transaction id
    if let Some(response) = services().transaction_ids.existing_txnid(
        sender_user,
        sender_device,
        &body.txn_id,
    )? {
        // The client might have sent a txnid of the /sendToDevice endpoint
        // This txnid has no response associated with it
        if response.is_empty() {
            return Err(Error::BadRequest(
                ErrorKind::InvalidParam,
                "Tried to use txn id already used for an incompatible \
                 endpoint.",
            ));
        }

        let event_id = utils::string_from_bytes(&response)
            .map_err(|_| {
                Error::bad_database("Invalid txnid bytes in database.")
            })?
            .try_into()
            .map_err(|_| {
                Error::bad_database("Invalid event id in txnid data.")
            })?;
        return Ok(Ra(send_message_event::v3::Response {
            event_id,
        }));
    }

    let mut unsigned = BTreeMap::new();
    unsigned
        .insert("transaction_id".to_owned(), body.txn_id.to_string().into());

    let event_id = services()
        .rooms
        .timeline
        .build_and_append_pdu(
            PduBuilder {
                event_type: body.event_type.to_string().into(),
                content: serde_json::from_str(body.body.body.json().get())
                    .map_err(|_| {
                        Error::BadRequest(
                            ErrorKind::BadJson,
                            "Invalid JSON body.",
                        )
                    })?,
                unsigned: Some(unsigned),
                state_key: None,
                redacts: None,
            },
            sender_user,
            &body.room_id,
            &state_lock,
        )
        .await?;

    services().transaction_ids.add_txnid(
        sender_user,
        sender_device,
        &body.txn_id,
        event_id.as_bytes(),
    )?;

    drop(state_lock);

    Ok(Ra(send_message_event::v3::Response::new((*event_id).to_owned())))
}

/// # `GET /_matrix/client/r0/rooms/{roomId}/messages`
///
/// Allows paginating through room history.
///
/// - Only works if the user is joined (TODO: always allow, but only show events
///   where the user was
/// joined, depending on `history_visibility`)
#[allow(clippy::too_many_lines)]
pub(crate) async fn get_message_events_route(
    body: Ar<get_message_events::v3::Request>,
) -> Result<Ra<get_message_events::v3::Response>> {
    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
    let sender_device =
        body.sender_device.as_ref().expect("user is authenticated");

    let Ok(filter) = CompiledRoomEventFilter::try_from(&body.filter) else {
        return Err(Error::BadRequest(
            ErrorKind::InvalidParam,
            "invalid 'filter' parameter",
        ));
    };

    let from = match body.from.clone() {
        Some(from) => PduCount::try_from_string(&from)?,
        None => match body.dir {
            ruma::api::Direction::Forward => PduCount::MIN,
            ruma::api::Direction::Backward => PduCount::MAX,
        },
    };

    if !filter.room_allowed(&body.room_id) {
        return Ok(Ra(get_message_events::v3::Response {
            start: from.stringify(),
            end: None,
            chunk: vec![],
            state: vec![],
        }));
    }

    let to = body.to.as_ref().and_then(|t| PduCount::try_from_string(t).ok());

    services()
        .rooms
        .lazy_loading
        .lazy_load_confirm_delivery(
            sender_user,
            sender_device,
            &body.room_id,
            from,
        )
        .await?;

    let limit = body
        .limit
        .min(body.filter.limit.unwrap_or(UInt::MAX))
        .min(uint!(100))
        .try_into()
        .expect("0-100 should fit in usize");

    let next_token;

    let mut resp = get_message_events::v3::Response::new();

    let mut lazy_loaded = HashSet::new();

    match body.dir {
        ruma::api::Direction::Forward => {
            let events_after: Vec<_> = services()
                .rooms
                .timeline
                .pdus_after(sender_user, &body.room_id, from)?
                .take(load_limit(limit))
                .filter_map(Result::ok)
                .filter(|(_, pdu)| {
                    filter.pdu_event_allowed(pdu)
                        && visibility_filter(pdu, sender_user, &body.room_id)
                })
                .take_while(|&(k, _)| Some(k) != to)
                .take(limit)
                .collect();

            for (_, event) in &events_after {
                // * https://github.com/vector-im/element-android/issues/3417
                // * https://github.com/vector-im/element-web/issues/21034
                //
                // TODO: When the above issues are resolved, uncomment this:
                /*
                if !services().rooms.lazy_loading.lazy_load_was_sent_before(
                    sender_user,
                    sender_device,
                    &body.room_id,
                    &event.sender,
                )? {
                    lazy_loaded.insert(event.sender.clone());
                }
                */
                // And delete this line:
                lazy_loaded.insert(event.sender.clone());
            }

            next_token = events_after.last().map(|(count, _)| count).copied();

            let events_after: Vec<_> = events_after
                .into_iter()
                .map(|(_, pdu)| pdu.to_room_event())
                .collect();

            resp.start = from.stringify();
            resp.end = next_token.map(|count| count.stringify());
            resp.chunk = events_after;
        }
        ruma::api::Direction::Backward => {
            services()
                .rooms
                .timeline
                .backfill_if_required(&body.room_id, from)
                .await?;
            let events_before: Vec<_> = services()
                .rooms
                .timeline
                .pdus_until(sender_user, &body.room_id, from)?
                .take(load_limit(limit))
                .filter_map(Result::ok)
                .filter(|(_, pdu)| {
                    filter.pdu_event_allowed(pdu)
                        && visibility_filter(pdu, sender_user, &body.room_id)
                })
                .take_while(|&(k, _)| Some(k) != to)
                .take(limit)
                .collect();

            for (_, event) in &events_before {
                // * https://github.com/vector-im/element-android/issues/3417
                // * https://github.com/vector-im/element-web/issues/21034
                //
                // TODO: When the above issues are resolved, uncomment this:
                /*
                if !services().rooms.lazy_loading.lazy_load_was_sent_before(
                    sender_user,
                    sender_device,
                    &body.room_id,
                    &event.sender,
                )? {
                    lazy_loaded.insert(event.sender.clone());
                }
                */
                // And delete this line:
                lazy_loaded.insert(event.sender.clone());
            }

            next_token = events_before.last().map(|(count, _)| count).copied();

            let events_before: Vec<_> = events_before
                .into_iter()
                .map(|(_, pdu)| pdu.to_room_event())
                .collect();

            resp.start = from.stringify();
            resp.end = next_token.map(|count| count.stringify());
            resp.chunk = events_before;
        }
    }

    resp.state = Vec::new();
    for ll_id in &lazy_loaded {
        if let Some(member_event) =
            services().rooms.state_accessor.room_state_get(
                &body.room_id,
                &StateEventType::RoomMember,
                ll_id.as_str(),
            )?
        {
            resp.state.push(member_event.to_state_event());
        }
    }

    // * https://github.com/vector-im/element-android/issues/3417
    // * https://github.com/vector-im/element-web/issues/21034
    //
    // TODO: When the above issues are resolved, uncomment this:
    /*
    if let Some(next_token) = next_token {
        services().rooms.lazy_loading.lazy_load_mark_sent(
            sender_user,
            sender_device,
            &body.room_id,
            lazy_loaded,
            next_token,
        );
    }
    */

    Ok(Ra(resp))
}

fn visibility_filter(
    pdu: &PduEvent,
    user_id: &UserId,
    room_id: &RoomId,
) -> bool {
    services()
        .rooms
        .state_accessor
        .user_can_see_event(user_id, room_id, &pdu.event_id)
        .unwrap_or(false)
}