conduit/src/client_server/message.rs

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

#[cfg(feature = "conduit_bin")]
use rocket::{get, put};

/// # `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
#[cfg_attr(
    feature = "conduit_bin",
    put("/_matrix/client/r0/rooms/<_>/send/<_>/<_>", data = "<body>")
)]
#[tracing::instrument(skip(db, body))]
pub async fn send_message_event_route(
    db: DatabaseGuard,
    body: Ruma<send_message_event::Request<'_>>,
) -> ConduitResult<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 }.into());
    }

    let mut unsigned = BTreeMap::new();
    unsigned.insert("transaction_id".to_owned(), body.txn_id.clone().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).into())
}

/// # `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)
#[cfg_attr(
    feature = "conduit_bin",
    get("/_matrix/client/r0/rooms/<_>/messages", data = "<body>")
)]
#[tracing::instrument(skip(db, body))]
pub async fn get_message_events_route(
    db: DatabaseGuard,
    body: Ruma<get_message_events::Request<'_>>,
) -> ConduitResult<get_message_events::Response> {
    let sender_user = body.sender_user.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());

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

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

            let end_token = events_after.last().map(|(count, _)| count.to_string());

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

            let mut resp = get_message_events::Response::new();
            resp.start = Some(body.from.to_owned());
            resp.end = end_token;
            resp.chunk = events_after;
            resp.state = Vec::new();

            Ok(resp.into())
        }
        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();

            let start_token = events_before.last().map(|(count, _)| count.to_string());

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

            let mut resp = get_message_events::Response::new();
            resp.start = Some(body.from.to_owned());
            resp.end = start_token;
            resp.chunk = events_before;
            resp.state = Vec::new();

            Ok(resp.into())
        }
    }
}
Sqlite 2021-07-14 07:07:08 +00:00			`use crate::{database::DatabaseGuard, pdu::PduBuilder, utils, ConduitResult, Error, Ruma};`
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`use ruma::{`
			`api::client::{`
			`error::ErrorKind,`
			`r0::message::{get_message_events, send_message_event},`
			`},`
improvement: update ruma 2021-04-22 11:26:20 +02:00			`events::EventType,`
Update state-res, use the new Event trait This also bumps ruma to latest and removes js_int infavor of the ruma re-export 2020-12-31 08:40:49 -05:00			`};`
Simplify identifier parsing code 2021-11-27 00:30:28 +01:00			`use std::{collections::BTreeMap, convert::TryInto, sync::Arc};`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
			`#[cfg(feature = "conduit_bin")]`
			`use rocket::{get, put};`

docs: documentation for every endpoint 2021-08-31 19:14:37 +02:00			/// # `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`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`#[cfg_attr(`
			`feature = "conduit_bin",`
			`put("/_matrix/client/r0/rooms/<_>/send/<_>/<_>", data = "<body>")`
			`)]`
feat: opentelemetry/jaeger support 2021-02-28 12:41:03 +01:00			`#[tracing::instrument(skip(db, body))]`
feat: send messages over federation 2020-09-14 20:23:19 +02:00			`pub async fn send_message_event_route(`
Sqlite 2021-07-14 07:07:08 +00:00			`db: DatabaseGuard,`
Update ruma version 2020-09-08 17:32:03 +02:00			`body: Ruma<send_message_event::Request<'_>>,`
Update ruma 2020-08-12 23:32:39 +02:00			`) -> ConduitResult<send_message_event::Response> {`
fix: use device_id when registering TIL body.device_id != (*body).device_id, which is pretty bad, so I renamed body.device_id to body.sender_device 2020-10-18 20:33:12 +02:00			`let sender_user = body.sender_user.as_ref().expect("user is authenticated");`
feat: implement appservices this also reverts some stateres changes 2020-12-08 10:33:44 +01:00			`let sender_device = body.sender_device.as_deref();`
feat: handle txn ids 2020-08-25 13:24:38 +02:00
fix: improve locks 2021-08-03 11:10:58 +02:00			`let mutex_state = Arc::clone(`
improvement: locks 2021-07-13 15:44:25 +02:00			`db.globals`
fix: improve locks 2021-08-03 11:10:58 +02:00			`.roomid_mutex_state`
improvement: locks 2021-07-13 15:44:25 +02:00			`.write()`
			`.unwrap()`
			`.entry(body.room_id.clone())`
			`.or_default(),`
			`);`
fix: improve locks 2021-08-03 11:10:58 +02:00			`let state_lock = mutex_state.lock().await;`
improvement: locks 2021-07-13 15:44:25 +02:00
Make allow_encryption work again, fixing #115 2021-09-24 22:44:26 +00:00			`// 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",`
			`));`
			`}`

feat: handle txn ids 2020-08-25 13:24:38 +02:00			`// Check if this is a new transaction id`
fix: use device_id when registering TIL body.device_id != (*body).device_id, which is pretty bad, so I renamed body.device_id to body.sender_device 2020-10-18 20:33:12 +02:00			`if let Some(response) =`
			`db.transaction_ids`
			`.existing_txnid(sender_user, sender_device, &body.txn_id)?`
feat: handle txn ids 2020-08-25 13:24:38 +02:00			`{`
			`// 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.",`
			`));`
			`}`

Simplify identifier parsing code 2021-11-27 00:30:28 +01:00			`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."))?;`
feat: handle txn ids 2020-08-25 13:24:38 +02:00			`return Ok(send_message_event::Response { event_id }.into());`
			`}`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
Update state-res, use the new Event trait This also bumps ruma to latest and removes js_int infavor of the ruma re-export 2020-12-31 08:40:49 -05:00			`let mut unsigned = BTreeMap::new();`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`unsigned.insert("transaction_id".to_owned(), body.txn_id.clone().into());`

feat: admin room 2020-10-05 22:19:22 +02:00			`let event_id = db.rooms.build_and_append_pdu(`
			`PduBuilder {`
improvement: update ruma 2021-04-22 11:26:20 +02:00			`event_type: EventType::from(&body.event_type),`
improvement: uiaa works like in synapse 2021-05-04 19:03:18 +02:00			`content: serde_json::from_str(body.body.body.json().get())`
			`.map_err(\|_\| Error::BadRequest(ErrorKind::BadJson, "Invalid JSON body."))?,`
feat: admin room 2020-10-05 22:19:22 +02:00			`unsigned: Some(unsigned),`
			`state_key: None,`
			`redacts: None,`
			`},`
Clean up (mostly automated with cargo clippy --fix) 2021-09-13 19:45:56 +02:00			`sender_user,`
feat: admin room 2020-10-05 22:19:22 +02:00			`&body.room_id,`
Fix ALL clippy warnings 2021-01-15 11:05:57 -05:00			`&db,`
fix: improve locks 2021-08-03 11:10:58 +02:00			`&state_lock,`
feat: admin room 2020-10-05 22:19:22 +02:00			`)?;`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
fix: use device_id when registering TIL body.device_id != (*body).device_id, which is pretty bad, so I renamed body.device_id to body.sender_device 2020-10-18 20:33:12 +02:00			`db.transaction_ids.add_txnid(`
			`sender_user,`
			`sender_device,`
			`&body.txn_id,`
			`event_id.as_bytes(),`
			`)?;`
Merge remote-tracking branch 'origin/master' into federation 2020-09-12 22:13:53 +02:00
fix: improve locks 2021-08-03 11:10:58 +02:00			`drop(state_lock);`
improvement: locks 2021-07-13 15:44:25 +02:00
improvement: make better use of sqlite connections 2021-08-02 10:13:34 +02:00			`db.flush()?;`
improvement: flush after every request that manipulates the db 2020-10-21 21:28:02 +02:00
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`Ok(send_message_event::Response::new(event_id).into())`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`}`

docs: documentation for every endpoint 2021-08-31 19:14:37 +02:00			/// # `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)`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`#[cfg_attr(`
			`feature = "conduit_bin",`
			`get("/_matrix/client/r0/rooms/<_>/messages", data = "<body>")`
			`)]`
feat: opentelemetry/jaeger support 2021-02-28 12:41:03 +01:00			`#[tracing::instrument(skip(db, body))]`
improvement: flush after every request that manipulates the db 2020-10-21 21:28:02 +02:00			`pub async fn get_message_events_route(`
Sqlite 2021-07-14 07:07:08 +00:00			`db: DatabaseGuard,`
Update ruma version 2020-09-08 17:32:03 +02:00			`body: Ruma<get_message_events::Request<'_>>,`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`) -> ConduitResult<get_message_events::Response> {`
fix: use device_id when registering TIL body.device_id != (*body).device_id, which is pretty bad, so I renamed body.device_id to body.sender_device 2020-10-18 20:33:12 +02:00			`let sender_user = body.sender_user.as_ref().expect("user is authenticated");`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
fix: use device_id when registering TIL body.device_id != (*body).device_id, which is pretty bad, so I renamed body.device_id to body.sender_device 2020-10-18 20:33:12 +02:00			`if !db.rooms.is_joined(sender_user, &body.room_id)? {`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`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());`

			`// Use limit or else 10`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`let limit = body.limit.try_into().map_or(10_usize, \|l: u32\| l as usize);`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
			`match body.dir {`
			`get_message_events::Direction::Forward => {`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`let events_after: Vec<_> = db`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.rooms`
Clean up (mostly automated with cargo clippy --fix) 2021-09-13 19:45:56 +02:00			`.pdus_after(sender_user, &body.room_id, from)?`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.take(limit)`
			`.filter_map(\|r\| r.ok()) // Filter out buggy events`
improvement: more efficient /sync with gaps 2020-09-17 14:44:47 +02:00			`.filter_map(\|(pdu_id, pdu)\| {`
			`db.rooms`
			`.pdu_count(&pdu_id)`
			`.map(\|pdu_count\| (pdu_count, pdu))`
			`.ok()`
			`})`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			.take_while(\|&(k, _)\| Some(Ok(k)) != to) // Stop at `to`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`.collect();`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
			`let end_token = events_after.last().map(\|(count, _)\| count.to_string());`

Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`let events_after: Vec<_> = events_after`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.into_iter()`
			`.map(\|(_, pdu)\| pdu.to_room_event())`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`.collect();`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`let mut resp = get_message_events::Response::new();`
Update ruma version 2020-09-08 17:32:03 +02:00			`resp.start = Some(body.from.to_owned());`
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`resp.end = end_token;`
			`resp.chunk = events_after;`
			`resp.state = Vec::new();`

			`Ok(resp.into())`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`}`
			`get_message_events::Direction::Backward => {`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`let events_before: Vec<_> = db`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.rooms`
Clean up (mostly automated with cargo clippy --fix) 2021-09-13 19:45:56 +02:00			`.pdus_until(sender_user, &body.room_id, from)?`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.take(limit)`
			`.filter_map(\|r\| r.ok()) // Filter out buggy events`
improvement: more efficient /sync with gaps 2020-09-17 14:44:47 +02:00			`.filter_map(\|(pdu_id, pdu)\| {`
			`db.rooms`
			`.pdu_count(&pdu_id)`
			`.map(\|pdu_count\| (pdu_count, pdu))`
			`.ok()`
			`})`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			.take_while(\|&(k, _)\| Some(Ok(k)) != to) // Stop at `to`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`.collect();`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
			`let start_token = events_before.last().map(\|(count, _)\| count.to_string());`

Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`let events_before: Vec<_> = events_before`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`.into_iter()`
			`.map(\|(_, pdu)\| pdu.to_room_event())`
Reduce turbofish usage Should make the code a little bit easier to read. 2021-10-13 11:51:30 +02:00			`.collect();`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`let mut resp = get_message_events::Response::new();`
Update ruma version 2020-09-08 17:32:03 +02:00			`resp.start = Some(body.from.to_owned());`
Keep track of State at event for state resolution feat: first steps towards joining rooms over federation Add state-res as a dependency of conduit Add reverse_topological_power_sort before append_pdu Implement statehashstatid_pduid tree for keeping track of state Clean up implementation of state_hash as key for tracking state 2020-08-06 08:29:59 -04:00			`resp.end = start_token;`
			`resp.chunk = events_before;`
			`resp.state = Vec::new();`

			`Ok(resp.into())`
refactor: put endpoints into modules 2020-07-30 18:14:47 +02:00			`}`
			`}`
			`}`