continuwuity/src/client_server/message.rs

use crate::{database::DatabaseGuard, pdu::PduBuilder, utils, Error, Result, Ruma};
use ruma::{
    api::client::{
        error::ErrorKind,
        r0::message::{get_message_events, send_message_event},
    },
    events::EventType,
};
use std::{
    collections::{BTreeMap, HashSet},
    sync::Arc,
};

/// # `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 async fn send_message_event_route(
    db: DatabaseGuard,
    body: Ruma<send_message_event::Request<'_>>,
) -> Result<send_message_event::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(
        db.globals
            .roomid_mutex_state
            .write()
            .unwrap()
            .entry(body.room_id.clone())
            .or_default(),
    );
    let state_lock = mutex_state.lock().await;

    // Forbid m.room.encrypted if encryption is disabled
    if &body.event_type == "m.room.encrypted" && !db.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) =
        db.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(send_message_event::Response { event_id });
    }

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

    let event_id = db.rooms.build_and_append_pdu(
        PduBuilder {
            event_type: EventType::from(&*body.event_type),
            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,
        &db,
        &state_lock,
    )?;

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

    drop(state_lock);

    db.flush()?;

    Ok(send_message_event::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)
pub async fn get_message_events_route(
    db: DatabaseGuard,
    body: Ruma<get_message_events::Request<'_>>,
) -> Result<get_message_events::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");

    if !db.rooms.is_joined(sender_user, &body.room_id)? {
        return Err(Error::BadRequest(
            ErrorKind::Forbidden,
            "You don't have permission to view this room.",
        ));
    }

    let from = body
        .from
        .clone()
        .parse()
        .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Invalid `from` value."))?;

    let to = body.to.as_ref().map(|t| t.parse());

    db.rooms
        .lazy_load_confirm_delivery(sender_user, sender_device, &body.room_id, from)?;

    // Use limit or else 10
    let limit = body.limit.try_into().map_or(10_usize, |l: u32| l as usize);

    let next_token;

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

    let mut lazy_loaded = HashSet::new();

    match body.dir {
        get_message_events::Direction::Forward => {
            let events_after: Vec<_> = db
                .rooms
                .pdus_after(sender_user, &body.room_id, from)?
                .take(limit)
                .filter_map(|r| r.ok()) // Filter out buggy events
                .filter_map(|(pdu_id, pdu)| {
                    db.rooms
                        .pdu_count(&pdu_id)
                        .map(|pdu_count| (pdu_count, pdu))
                        .ok()
                })
                .take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
                .collect();

            for (_, event) in &events_after {
                if !db.rooms.lazy_load_was_sent_before(
                    sender_user,
                    sender_device,
                    &body.room_id,
                    &event.sender,
                )? {
                    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 = body.from.to_owned();
            resp.end = next_token.map(|count| count.to_string());
            resp.chunk = events_after;
        }
        get_message_events::Direction::Backward => {
            let events_before: Vec<_> = db
                .rooms
                .pdus_until(sender_user, &body.room_id, from)?
                .take(limit)
                .filter_map(|r| r.ok()) // Filter out buggy events
                .filter_map(|(pdu_id, pdu)| {
                    db.rooms
                        .pdu_count(&pdu_id)
                        .map(|pdu_count| (pdu_count, pdu))
                        .ok()
                })
                .take_while(|&(k, _)| Some(Ok(k)) != to) // Stop at `to`
                .collect();

            for (_, event) in &events_before {
                if !db.rooms.lazy_load_was_sent_before(
                    sender_user,
                    sender_device,
                    &body.room_id,
                    &event.sender,
                )? {
                    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 = body.from.to_owned();
            resp.end = next_token.map(|count| count.to_string());
            resp.chunk = events_before;
        }
    }

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

    if let Some(next_token) = next_token {
        db.rooms.lazy_load_mark_sent(
            sender_user,
            sender_device,
            &body.room_id,
            lazy_loaded,
            next_token,
        );
    }

    Ok(resp)
}