use rusqlite::{params, Connection, OptionalExtension, Transaction};
use std::collections::HashSet;
use std::fmt;
static SERVER_NAME_CELL: std::sync::OnceLock<String> = std::sync::OnceLock::new();
pub fn set_matrix_server_name(name: impl Into<String>) -> Result<(), &'static str> {
let name = name.into();
if name.is_empty() || name.contains(':') || name.contains('/') {
return Err("server name must be one hostname");
}
SERVER_NAME_CELL.set(name).map_err(|_| "server name already set")
}
static LOCAL_ALIASES: std::sync::OnceLock<Vec<String>> = std::sync::OnceLock::new();
pub fn set_local_aliases(names: impl IntoIterator<Item = String>) {
let list: Vec<String> = names.into_iter().map(|n| n.trim().to_ascii_lowercase()).filter(|n| !n.is_empty() && !n.contains(':') && !n.contains('/')).collect();
let _ = LOCAL_ALIASES.set(list);
}
pub fn is_local_server_name(name: &str) -> bool {
name == matrix_server_name() || LOCAL_ALIASES.get().is_some_and(|aliases| aliases.iter().any(|a| a.eq_ignore_ascii_case(name)))
}
pub fn matrix_server_name() -> &'static str {
SERVER_NAME_CELL.get_or_init(|| "example.org".to_string()).as_str()
}
pub const MATRIX_ROOM_VERSION: &str = "11";
pub const MATRIX_EVENT_CONTENT_MAX_BYTES: usize = 16 * 1024;
pub const RESERVED_LOCALPART_PREFIX: &str = "_bridge_";
pub fn is_reserved_localpart(localpart: &str) -> bool {
localpart.starts_with(RESERVED_LOCALPART_PREFIX)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MatrixIdError {
MissingSigil,
MissingServerName,
ForeignServerName,
}
impl fmt::Display for MatrixIdError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
MatrixIdError::MissingSigil => write!(f, "mxid is missing its '@' sigil"),
MatrixIdError::MissingServerName => write!(f, "mxid is missing a ':server_name' suffix"),
MatrixIdError::ForeignServerName => write!(f, "mxid is addressed to a foreign server name"),
}
}
}
pub fn mxid_for_public_id(public_id: &str) -> String {
format!("@{public_id}:{}", matrix_server_name())
}
pub fn public_id_from_mxid(mxid: &str) -> Result<&str, MatrixIdError> {
let rest = mxid.strip_prefix('@').ok_or(MatrixIdError::MissingSigil)?;
let (localpart, server_name) = rest.split_once(':').ok_or(MatrixIdError::MissingServerName)?;
if !is_local_server_name(server_name) {
return Err(MatrixIdError::ForeignServerName);
}
Ok(localpart)
}
fn random_id_component() -> String {
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use rand::Rng;
let bytes: [u8; 16] = rand::thread_rng().gen();
URL_SAFE_NO_PAD.encode(bytes)
}
pub fn new_room_id() -> String {
format!("!{}:{}", random_id_component(), matrix_server_name())
}
pub fn new_event_id() -> String {
format!("${}", random_id_component())
}
#[derive(Debug)]
pub enum MatrixStoreError {
Db(rusqlite::Error),
Json(serde_json::Error),
ReservedLocalpart,
DuplicateAnnotation,
UnknownEventId(String),
UnknownMxid(String),
InvalidMembership(String),
InvalidRelationTarget(String),
WrongRoom(String),
UnredactableEvent(String),
}
impl From<rusqlite::Error> for MatrixStoreError {
fn from(e: rusqlite::Error) -> Self {
MatrixStoreError::Db(e)
}
}
impl From<serde_json::Error> for MatrixStoreError {
fn from(e: serde_json::Error) -> Self {
MatrixStoreError::Json(e)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RoomKind {
Dm,
Group,
Channel,
}
impl RoomKind {
pub fn as_str(self) -> &'static str {
match self {
RoomKind::Dm => "dm",
RoomKind::Group => "group",
RoomKind::Channel => "channel",
}
}
pub fn from_wire_name(s: &str) -> Option<Self> {
match s {
"dm" => Some(RoomKind::Dm),
"group" => Some(RoomKind::Group),
"channel" => Some(RoomKind::Channel),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JoinRule {
Invite,
Public,
}
impl JoinRule {
pub fn as_str(self) -> &'static str {
match self {
JoinRule::Invite => "invite",
JoinRule::Public => "public",
}
}
pub fn from_wire_name(s: &str) -> Option<Self> {
match s {
"invite" => Some(JoinRule::Invite),
"public" => Some(JoinRule::Public),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HistoryVisibility {
Shared,
WorldReadable,
Invited,
Joined,
}
impl HistoryVisibility {
pub fn as_str(self) -> &'static str {
match self {
HistoryVisibility::Shared => "shared",
HistoryVisibility::WorldReadable => "world_readable",
HistoryVisibility::Invited => "invited",
HistoryVisibility::Joined => "joined",
}
}
pub fn from_wire_name(s: &str) -> Option<Self> {
match s {
"shared" => Some(HistoryVisibility::Shared),
"world_readable" => Some(HistoryVisibility::WorldReadable),
"invited" => Some(HistoryVisibility::Invited),
"joined" => Some(HistoryVisibility::Joined),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Membership {
Join,
Invite,
Leave,
Ban,
}
impl Membership {
pub fn as_str(self) -> &'static str {
match self {
Membership::Join => "join",
Membership::Invite => "invite",
Membership::Leave => "leave",
Membership::Ban => "ban",
}
}
pub fn from_wire_name(s: &str) -> Option<Self> {
match s {
"join" => Some(Membership::Join),
"invite" => Some(Membership::Invite),
"leave" => Some(Membership::Leave),
"ban" => Some(Membership::Ban),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReceiptType {
Read,
ReadPrivate,
}
impl ReceiptType {
pub fn as_str(self) -> &'static str {
match self {
ReceiptType::Read => "m.read",
ReceiptType::ReadPrivate => "m.read.private",
}
}
pub fn from_wire_name(s: &str) -> Option<Self> {
match s {
"m.read" => Some(ReceiptType::Read),
"m.read.private" => Some(ReceiptType::ReadPrivate),
_ => None,
}
}
}
fn decode_enum<T>(idx: usize, column: &'static str, raw: &str, parse: fn(&str) -> Option<T>) -> rusqlite::Result<T> {
parse(raw).ok_or_else(|| rusqlite::Error::InvalidColumnType(idx, column.to_string(), rusqlite::types::Type::Text))
}
pub fn create_matrix_schema(conn: &Connection) -> rusqlite::Result<()> {
conn.execute_batch(
r#"
-- Global stream ordering — see this module's doc comment on the
-- single-writer guarantee that makes `UPDATE ... RETURNING` safe.
CREATE TABLE IF NOT EXISTS stream_counter (
id INTEGER PRIMARY KEY CHECK (id = 1),
value INTEGER NOT NULL
);
INSERT OR IGNORE INTO stream_counter (id, value) VALUES (1, 0);
-- Federation F0: this server's signing keys and the verify keys
-- cached from remote servers. Secrets live only in the (encrypted) DB.
CREATE TABLE IF NOT EXISTS fed_signing_keys (
key_id TEXT PRIMARY KEY,
secret BLOB NOT NULL,
created_ms INTEGER NOT NULL,
retired_ms INTEGER
);
CREATE TABLE IF NOT EXISTS fed_remote_keys (
server_name TEXT NOT NULL,
key_id TEXT NOT NULL,
public_key TEXT NOT NULL,
valid_until_ms INTEGER NOT NULL,
fetched_ms INTEGER NOT NULL,
PRIMARY KEY (server_name, key_id)
);
-- user_id -> mxid, filled on first touch by ensure_matrix_user
-- (plan §2 manager decision: new table, not in the original DDL
-- text). public_id is immutable, so this mapping never changes.
CREATE TABLE IF NOT EXISTS matrix_users (
user_id INTEGER PRIMARY KEY,
mxid TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
nick TEXT
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_matrix_users_nick_lower
ON matrix_users(LOWER(nick)) WHERE nick IS NOT NULL;
-- Nick belongs to a session, not to matrix_users and not to the
-- device. device_id is only a mark. One device may have many sessions.
-- matrix_users.nick stays for old databases and is not read.
CREATE TABLE IF NOT EXISTS messenger_sessions (
session_id TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
device_id TEXT NOT NULL,
nick TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messenger_sessions_user
ON messenger_sessions(user_id);
CREATE UNIQUE INDEX IF NOT EXISTS idx_messenger_sessions_nick_lower
ON messenger_sessions(LOWER(nick));
CREATE TABLE IF NOT EXISTS rooms (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
room_version TEXT NOT NULL DEFAULT '11',
creator_user_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
is_encrypted INTEGER NOT NULL DEFAULT 0,
join_rule TEXT NOT NULL DEFAULT 'invite',
history_visibility TEXT NOT NULL DEFAULT 'shared',
dm_pair_key TEXT UNIQUE,
legacy_dm_id INTEGER UNIQUE
);
CREATE INDEX IF NOT EXISTS idx_rooms_kind ON rooms(kind);
CREATE TABLE IF NOT EXISTS events (
stream_id INTEGER PRIMARY KEY,
event_id TEXT NOT NULL UNIQUE,
room_id TEXT NOT NULL REFERENCES rooms(id),
sender_user_id INTEGER NOT NULL,
event_type TEXT NOT NULL,
state_key TEXT,
content TEXT NOT NULL,
origin_server_ts INTEGER NOT NULL,
txn_id TEXT,
redacts TEXT REFERENCES events(event_id),
redacted_by TEXT REFERENCES events(event_id)
);
CREATE INDEX IF NOT EXISTS idx_events_room_stream ON events(room_id, stream_id);
CREATE INDEX IF NOT EXISTS idx_events_room_type_state ON events(room_id, event_type, state_key);
CREATE INDEX IF NOT EXISTS idx_events_sender ON events(sender_user_id, stream_id);
CREATE TABLE IF NOT EXISTS current_state (
room_id TEXT NOT NULL REFERENCES rooms(id),
event_type TEXT NOT NULL,
state_key TEXT NOT NULL,
event_id TEXT NOT NULL REFERENCES events(event_id),
PRIMARY KEY (room_id, event_type, state_key)
);
CREATE TABLE IF NOT EXISTS room_members (
room_id TEXT NOT NULL REFERENCES rooms(id),
user_id INTEGER NOT NULL,
membership TEXT NOT NULL,
power_level INTEGER,
updated_at TEXT NOT NULL,
PRIMARY KEY (room_id, user_id)
);
CREATE INDEX IF NOT EXISTS idx_room_members_user ON room_members(user_id, membership);
CREATE TABLE IF NOT EXISTS relations (
event_id TEXT PRIMARY KEY REFERENCES events(event_id),
room_id TEXT NOT NULL REFERENCES rooms(id),
rel_type TEXT NOT NULL,
target_id TEXT NOT NULL REFERENCES events(event_id),
agg_key TEXT
);
CREATE INDEX IF NOT EXISTS idx_relations_target ON relations(target_id, rel_type);
CREATE TABLE IF NOT EXISTS receipts (
room_id TEXT NOT NULL REFERENCES rooms(id),
user_id INTEGER NOT NULL,
receipt_type TEXT NOT NULL,
event_id TEXT NOT NULL REFERENCES events(event_id),
ts INTEGER NOT NULL,
stream_id INTEGER NOT NULL,
PRIMARY KEY (room_id, user_id, receipt_type)
);
CREATE INDEX IF NOT EXISTS idx_receipts_room_stream ON receipts(room_id, stream_id);
-- DEVIATION from plan §2's literal text: room_id has no
-- `REFERENCES rooms(id)` — see this function's doc comment.
CREATE TABLE IF NOT EXISTS account_data (
user_id INTEGER NOT NULL,
room_id TEXT NOT NULL DEFAULT '',
data_type TEXT NOT NULL,
content TEXT NOT NULL,
stream_id INTEGER NOT NULL,
PRIMARY KEY (user_id, room_id, data_type)
);
CREATE INDEX IF NOT EXISTS idx_account_data_user_stream ON account_data(user_id, stream_id);
CREATE TABLE IF NOT EXISTS txn_dedup (
user_id INTEGER NOT NULL,
device_id TEXT NOT NULL,
txn_id TEXT NOT NULL,
event_id TEXT REFERENCES events(event_id),
created_at TEXT NOT NULL,
PRIMARY KEY (user_id, device_id, txn_id)
);
CREATE TABLE IF NOT EXISTS filters (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
definition TEXT NOT NULL
);
-- legacy_dm_message_map removed (M2); drop_legacy_dm_scaffold_if_empty cleans old DBs
"#,
)?;
crate::public_channels::create_public_schema(conn)?;
crate::public_forum::create_forum_schema(conn)?;
crate::media::create_media_schema(conn)?;
crate::fed_rooms::create_fed_schema(conn)?;
crate::identities::create_identities_schema(conn)
}
pub fn parse_db_key(key_hex: &str) -> Result<[u8; 32], String> {
let h = key_hex.trim();
if h.len() != 64 || !h.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err("database key must be 64 hex characters (32 bytes)".into());
}
let mut key = [0u8; 32];
for (i, b) in key.iter_mut().enumerate() {
*b = u8::from_str_radix(&h[2 * i..2 * i + 2], 16).map_err(|_| "database key must be hex".to_string())?;
}
Ok(key)
}
pub fn messenger_db_config(path: &str, key_hex: &str) -> Result<tesserax_store::DbConfig, String> {
let key = parse_db_key(key_hex)?;
Ok(tesserax_store::DbConfig::encrypted_native(path, std::sync::Arc::new(tesserax_store::keysource::StaticKeySource(key))))
}
pub fn open_messenger_db(path: &str, key_hex: &str) -> Result<tesserax_store::Db, String> {
let cfg = messenger_db_config(path, key_hex)?;
let db = tesserax_store::Db::open(&cfg).map_err(|e| e.to_string())?;
db.blocking(|conn| ensure_schema(conn)).map_err(|e| e.to_string())?;
Ok(db)
}
pub fn open_read_pool(path: &str, key_hex: &str, size: usize) -> Result<tesserax_store::ReadPool, String> {
let cfg = messenger_db_config(path, key_hex)?;
tesserax_store::ReadPoolConfig::from_config(cfg).pool_size(size.max(1)).open().map_err(|e| e.to_string())
}
pub fn ensure_schema(conn: &Connection) -> rusqlite::Result<()> {
create_matrix_schema(conn)?;
crate::keys::create_matrix_keys_schema(conn)?;
crate::retention::create_retention_schema(conn)?;
crate::public_channels::create_public_schema(conn)?;
Ok(())
}
pub(crate) fn next_stream_id(tx: &Transaction) -> rusqlite::Result<i64> {
tx.query_row("UPDATE stream_counter SET value = value + 1 WHERE id = 1 RETURNING value", [], |row| row.get(0))
}
pub fn max_stream_id(conn: &Connection) -> rusqlite::Result<i64> {
conn.query_row("SELECT value FROM stream_counter WHERE id = 1", [], |row| row.get(0))
}
pub fn ensure_matrix_user(conn: &Connection, user_id: i64, public_id: &str, now: &str) -> Result<String, MatrixStoreError> {
if is_reserved_localpart(public_id) {
return Err(MatrixStoreError::ReservedLocalpart);
}
let mxid = mxid_for_public_id(public_id);
conn.execute(
"INSERT INTO matrix_users (user_id, mxid, created_at) VALUES (?1, ?2, ?3)
ON CONFLICT(user_id) DO NOTHING",
params![user_id, mxid, now],
)?;
Ok(mxid)
}
pub fn mxid_of(conn: &Connection, user_id: i64) -> rusqlite::Result<Option<String>> {
conn.query_row("SELECT mxid FROM matrix_users WHERE user_id = ?1", params![user_id], |row| row.get(0))
.optional()
}
pub fn user_id_of(conn: &Connection, mxid: &str) -> rusqlite::Result<Option<i64>> {
conn.query_row("SELECT user_id FROM matrix_users WHERE mxid = ?1", params![mxid], |row| row.get(0))
.optional()
}
#[derive(Debug, Clone, PartialEq)]
pub struct Room {
pub id: String,
pub kind: RoomKind,
pub room_version: String,
pub creator_user_id: i64,
pub created_at: String,
pub is_encrypted: bool,
pub join_rule: JoinRule,
pub history_visibility: HistoryVisibility,
pub dm_pair_key: Option<String>,
pub legacy_dm_id: Option<i64>,
}
const ROOM_SELECT_COLUMNS: &str =
"id, kind, room_version, creator_user_id, created_at, is_encrypted, join_rule, history_visibility, dm_pair_key, legacy_dm_id";
fn room_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Room> {
let kind_raw: String = row.get(1)?;
let join_rule_raw: String = row.get(6)?;
let history_visibility_raw: String = row.get(7)?;
Ok(Room {
id: row.get(0)?,
kind: decode_enum(1, "kind", &kind_raw, RoomKind::from_wire_name)?,
room_version: row.get(2)?,
creator_user_id: row.get(3)?,
created_at: row.get(4)?,
is_encrypted: row.get(5)?,
join_rule: decode_enum(6, "join_rule", &join_rule_raw, JoinRule::from_wire_name)?,
history_visibility: decode_enum(7, "history_visibility", &history_visibility_raw, HistoryVisibility::from_wire_name)?,
dm_pair_key: row.get(8)?,
legacy_dm_id: row.get(9)?,
})
}
#[allow(clippy::too_many_arguments)]
fn insert_room_row(
conn: &Connection,
room_id: &str,
kind: RoomKind,
creator_user_id: i64,
created_at: &str,
is_encrypted: bool,
join_rule: JoinRule,
history_visibility: HistoryVisibility,
dm_pair_key: Option<&str>,
legacy_dm_id: Option<i64>,
) -> rusqlite::Result<()> {
conn.execute(
"INSERT INTO rooms (id, kind, room_version, creator_user_id, created_at, is_encrypted, join_rule, history_visibility, dm_pair_key, legacy_dm_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
room_id,
kind.as_str(),
MATRIX_ROOM_VERSION,
creator_user_id,
created_at,
is_encrypted,
join_rule.as_str(),
history_visibility.as_str(),
dm_pair_key,
legacy_dm_id,
],
)?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub fn create_room(
conn: &Connection,
room_id: &str,
kind: RoomKind,
creator_user_id: i64,
created_at: &str,
is_encrypted: bool,
join_rule: JoinRule,
history_visibility: HistoryVisibility,
dm_pair_key: Option<&str>,
legacy_dm_id: Option<i64>,
) -> rusqlite::Result<()> {
insert_room_row(conn, room_id, kind, creator_user_id, created_at, is_encrypted, join_rule, history_visibility, dm_pair_key, legacy_dm_id)
}
pub fn get_room(conn: &Connection, room_id: &str) -> rusqlite::Result<Option<Room>> {
conn.query_row(&format!("SELECT {ROOM_SELECT_COLUMNS} FROM rooms WHERE id = ?1"), params![room_id], room_from_row)
.optional()
}
pub fn room_by_dm_pair_key(conn: &Connection, pair_key: &str) -> rusqlite::Result<Option<Room>> {
conn.query_row(&format!("SELECT {ROOM_SELECT_COLUMNS} FROM rooms WHERE dm_pair_key = ?1"), params![pair_key], room_from_row)
.optional()
}
pub fn room_by_legacy_dm_id(conn: &Connection, legacy_dm_id: i64) -> rusqlite::Result<Option<Room>> {
conn.query_row(&format!("SELECT {ROOM_SELECT_COLUMNS} FROM rooms WHERE legacy_dm_id = ?1"), params![legacy_dm_id], room_from_row)
.optional()
}
pub fn clear_dm_pair_key(conn: &Connection, room_id: &str) -> rusqlite::Result<()> {
conn.execute("UPDATE rooms SET dm_pair_key = NULL WHERE id = ?1", params![room_id])?;
Ok(())
}
#[derive(Debug, Clone, PartialEq)]
pub struct MatrixEvent {
pub stream_id: i64,
pub event_id: String,
pub room_id: String,
pub sender_user_id: i64,
pub event_type: String,
pub state_key: Option<String>,
pub content: String,
pub origin_server_ts: i64,
pub txn_id: Option<String>,
pub redacts: Option<String>,
pub redacted_by: Option<String>,
}
const EVENT_SELECT_COLUMNS: &str =
"stream_id, event_id, room_id, sender_user_id, event_type, state_key, content, origin_server_ts, txn_id, redacts, redacted_by";
const EVENT_SELECT_COLUMNS_ALIASED: &str = "e.stream_id, e.event_id, e.room_id, e.sender_user_id, e.event_type, e.state_key, e.content, e.origin_server_ts, e.txn_id, e.redacts, e.redacted_by";
fn event_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<MatrixEvent> {
Ok(MatrixEvent {
stream_id: row.get(0)?,
event_id: row.get(1)?,
room_id: row.get(2)?,
sender_user_id: row.get(3)?,
event_type: row.get(4)?,
state_key: row.get(5)?,
content: row.get(6)?,
origin_server_ts: row.get(7)?,
txn_id: row.get(8)?,
redacts: row.get(9)?,
redacted_by: row.get(10)?,
})
}
fn collect_events(rows: &mut rusqlite::Rows<'_>) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut out = Vec::new();
while let Some(row) = rows.next()? {
out.push(event_from_row(row)?);
}
Ok(out)
}
#[derive(Debug, Clone, Copy)]
struct TimelineEventRow<'a> {
event_id: &'a str,
room_id: &'a str,
sender_user_id: i64,
event_type: &'a str,
content: &'a str,
origin_server_ts: i64,
txn_id: Option<&'a str>,
}
fn insert_timeline_event_in_tx(tx: &Transaction, row: &TimelineEventRow<'_>) -> Result<MatrixEvent, MatrixStoreError> {
let TimelineEventRow { event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id } = *row;
let stream_id = next_stream_id(tx)?;
tx.execute(
"INSERT INTO events (stream_id, event_id, room_id, sender_user_id, event_type, state_key, content, origin_server_ts, txn_id)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6, ?7, ?8)",
params![stream_id, event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id],
)?;
populate_relations(tx, event_id, room_id, sender_user_id, content)?;
Ok(MatrixEvent {
stream_id,
event_id: event_id.to_string(),
room_id: room_id.to_string(),
sender_user_id,
event_type: event_type.to_string(),
state_key: None,
content: content.to_string(),
origin_server_ts,
txn_id: txn_id.map(str::to_string),
redacts: None,
redacted_by: None,
})
}
pub fn insert_timeline_event(
conn: &mut Connection,
event_id: &str,
room_id: &str,
sender_user_id: i64,
event_type: &str,
content: &str,
origin_server_ts: i64,
) -> Result<MatrixEvent, MatrixStoreError> {
let tx = conn.transaction()?;
let row = TimelineEventRow { event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id: None };
let event = insert_timeline_event_in_tx(&tx, &row)?;
tx.commit()?;
Ok(event)
}
#[derive(Debug, Clone, PartialEq)]
pub enum DedupedWrite {
New(MatrixEvent),
Existing(MatrixEvent),
}
#[allow(clippy::too_many_arguments)]
pub fn insert_timeline_event_deduped(
conn: &mut Connection,
device_id: &str,
txn_id: &str,
event_id: &str,
room_id: &str,
sender_user_id: i64,
event_type: &str,
content: &str,
origin_server_ts: i64,
now: &str,
) -> Result<DedupedWrite, MatrixStoreError> {
let tx = conn.transaction()?;
if let TxnDedupEntry::Seen(existing_event_id) = txn_dedup_lookup(&tx, sender_user_id, device_id, txn_id)? {
let existing_event_id = existing_event_id.ok_or_else(|| MatrixStoreError::UnknownEventId(txn_id.to_string()))?;
let event = get_event(&tx, &existing_event_id)?.ok_or_else(|| MatrixStoreError::UnknownEventId(existing_event_id.clone()))?;
tx.commit()?;
return Ok(DedupedWrite::Existing(event));
}
let row = TimelineEventRow { event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id: Some(txn_id) };
let event = insert_timeline_event_in_tx(&tx, &row)?;
txn_dedup_record(&tx, sender_user_id, device_id, txn_id, Some(event_id), now)?;
tx.commit()?;
Ok(DedupedWrite::New(event))
}
#[allow(clippy::too_many_arguments)]
fn apply_state_event_in_tx(
tx: &Transaction,
event_id: &str,
room_id: &str,
sender_user_id: i64,
event_type: &str,
state_key: &str,
content: &str,
origin_server_ts: i64,
now: &str,
) -> Result<MatrixEvent, MatrixStoreError> {
let stream_id = next_stream_id(tx)?;
tx.execute(
"INSERT INTO events (stream_id, event_id, room_id, sender_user_id, event_type, state_key, content, origin_server_ts, txn_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL)",
params![stream_id, event_id, room_id, sender_user_id, event_type, state_key, content, origin_server_ts],
)?;
tx.execute(
"INSERT INTO current_state (room_id, event_type, state_key, event_id) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(room_id, event_type, state_key) DO UPDATE SET event_id = excluded.event_id",
params![room_id, event_type, state_key, event_id],
)?;
if event_type == "m.room.member" {
refresh_room_member(tx, room_id, state_key, content, now)?;
}
if event_type == "m.room.power_levels" {
refresh_power_levels(tx, room_id, content)?;
}
Ok(MatrixEvent {
stream_id,
event_id: event_id.to_string(),
room_id: room_id.to_string(),
sender_user_id,
event_type: event_type.to_string(),
state_key: Some(state_key.to_string()),
content: content.to_string(),
origin_server_ts,
txn_id: None,
redacts: None,
redacted_by: None,
})
}
#[derive(Debug, Clone, Copy)]
pub struct StateEventWrite<'a> {
pub event_id: &'a str,
pub room_id: &'a str,
pub sender_user_id: i64,
pub event_type: &'a str,
pub state_key: &'a str,
pub content: &'a str,
pub origin_server_ts: i64,
pub now: &'a str,
}
pub fn apply_state_event(conn: &mut Connection, write: &StateEventWrite<'_>) -> Result<MatrixEvent, MatrixStoreError> {
let tx = conn.transaction()?;
let event = apply_state_event_in_tx(
&tx,
write.event_id,
write.room_id,
write.sender_user_id,
write.event_type,
write.state_key,
write.content,
write.origin_server_ts,
write.now,
)?;
tx.commit()?;
Ok(event)
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct DisplaynameRefresh {
pub rooms_updated: usize,
pub affected_user_ids: HashSet<i64>,
}
pub fn refresh_member_displayname(
conn: &mut Connection,
user_id: i64,
displayname: &str,
now: &str,
origin_server_ts: i64,
) -> Result<DisplaynameRefresh, MatrixStoreError> {
let mut outcome = DisplaynameRefresh::default();
if displayname.is_empty() {
return Ok(outcome);
}
let Some(mxid) = mxid_of(conn, user_id)? else {
return Ok(outcome);
};
let tx = conn.transaction()?;
for membership in [Membership::Join, Membership::Invite] {
for room_id in rooms_for_user(&tx, user_id, Some(membership))? {
let Some(current) = current_state_event(&tx, &room_id, "m.room.member", &mxid)? else {
continue;
};
let mut content: serde_json::Value = serde_json::from_str(¤t.content)?;
if content.get("displayname").and_then(|v| v.as_str()) == Some(displayname) {
continue;
}
let Some(fields) = content.as_object_mut() else {
continue;
};
fields.insert("displayname".to_string(), serde_json::Value::String(displayname.to_string()));
let sender_user_id = if membership == Membership::Join { user_id } else { current.sender_user_id };
apply_state_event_in_tx(
&tx,
&new_event_id(),
&room_id,
sender_user_id,
"m.room.member",
&mxid,
&content.to_string(),
origin_server_ts,
now,
)?;
outcome.rooms_updated += 1;
outcome.affected_user_ids.insert(user_id);
for member in room_members(&tx, &room_id, None)? {
if matches!(member.membership, Membership::Join | Membership::Invite) {
outcome.affected_user_ids.insert(member.user_id);
}
}
}
}
tx.commit()?;
Ok(outcome)
}
pub fn matrix_user_ids(conn: &Connection) -> rusqlite::Result<Vec<i64>> {
let mut stmt = conn.prepare("SELECT user_id FROM matrix_users ORDER BY user_id")?;
let rows = stmt.query_map([], |row| row.get(0))?;
rows.collect()
}
#[derive(Debug, Clone, PartialEq)]
pub struct NewStateEvent {
pub event_id: String,
pub sender_user_id: i64,
pub event_type: String,
pub state_key: String,
pub content: String,
}
#[derive(Debug, Clone, Copy)]
pub struct RoomBootstrap<'a> {
pub room_id: &'a str,
pub kind: RoomKind,
pub creator_user_id: i64,
pub created_at: &'a str,
pub is_encrypted: bool,
pub join_rule: JoinRule,
pub history_visibility: HistoryVisibility,
pub dm_pair_key: Option<&'a str>,
pub legacy_dm_id: Option<i64>,
}
pub fn create_room_with_state(
conn: &mut Connection,
bootstrap: RoomBootstrap<'_>,
state_events: &[NewStateEvent],
origin_server_ts: i64,
) -> Result<(Room, Vec<MatrixEvent>), MatrixStoreError> {
let tx = conn.transaction()?;
insert_room_row(
&tx,
bootstrap.room_id,
bootstrap.kind,
bootstrap.creator_user_id,
bootstrap.created_at,
bootstrap.is_encrypted,
bootstrap.join_rule,
bootstrap.history_visibility,
bootstrap.dm_pair_key,
bootstrap.legacy_dm_id,
)?;
let mut applied = Vec::with_capacity(state_events.len());
for event in state_events {
applied.push(apply_state_event_in_tx(
&tx,
&event.event_id,
bootstrap.room_id,
event.sender_user_id,
&event.event_type,
&event.state_key,
&event.content,
origin_server_ts,
bootstrap.created_at,
)?);
}
tx.commit()?;
Ok((
Room {
id: bootstrap.room_id.to_string(),
kind: bootstrap.kind,
room_version: MATRIX_ROOM_VERSION.to_string(),
creator_user_id: bootstrap.creator_user_id,
created_at: bootstrap.created_at.to_string(),
is_encrypted: bootstrap.is_encrypted,
join_rule: bootstrap.join_rule,
history_visibility: bootstrap.history_visibility,
dm_pair_key: bootstrap.dm_pair_key.map(str::to_string),
legacy_dm_id: bootstrap.legacy_dm_id,
},
applied,
))
}
fn refresh_room_member(tx: &Transaction, room_id: &str, state_key: &str, content: &str, now: &str) -> Result<(), MatrixStoreError> {
let value: serde_json::Value = serde_json::from_str(content)?;
let membership_str = value
.get("membership")
.and_then(|v| v.as_str())
.ok_or_else(|| MatrixStoreError::InvalidMembership("missing 'membership' field".to_string()))?;
let membership =
Membership::from_wire_name(membership_str).ok_or_else(|| MatrixStoreError::InvalidMembership(membership_str.to_string()))?;
let user_id = user_id_of(tx, state_key)?.ok_or_else(|| MatrixStoreError::UnknownMxid(state_key.to_string()))?;
tx.execute(
"INSERT INTO room_members (room_id, user_id, membership, power_level, updated_at)
VALUES (?1, ?2, ?3, NULL, ?4)
ON CONFLICT(room_id, user_id) DO UPDATE SET membership = excluded.membership, updated_at = excluded.updated_at",
params![room_id, user_id, membership.as_str(), now],
)?;
Ok(())
}
fn refresh_power_levels(tx: &Transaction, room_id: &str, content: &str) -> Result<(), MatrixStoreError> {
let value: serde_json::Value = serde_json::from_str(content)?;
tx.execute("UPDATE room_members SET power_level = NULL WHERE room_id = ?1", params![room_id])?;
if let Some(users) = value.get("users").and_then(|v| v.as_object()) {
for (mxid, level) in users {
let Some(level) = level.as_i64() else { continue };
let Some(user_id) = user_id_of(tx, mxid)? else { continue };
tx.execute(
"UPDATE room_members SET power_level = ?1 WHERE room_id = ?2 AND user_id = ?3",
params![level, room_id, user_id],
)?;
}
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PowerAction {
Invite,
Kick,
Ban,
Redact,
StateDefault,
}
impl PowerAction {
fn field_and_default(self) -> (&'static str, i64) {
match self {
PowerAction::Invite => ("invite", 50),
PowerAction::Kick => ("kick", 50),
PowerAction::Ban => ("ban", 50),
PowerAction::Redact => ("redact", 50),
PowerAction::StateDefault => ("state_default", 50),
}
}
}
pub fn user_level(power_levels: &serde_json::Value, mxid: &str) -> i64 {
power_levels
.get("users")
.and_then(|users| users.get(mxid))
.and_then(serde_json::Value::as_i64)
.unwrap_or_else(|| power_levels.get("users_default").and_then(serde_json::Value::as_i64).unwrap_or(0))
}
pub fn event_level(power_levels: &serde_json::Value, event_type: &str, is_state: bool) -> i64 {
if let Some(level) = power_levels.get("events").and_then(|events| events.get(event_type)).and_then(serde_json::Value::as_i64) {
return level;
}
let (key, default) = if is_state { ("state_default", 50) } else { ("events_default", 0) };
power_levels.get(key).and_then(serde_json::Value::as_i64).unwrap_or(default)
}
pub fn can(power_levels: &serde_json::Value, action: PowerAction, mxid: &str) -> bool {
let (field, default) = action.field_and_default();
let required = power_levels.get(field).and_then(serde_json::Value::as_i64).unwrap_or(default);
user_level(power_levels, mxid) >= required
}
pub fn can_act_on(power_levels: &serde_json::Value, action: PowerAction, sender_mxid: &str, target_mxid: &str, self_leave: bool) -> bool {
if self_leave && sender_mxid == target_mxid {
return true;
}
can(power_levels, action, sender_mxid) && user_level(power_levels, sender_mxid) > user_level(power_levels, target_mxid)
}
fn power_levels_scalar_default(key: &str) -> i64 {
match key {
"events_default" | "users_default" => 0,
_ => 50,
}
}
const POWER_LEVELS_SCALAR_KEYS: [&str; 7] = ["ban", "kick", "redact", "invite", "state_default", "events_default", "users_default"];
fn reject_if_either_side_exceeds(old_value: Option<i64>, new_value: Option<i64>, sender_level: i64) -> Result<(), &'static str> {
if old_value != new_value && (old_value.is_some_and(|v| v > sender_level) || new_value.is_some_and(|v| v > sender_level)) {
return Err("cannot change a power-level field at or above your own level");
}
Ok(())
}
pub fn validate_power_levels_change(old: &serde_json::Value, new: &serde_json::Value, sender_mxid: &str) -> Result<(), &'static str> {
let sender_level = user_level(old, sender_mxid);
for key in POWER_LEVELS_SCALAR_KEYS {
let default = power_levels_scalar_default(key);
let old_value = old.get(key).and_then(serde_json::Value::as_i64).unwrap_or(default);
let new_value = new.get(key).and_then(serde_json::Value::as_i64).unwrap_or(default);
reject_if_either_side_exceeds(Some(old_value), Some(new_value), sender_level)?;
}
let old_events = old.get("events").and_then(serde_json::Value::as_object);
let new_events = new.get("events").and_then(serde_json::Value::as_object);
let mut event_type_keys: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
if let Some(map) = old_events {
event_type_keys.extend(map.keys().map(String::as_str));
}
if let Some(map) = new_events {
event_type_keys.extend(map.keys().map(String::as_str));
}
for event_type in event_type_keys {
let old_value = old_events.and_then(|m| m.get(event_type)).and_then(serde_json::Value::as_i64);
let new_value = new_events.and_then(|m| m.get(event_type)).and_then(serde_json::Value::as_i64);
reject_if_either_side_exceeds(old_value, new_value, sender_level)?;
}
let old_notif_room = old.get("notifications").and_then(|v| v.get("room")).and_then(serde_json::Value::as_i64);
let new_notif_room = new.get("notifications").and_then(|v| v.get("room")).and_then(serde_json::Value::as_i64);
reject_if_either_side_exceeds(old_notif_room, new_notif_room, sender_level)?;
let old_users = old.get("users").and_then(serde_json::Value::as_object);
let new_users = new.get("users").and_then(serde_json::Value::as_object);
let mut user_keys: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
if let Some(map) = old_users {
user_keys.extend(map.keys().map(String::as_str));
}
if let Some(map) = new_users {
user_keys.extend(map.keys().map(String::as_str));
}
for target_mxid in user_keys {
let old_effective = user_level(old, target_mxid);
let new_effective = user_level(new, target_mxid);
if old_effective == new_effective {
continue;
}
if target_mxid != sender_mxid && old_effective >= sender_level {
return Err("cannot change the level of a user at or above your own level");
}
if new_effective > sender_level {
return Err("cannot set a user's level above your own");
}
}
Ok(())
}
pub fn stripped_state_json(conn: &Connection, event: &MatrixEvent) -> Result<serde_json::Value, MatrixStoreError> {
let sender = mxid_of(conn, event.sender_user_id)?.unwrap_or_default();
let content: serde_json::Value = serde_json::from_str(&event.content)?;
Ok(serde_json::json!({
"content": content,
"state_key": event.state_key.clone().unwrap_or_default(),
"type": event.event_type,
"sender": sender,
}))
}
pub fn stripped_invite_state(conn: &Connection, room_id: &str, inviter_user_id: i64) -> Result<Vec<serde_json::Value>, MatrixStoreError> {
let mut out = Vec::new();
for event_type in ["m.room.create", "m.room.join_rules", "m.room.encryption", "m.room.name"] {
if let Some(event) = current_state_event(conn, room_id, event_type, "")? {
out.push(stripped_state_json(conn, &event)?);
}
}
if let Some(inviter_mxid) = mxid_of(conn, inviter_user_id)? {
if let Some(event) = current_state_event(conn, room_id, "m.room.member", &inviter_mxid)? {
out.push(stripped_state_json(conn, &event)?);
}
}
Ok(out)
}
pub fn get_event(conn: &Connection, event_id: &str) -> rusqlite::Result<Option<MatrixEvent>> {
let closed = conn
.query_row(&format!("SELECT {EVENT_SELECT_COLUMNS} FROM events WHERE event_id = ?1"), params![event_id], event_from_row)
.optional()?;
if closed.is_some() {
return Ok(closed);
}
match crate::public_channels::get_event(conn, event_id) {
Ok(found) => Ok(found),
Err(rusqlite::Error::SqliteFailure(_, Some(msg))) if msg.contains("no such table") => Ok(None),
Err(e) => Err(e),
}
}
pub fn current_state_event(conn: &Connection, room_id: &str, event_type: &str, state_key: &str) -> rusqlite::Result<Option<MatrixEvent>> {
conn.query_row(
&format!(
"SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM current_state cs JOIN events e ON e.event_id = cs.event_id
WHERE cs.room_id = ?1 AND cs.event_type = ?2 AND cs.state_key = ?3"
),
params![room_id, event_type, state_key],
event_from_row,
)
.optional()
}
pub fn current_state_all(conn: &Connection, room_id: &str) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM current_state cs JOIN events e ON e.event_id = cs.event_id WHERE cs.room_id = ?1"
))?;
let mut rows = stmt.query(params![room_id])?;
collect_events(&mut rows)
}
pub fn state_events_of_type_at(conn: &Connection, room_id: &str, event_type: &str, at_stream_id: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM events e
WHERE e.room_id = ?1 AND e.event_type = ?2 AND e.state_key IS NOT NULL AND e.stream_id <= ?3
AND e.stream_id = (
SELECT MAX(stream_id) FROM events e2
WHERE e2.room_id = e.room_id AND e2.event_type = e.event_type AND e2.state_key = e.state_key AND e2.stream_id <= ?3
)"
))?;
let mut rows = stmt.query(params![room_id, event_type, at_stream_id])?;
collect_events(&mut rows)
}
pub fn member_state_changed_in_window(conn: &Connection, room_id: &str, since_exclusive: i64, upto_inclusive: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM events e
WHERE e.room_id = ?1 AND e.event_type = 'm.room.member' AND e.state_key IS NOT NULL
AND e.stream_id = (
SELECT MAX(e2.stream_id) FROM events e2
WHERE e2.room_id = e.room_id AND e2.event_type = 'm.room.member' AND e2.state_key = e.state_key AND e2.stream_id <= ?3
)
AND e.stream_id > ?2"
))?;
let mut rows = stmt.query(params![room_id, since_exclusive, upto_inclusive])?;
collect_events(&mut rows)
}
pub fn non_member_state_changed_in_window(conn: &Connection, room_id: &str, since_exclusive: i64, upto_inclusive: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM events e
WHERE e.room_id = ?1 AND e.event_type != 'm.room.member' AND e.state_key IS NOT NULL
AND e.stream_id = (
SELECT MAX(e2.stream_id) FROM events e2
WHERE e2.room_id = e.room_id AND e2.event_type = e.event_type AND e2.state_key = e.state_key AND e2.stream_id <= ?3
)
AND e.stream_id > ?2"
))?;
let mut rows = stmt.query(params![room_id, since_exclusive, upto_inclusive])?;
collect_events(&mut rows)
}
pub fn events_in_room_after(conn: &Connection, room_id: &str, since_stream: i64, limit: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS} FROM events WHERE room_id = ?1 AND stream_id > ?2 ORDER BY stream_id ASC LIMIT ?3"
))?;
let mut rows = stmt.query(params![room_id, since_stream, limit])?;
let mut out = collect_events(&mut rows)?;
if crate::public_channels::is_public_room(conn, room_id)? {
out.extend(crate::public_channels::events_after(conn, room_id, since_stream, limit)?);
out.sort_by_key(|e| e.stream_id);
out.truncate(limit.max(0) as usize);
}
Ok(out)
}
pub fn events_in_room_before(conn: &Connection, room_id: &str, before_stream: i64, limit: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut stmt = conn.prepare(&format!(
"SELECT {EVENT_SELECT_COLUMNS} FROM events WHERE room_id = ?1 AND stream_id < ?2 ORDER BY stream_id DESC LIMIT ?3"
))?;
let mut rows = stmt.query(params![room_id, before_stream, limit])?;
let mut out = collect_events(&mut rows)?;
if crate::public_channels::is_public_room(conn, room_id)? {
out.extend(crate::public_channels::events_before(conn, room_id, before_stream, limit)?);
out.sort_by_key(|e| std::cmp::Reverse(e.stream_id));
out.truncate(limit.max(0) as usize);
}
Ok(out)
}
fn populate_relations(tx: &Transaction, event_id: &str, room_id: &str, sender_user_id: i64, content: &str) -> Result<(), MatrixStoreError> {
let Ok(value) = serde_json::from_str::<serde_json::Value>(content) else {
return Ok(());
};
let Some(relates_to) = value.get("m.relates_to") else {
return Ok(());
};
let (rel_type, target_id, agg_key): (String, String, Option<String>) = if let Some(reply) = relates_to.get("m.in_reply_to") {
match reply.get("event_id").and_then(|v| v.as_str()) {
Some(target) => ("m.in_reply_to".to_string(), target.to_string(), None),
None => return Ok(()),
}
} else {
let rel_type = relates_to.get("rel_type").and_then(|v| v.as_str());
let target = relates_to.get("event_id").and_then(|v| v.as_str());
match (rel_type, target) {
(Some(rt), Some(target)) => {
let key = relates_to.get("key").and_then(|v| v.as_str()).map(str::to_string);
(rt.to_string(), target.to_string(), key)
}
_ => return Ok(()),
}
};
let target_room: Option<String> = tx
.query_row("SELECT room_id FROM events WHERE event_id = ?1", params![target_id], |row| row.get(0))
.optional()?;
match target_room {
Some(ref found_room) if found_room == room_id => {}
_ => return Err(MatrixStoreError::InvalidRelationTarget(target_id)),
}
if rel_type == "m.annotation" {
let duplicate: Option<i64> = tx
.query_row(
"SELECT 1 FROM relations r JOIN events e ON e.event_id = r.event_id
WHERE r.target_id = ?1 AND r.rel_type = 'm.annotation' AND r.agg_key IS ?2 AND e.sender_user_id = ?3
AND e.redacted_by IS NULL
LIMIT 1",
params![target_id, agg_key, sender_user_id],
|row| row.get(0),
)
.optional()?;
if duplicate.is_some() {
return Err(MatrixStoreError::DuplicateAnnotation);
}
}
tx.execute(
"INSERT INTO relations (event_id, room_id, rel_type, target_id, agg_key) VALUES (?1, ?2, ?3, ?4, ?5)",
params![event_id, room_id, rel_type, target_id, agg_key],
)?;
Ok(())
}
pub fn relations_of(
conn: &Connection,
target_event_id: &str,
rel_type: Option<&str>,
event_type: Option<&str>,
before_stream: i64,
limit: i64,
) -> rusqlite::Result<Vec<MatrixEvent>> {
let mut sql = format!("SELECT {EVENT_SELECT_COLUMNS_ALIASED} FROM relations r JOIN events e ON e.event_id = r.event_id WHERE r.target_id = ? AND e.stream_id < ?");
let mut values: Vec<&dyn rusqlite::ToSql> = vec![&target_event_id, &before_stream];
if let Some(rt) = &rel_type {
sql.push_str(" AND r.rel_type = ?");
values.push(rt);
}
if let Some(et) = &event_type {
sql.push_str(" AND e.event_type = ?");
values.push(et);
}
sql.push_str(" ORDER BY e.stream_id DESC LIMIT ?");
values.push(&limit);
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(values.as_slice())?;
collect_events(&mut rows)
}
fn redact_content_per_v11(event_type: &str, content: &str) -> Result<String, MatrixStoreError> {
let value: serde_json::Value = serde_json::from_str(content)?;
let obj = value.as_object().cloned().unwrap_or_default();
let mut kept = serde_json::Map::new();
match event_type {
"m.room.create" => kept = obj,
"m.room.member" => {
for key in ["membership", "join_authorised_via_users_server"] {
if let Some(v) = obj.get(key) {
kept.insert(key.to_string(), v.clone());
}
}
if let Some(signed) = obj.get("third_party_invite").and_then(|v| v.get("signed")) {
let mut third_party_invite = serde_json::Map::new();
third_party_invite.insert("signed".to_string(), signed.clone());
kept.insert("third_party_invite".to_string(), serde_json::Value::Object(third_party_invite));
}
}
"m.room.join_rules" => {
for key in ["join_rule", "allow"] {
if let Some(v) = obj.get(key) {
kept.insert(key.to_string(), v.clone());
}
}
}
"m.room.power_levels" => {
for key in [
"ban",
"events",
"events_default",
"invite",
"kick",
"redact",
"state_default",
"users",
"users_default",
] {
if let Some(v) = obj.get(key) {
kept.insert(key.to_string(), v.clone());
}
}
}
"m.room.history_visibility" => {
if let Some(v) = obj.get("history_visibility") {
kept.insert("history_visibility".to_string(), v.clone());
}
}
_ => {}
}
Ok(serde_json::Value::Object(kept).to_string())
}
fn redact_event_in_tx(
tx: &Transaction,
room_id: &str,
target_event_id: &str,
redaction_event_id: &str,
sender_user_id: i64,
reason: Option<&str>,
origin_server_ts: i64,
) -> Result<MatrixEvent, MatrixStoreError> {
let target: Option<(String, String, String)> = tx
.query_row(
"SELECT event_type, content, room_id FROM events WHERE event_id = ?1",
params![target_event_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.optional()?;
let (target_type, target_content, target_room) = target.ok_or_else(|| MatrixStoreError::UnknownEventId(target_event_id.to_string()))?;
if target_room != room_id {
return Err(MatrixStoreError::WrongRoom(target_event_id.to_string()));
}
if target_type == "m.room.create" || target_type == "m.room.encryption" {
return Err(MatrixStoreError::UnredactableEvent(target_type));
}
let stream_id = next_stream_id(tx)?;
let mut redaction_content = serde_json::json!({ "redacts": target_event_id });
if let Some(r) = reason {
redaction_content["reason"] = serde_json::Value::String(r.to_string());
}
let redaction_content = redaction_content.to_string();
tx.execute(
"INSERT INTO events (stream_id, event_id, room_id, sender_user_id, event_type, state_key, content, origin_server_ts, txn_id, redacts)
VALUES (?1, ?2, ?3, ?4, 'm.room.redaction', NULL, ?5, ?6, NULL, ?7)",
params![stream_id, redaction_event_id, room_id, sender_user_id, redaction_content, origin_server_ts, target_event_id],
)?;
let stripped_content = redact_content_per_v11(&target_type, &target_content)?;
tx.execute(
"UPDATE events SET content = ?1, redacted_by = ?2 WHERE event_id = ?3",
params![stripped_content, redaction_event_id, target_event_id],
)?;
Ok(MatrixEvent {
stream_id,
event_id: redaction_event_id.to_string(),
room_id: room_id.to_string(),
sender_user_id,
event_type: "m.room.redaction".to_string(),
state_key: None,
content: redaction_content,
origin_server_ts,
txn_id: None,
redacts: Some(target_event_id.to_string()),
redacted_by: None,
})
}
pub fn redact_event(
conn: &mut Connection,
room_id: &str,
target_event_id: &str,
redaction_event_id: &str,
sender_user_id: i64,
reason: Option<&str>,
origin_server_ts: i64,
) -> Result<MatrixEvent, MatrixStoreError> {
let tx = conn.transaction()?;
let event = redact_event_in_tx(&tx, room_id, target_event_id, redaction_event_id, sender_user_id, reason, origin_server_ts)?;
tx.commit()?;
Ok(event)
}
#[derive(Debug, Clone, Copy)]
pub struct Redaction<'a> {
pub room_id: &'a str,
pub target_event_id: &'a str,
pub redaction_event_id: &'a str,
pub sender_user_id: i64,
pub reason: Option<&'a str>,
pub origin_server_ts: i64,
}
pub fn redact_event_marked(
conn: &mut Connection,
redaction: &Redaction<'_>,
extra_content: &serde_json::Value,
) -> Result<MatrixEvent, MatrixStoreError> {
let Redaction { room_id, target_event_id, redaction_event_id, sender_user_id, reason, origin_server_ts } = *redaction;
let tx = conn.transaction()?;
let mut event = redact_event_in_tx(&tx, room_id, target_event_id, redaction_event_id, sender_user_id, reason, origin_server_ts)?;
let mut content: serde_json::Value = serde_json::from_str(&event.content)?;
if let (Some(content_obj), Some(extra_obj)) = (content.as_object_mut(), extra_content.as_object()) {
for (key, value) in extra_obj {
content_obj.insert(key.clone(), value.clone());
}
}
let content_str = content.to_string();
tx.execute("UPDATE events SET content = ?1 WHERE event_id = ?2", params![content_str, redaction_event_id])?;
event.content = content_str;
tx.commit()?;
Ok(event)
}
#[allow(clippy::too_many_arguments)]
pub fn redact_event_deduped(
conn: &mut Connection,
device_id: &str,
txn_id: &str,
room_id: &str,
target_event_id: &str,
redaction_event_id: &str,
sender_user_id: i64,
reason: Option<&str>,
origin_server_ts: i64,
now: &str,
) -> Result<DedupedWrite, MatrixStoreError> {
let tx = conn.transaction()?;
if let TxnDedupEntry::Seen(existing_event_id) = txn_dedup_lookup(&tx, sender_user_id, device_id, txn_id)? {
let existing_event_id = existing_event_id.ok_or_else(|| MatrixStoreError::UnknownEventId(txn_id.to_string()))?;
let event = get_event(&tx, &existing_event_id)?.ok_or_else(|| MatrixStoreError::UnknownEventId(existing_event_id.clone()))?;
tx.commit()?;
return Ok(DedupedWrite::Existing(event));
}
let event = redact_event_in_tx(&tx, room_id, target_event_id, redaction_event_id, sender_user_id, reason, origin_server_ts)?;
txn_dedup_record(&tx, sender_user_id, device_id, txn_id, Some(redaction_event_id), now)?;
tx.commit()?;
Ok(DedupedWrite::New(event))
}
#[derive(Debug, Clone, PartialEq)]
pub struct RoomMember {
pub room_id: String,
pub user_id: i64,
pub membership: Membership,
pub power_level: Option<i64>,
pub updated_at: String,
}
fn room_member_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RoomMember> {
let membership_raw: String = row.get(2)?;
Ok(RoomMember {
room_id: row.get(0)?,
user_id: row.get(1)?,
membership: decode_enum(2, "membership", &membership_raw, Membership::from_wire_name)?,
power_level: row.get(3)?,
updated_at: row.get(4)?,
})
}
const ROOM_MEMBER_SELECT_COLUMNS: &str = "room_id, user_id, membership, power_level, updated_at";
pub fn room_members(conn: &Connection, room_id: &str, membership: Option<Membership>) -> rusqlite::Result<Vec<RoomMember>> {
match membership {
Some(m) => {
let mut stmt = conn.prepare(&format!(
"SELECT {ROOM_MEMBER_SELECT_COLUMNS} FROM room_members WHERE room_id = ?1 AND membership = ?2"
))?;
let rows = stmt.query_map(params![room_id, m.as_str()], room_member_from_row)?;
rows.collect()
}
None => {
let mut stmt = conn.prepare(&format!("SELECT {ROOM_MEMBER_SELECT_COLUMNS} FROM room_members WHERE room_id = ?1"))?;
let rows = stmt.query_map(params![room_id], room_member_from_row)?;
rows.collect()
}
}
}
pub fn room_member(conn: &Connection, room_id: &str, user_id: i64) -> rusqlite::Result<Option<RoomMember>> {
conn.query_row(
&format!("SELECT {ROOM_MEMBER_SELECT_COLUMNS} FROM room_members WHERE room_id = ?1 AND user_id = ?2"),
params![room_id, user_id],
room_member_from_row,
)
.optional()
}
pub fn forget_membership(conn: &Connection, room_id: &str, user_id: i64) -> rusqlite::Result<usize> {
conn.execute(
"DELETE FROM room_members WHERE room_id = ?1 AND user_id = ?2 AND membership = 'leave'",
params![room_id, user_id],
)
}
pub fn room_heroes(conn: &Connection, room_id: &str, exclude_user_id: i64, limit: i64) -> rusqlite::Result<Vec<i64>> {
let mut stmt = conn.prepare(
"SELECT user_id FROM room_members
WHERE room_id = ?1 AND user_id != ?2 AND membership IN ('join', 'invite')
ORDER BY updated_at ASC LIMIT ?3",
)?;
let rows = stmt.query_map(params![room_id, exclude_user_id, limit], |row| row.get(0))?;
rows.collect()
}
pub fn rooms_for_user(conn: &Connection, user_id: i64, membership: Option<Membership>) -> rusqlite::Result<Vec<String>> {
match membership {
Some(m) => {
let mut stmt = conn.prepare("SELECT room_id FROM room_members WHERE user_id = ?1 AND membership = ?2")?;
let rows = stmt.query_map(params![user_id, m.as_str()], |row| row.get(0))?;
rows.collect()
}
None => {
let mut stmt = conn.prepare("SELECT room_id FROM room_members WHERE user_id = ?1")?;
let rows = stmt.query_map(params![user_id], |row| row.get(0))?;
rows.collect()
}
}
}
pub fn rooms_changed_in_window(
conn: &Connection,
room_ids: &[String],
caller_user_id: i64,
since_exclusive: i64,
upto_inclusive: i64,
) -> rusqlite::Result<HashSet<String>> {
let mut changed = HashSet::new();
if room_ids.is_empty() {
return Ok(changed);
}
let placeholders = vec!["?"; room_ids.len()].join(",");
for table in ["events", "receipts", "pub_events"] {
let sql = format!("SELECT DISTINCT room_id FROM {table} WHERE stream_id > ? AND stream_id <= ? AND room_id IN ({placeholders})");
let mut stmt = conn.prepare(&sql)?;
let mut bound: Vec<&dyn rusqlite::ToSql> = vec![&since_exclusive, &upto_inclusive];
for room_id in room_ids {
bound.push(room_id as &dyn rusqlite::ToSql);
}
let mut rows = stmt.query(bound.as_slice())?;
while let Some(row) = rows.next()? {
changed.insert(row.get::<_, String>(0)?);
}
}
let room_id_set: HashSet<&str> = room_ids.iter().map(String::as_str).collect();
let mut stmt = conn.prepare("SELECT DISTINCT room_id FROM account_data WHERE user_id = ?1 AND stream_id > ?2 AND stream_id <= ?3")?;
let mut rows = stmt.query(params![caller_user_id, since_exclusive, upto_inclusive])?;
while let Some(row) = rows.next()? {
let room_id: String = row.get(0)?;
if room_id_set.contains(room_id.as_str()) {
changed.insert(room_id);
}
}
Ok(changed)
}
pub fn membership_at(conn: &Connection, room_id: &str, mxid: &str, at_stream_id: i64) -> rusqlite::Result<Option<Membership>> {
let content: Option<String> = conn
.query_row(
"SELECT content FROM events
WHERE room_id = ?1 AND event_type = 'm.room.member' AND state_key = ?2 AND stream_id <= ?3
ORDER BY stream_id DESC LIMIT 1",
params![room_id, mxid, at_stream_id],
|row| row.get(0),
)
.optional()?;
Ok(content
.and_then(|raw| serde_json::from_str::<serde_json::Value>(&raw).ok())
.and_then(|value| value.get("membership").and_then(|m| m.as_str()).and_then(Membership::from_wire_name)))
}
pub fn rooms_with_member_events_in_window(conn: &Connection, room_ids: &[String], from_exclusive: i64, to_inclusive: i64) -> rusqlite::Result<Vec<String>> {
if room_ids.is_empty() {
return Ok(Vec::new());
}
let placeholders = vec!["?"; room_ids.len()].join(",");
let sql = format!(
"SELECT DISTINCT room_id FROM events
WHERE event_type = 'm.room.member' AND stream_id > ? AND stream_id <= ? AND room_id IN ({placeholders})"
);
let mut stmt = conn.prepare(&sql)?;
let mut bound: Vec<&dyn rusqlite::ToSql> = vec![&from_exclusive, &to_inclusive];
for room_id in room_ids {
bound.push(room_id as &dyn rusqlite::ToSql);
}
let rows = stmt.query_map(bound.as_slice(), |row| row.get(0))?;
rows.collect()
}
pub fn member_state_keys_in_window(conn: &Connection, room_id: &str, from_exclusive: i64, to_inclusive: i64) -> rusqlite::Result<Vec<String>> {
let mut stmt = conn.prepare(
"SELECT DISTINCT state_key FROM events
WHERE room_id = ?1 AND event_type = 'm.room.member' AND state_key IS NOT NULL AND stream_id > ?2 AND stream_id <= ?3",
)?;
let rows = stmt.query_map(params![room_id, from_exclusive, to_inclusive], |row| row.get(0))?;
rows.collect()
}
pub fn user_ids_with_leave_transition_in_rooms(
conn: &Connection,
room_ids: &[String],
from_exclusive: i64,
to_inclusive: i64,
) -> rusqlite::Result<Vec<i64>> {
if room_ids.is_empty() {
return Ok(Vec::new());
}
let placeholders = vec!["?"; room_ids.len()].join(",");
let sql = format!(
"SELECT DISTINCT e.state_key, e.content FROM events e
WHERE e.event_type = 'm.room.member' AND e.state_key IS NOT NULL
AND e.stream_id > ? AND e.stream_id <= ?
AND e.room_id IN ({placeholders})"
);
let mut stmt = conn.prepare(&sql)?;
let mut bound: Vec<&dyn rusqlite::ToSql> = vec![&from_exclusive, &to_inclusive];
for room_id in room_ids {
bound.push(room_id as &dyn rusqlite::ToSql);
}
let mut rows = stmt.query(bound.as_slice())?;
let mut mxids = Vec::new();
while let Some(row) = rows.next()? {
let mxid: String = row.get(0)?;
let content: String = row.get(1)?;
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&content) {
if matches!(value.get("membership").and_then(|m| m.as_str()), Some("leave") | Some("ban")) {
mxids.push(mxid);
}
}
}
let mut user_ids = Vec::new();
for mxid in mxids {
if let Some(user_id) = user_id_of(conn, &mxid)? {
user_ids.push(user_id);
}
}
Ok(user_ids)
}
pub fn private_room_messages_sent_since(conn: &Connection, sender_user_id: i64, since_ms: i64) -> rusqlite::Result<i64> {
conn.query_row(
"SELECT COUNT(*) FROM events e JOIN rooms r ON r.id = e.room_id
WHERE e.sender_user_id = ?1 AND e.state_key IS NULL AND e.event_type != 'm.room.redaction'
AND e.origin_server_ts >= ?2 AND r.join_rule = 'invite'",
params![sender_user_id, since_ms],
|row| row.get(0),
)
}
pub fn txn_dedup_count_since(conn: &Connection, user_id: i64, device_id: &str, since: &str) -> rusqlite::Result<i64> {
conn.query_row(
"SELECT COUNT(*) FROM txn_dedup WHERE user_id = ?1 AND device_id = ?2 AND created_at >= ?3",
params![user_id, device_id, since],
|row| row.get(0),
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HistoryWindow {
Nothing,
All,
UpTo(i64),
}
impl HistoryWindow {
pub fn contains(self, stream_id: i64) -> bool {
match self {
HistoryWindow::Nothing => false,
HistoryWindow::All => true,
HistoryWindow::UpTo(upto) => stream_id <= upto,
}
}
}
pub fn visible_upper_bound(conn: &Connection, room: &Room, caller_user_id: i64) -> Result<HistoryWindow, MatrixStoreError> {
if room.history_visibility == HistoryVisibility::WorldReadable {
return Ok(HistoryWindow::All);
}
let Some(mxid) = mxid_of(conn, caller_user_id)? else {
return Ok(HistoryWindow::Nothing);
};
let Some(member_event) = current_state_event(conn, &room.id, "m.room.member", &mxid)? else {
return Ok(HistoryWindow::Nothing);
};
let content: serde_json::Value = serde_json::from_str(&member_event.content)?;
match content.get("membership").and_then(|v| v.as_str()) {
Some("join") => Ok(HistoryWindow::All),
Some("leave") | Some("ban") => Ok(HistoryWindow::UpTo(member_event.stream_id)),
_ => Ok(HistoryWindow::Nothing),
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ReceiptRow {
pub room_id: String,
pub user_id: i64,
pub receipt_type: ReceiptType,
pub event_id: String,
pub ts: i64,
pub stream_id: i64,
}
fn receipt_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ReceiptRow> {
let receipt_type_raw: String = row.get(2)?;
Ok(ReceiptRow {
room_id: row.get(0)?,
user_id: row.get(1)?,
receipt_type: decode_enum(2, "receipt_type", &receipt_type_raw, ReceiptType::from_wire_name)?,
event_id: row.get(3)?,
ts: row.get(4)?,
stream_id: row.get(5)?,
})
}
const RECEIPT_SELECT_COLUMNS: &str = "room_id, user_id, receipt_type, event_id, ts, stream_id";
pub fn get_receipt(conn: &Connection, room_id: &str, user_id: i64, receipt_type: ReceiptType) -> rusqlite::Result<Option<ReceiptRow>> {
conn.query_row(
&format!("SELECT {RECEIPT_SELECT_COLUMNS} FROM receipts WHERE room_id = ?1 AND user_id = ?2 AND receipt_type = ?3"),
params![room_id, user_id, receipt_type.as_str()],
receipt_from_row,
)
.optional()
}
pub fn receipts_changed_in_room(conn: &Connection, room_id: &str, since_exclusive: i64, upto_inclusive: i64) -> rusqlite::Result<Vec<ReceiptRow>> {
let mut stmt = conn.prepare(&format!(
"SELECT {RECEIPT_SELECT_COLUMNS} FROM receipts WHERE room_id = ?1 AND stream_id > ?2 AND stream_id <= ?3"
))?;
let rows = stmt.query_map(params![room_id, since_exclusive, upto_inclusive], receipt_from_row)?;
rows.collect()
}
pub fn upsert_receipt(
conn: &mut Connection,
room_id: &str,
user_id: i64,
receipt_type: ReceiptType,
event_id: &str,
ts: i64,
) -> Result<i64, MatrixStoreError> {
let tx = conn.transaction()?;
let stream_id = upsert_receipt_in_tx(&tx, room_id, user_id, receipt_type, event_id, ts)?;
tx.commit()?;
Ok(stream_id)
}
fn upsert_receipt_in_tx(
tx: &Transaction,
room_id: &str,
user_id: i64,
receipt_type: ReceiptType,
event_id: &str,
ts: i64,
) -> Result<i64, MatrixStoreError> {
let target: Option<(i64, String)> = tx
.query_row("SELECT stream_id, room_id FROM events WHERE event_id = ?1", params![event_id], |row| {
Ok((row.get(0)?, row.get(1)?))
})
.optional()?;
let Some((target_position, target_room)) = target else {
if let Some(public) = crate::public_channels::get_event(tx, event_id)? {
if public.room_id != room_id {
return Err(MatrixStoreError::WrongRoom(event_id.to_string()));
}
return Ok(public.stream_id);
}
return Err(MatrixStoreError::UnknownEventId(event_id.to_string()));
};
if target_room != room_id {
return Err(MatrixStoreError::WrongRoom(event_id.to_string()));
}
let existing: Option<(String, i64)> = tx
.query_row(
"SELECT event_id, stream_id FROM receipts WHERE room_id = ?1 AND user_id = ?2 AND receipt_type = ?3",
params![room_id, user_id, receipt_type.as_str()],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((existing_event_id, existing_stream_id)) = &existing {
let existing_position: i64 =
tx.query_row("SELECT stream_id FROM events WHERE event_id = ?1", params![existing_event_id], |row| row.get(0))?;
if target_position <= existing_position {
return Ok(*existing_stream_id);
}
}
let stream_id = next_stream_id(tx)?;
tx.execute(
"INSERT INTO receipts (room_id, user_id, receipt_type, event_id, ts, stream_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(room_id, user_id, receipt_type) DO UPDATE SET
event_id = excluded.event_id, ts = excluded.ts, stream_id = excluded.stream_id",
params![room_id, user_id, receipt_type.as_str(), event_id, ts, stream_id],
)?;
Ok(stream_id)
}
const NOTIFICATION_MESSAGE_TYPES_SQL: &str = "('m.room.message', 'm.room.encrypted', 'org.example.legacy_dm')";
pub fn notification_count(conn: &Connection, room: &Room, user_id: i64) -> Result<i64, MatrixStoreError> {
if room.kind == RoomKind::Channel && !room.is_encrypted {
return Ok(0); }
let upper_bound = match visible_upper_bound(conn, room, user_id)? {
HistoryWindow::Nothing => return Ok(0),
HistoryWindow::All => i64::MAX,
HistoryWindow::UpTo(upper) => upper,
};
let mut after_stream_id: i64 = 0;
for receipt_type in [ReceiptType::Read, ReceiptType::ReadPrivate] {
if let Some(receipt) = get_receipt(conn, &room.id, user_id, receipt_type)? {
if let Some(target) = get_event(conn, &receipt.event_id)? {
after_stream_id = after_stream_id.max(target.stream_id);
}
}
}
let count = conn.query_row(
&format!(
"SELECT COUNT(*) FROM events
WHERE room_id = ?1 AND state_key IS NULL AND sender_user_id != ?2
AND stream_id > ?3 AND stream_id <= ?4
AND event_type IN {NOTIFICATION_MESSAGE_TYPES_SQL}"
),
params![room.id, user_id, after_stream_id, upper_bound],
|row| row.get(0),
)?;
Ok(count)
}
pub const GLOBAL_ACCOUNT_DATA_ROOM: &str = "";
#[derive(Debug, Clone, PartialEq)]
pub struct AccountDataRow {
pub user_id: i64,
pub room_id: String,
pub data_type: String,
pub content: String,
pub stream_id: i64,
}
fn account_data_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<AccountDataRow> {
Ok(AccountDataRow {
user_id: row.get(0)?,
room_id: row.get(1)?,
data_type: row.get(2)?,
content: row.get(3)?,
stream_id: row.get(4)?,
})
}
const ACCOUNT_DATA_SELECT_COLUMNS: &str = "user_id, room_id, data_type, content, stream_id";
pub fn upsert_account_data(conn: &mut Connection, user_id: i64, room_id: &str, data_type: &str, content: &str) -> Result<i64, MatrixStoreError> {
let tx = conn.transaction()?;
let stream_id = next_stream_id(&tx)?;
tx.execute(
"INSERT INTO account_data (user_id, room_id, data_type, content, stream_id) VALUES (?1, ?2, ?3, ?4, ?5)
ON CONFLICT(user_id, room_id, data_type) DO UPDATE SET content = excluded.content, stream_id = excluded.stream_id",
params![user_id, room_id, data_type, content, stream_id],
)?;
tx.commit()?;
Ok(stream_id)
}
pub fn get_account_data(conn: &Connection, user_id: i64, room_id: &str, data_type: &str) -> rusqlite::Result<Option<AccountDataRow>> {
conn.query_row(
&format!("SELECT {ACCOUNT_DATA_SELECT_COLUMNS} FROM account_data WHERE user_id = ?1 AND room_id = ?2 AND data_type = ?3"),
params![user_id, room_id, data_type],
account_data_from_row,
)
.optional()
}
pub fn account_data_since(conn: &Connection, user_id: i64, room_id: &str, since_stream: i64) -> rusqlite::Result<Vec<AccountDataRow>> {
let mut stmt = conn.prepare(&format!(
"SELECT {ACCOUNT_DATA_SELECT_COLUMNS} FROM account_data WHERE user_id = ?1 AND room_id = ?2 AND stream_id > ?3 ORDER BY stream_id ASC"
))?;
let rows = stmt.query_map(params![user_id, room_id, since_stream], account_data_from_row)?;
rows.collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TxnDedupEntry {
NotSeen,
Seen(Option<String>),
}
pub fn txn_dedup_lookup(conn: &Connection, user_id: i64, device_id: &str, txn_id: &str) -> rusqlite::Result<TxnDedupEntry> {
let found: Option<Option<String>> = conn
.query_row(
"SELECT event_id FROM txn_dedup WHERE user_id = ?1 AND device_id = ?2 AND txn_id = ?3",
params![user_id, device_id, txn_id],
|row| row.get(0),
)
.optional()?;
Ok(match found {
None => TxnDedupEntry::NotSeen,
Some(event_id) => TxnDedupEntry::Seen(event_id),
})
}
pub fn txn_dedup_record(conn: &Connection, user_id: i64, device_id: &str, txn_id: &str, event_id: Option<&str>, now: &str) -> rusqlite::Result<()> {
conn.execute(
"INSERT INTO txn_dedup (user_id, device_id, txn_id, event_id, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
params![user_id, device_id, txn_id, event_id, now],
)?;
Ok(())
}
pub fn txn_id_for_event(conn: &Connection, user_id: i64, device_id: &str, event_id: &str) -> rusqlite::Result<Option<String>> {
conn.query_row(
"SELECT txn_id FROM txn_dedup WHERE user_id = ?1 AND device_id = ?2 AND event_id = ?3",
params![user_id, device_id, event_id],
|row| row.get(0),
)
.optional()
}
pub fn create_filter(conn: &Connection, user_id: i64, definition: &str) -> rusqlite::Result<i64> {
conn.execute("INSERT INTO filters (user_id, definition) VALUES (?1, ?2)", params![user_id, definition])?;
Ok(conn.last_insert_rowid())
}
pub fn get_filter(conn: &Connection, user_id: i64, filter_id: i64) -> rusqlite::Result<Option<String>> {
conn.query_row(
"SELECT definition FROM filters WHERE id = ?1 AND user_id = ?2",
params![filter_id, user_id],
|row| row.get(0),
)
.optional()
}
pub fn insert_legacy_dm_message_map(conn: &Connection, legacy_message_id: i64, event_id: &str) -> rusqlite::Result<()> {
conn.execute(
"INSERT INTO legacy_dm_message_map (legacy_message_id, event_id) VALUES (?1, ?2)",
params![legacy_message_id, event_id],
)?;
Ok(())
}
pub fn legacy_dm_message_event_id(conn: &Connection, legacy_message_id: i64) -> rusqlite::Result<Option<String>> {
conn.query_row(
"SELECT event_id FROM legacy_dm_message_map WHERE legacy_message_id = ?1",
params![legacy_message_id],
|row| row.get(0),
)
.optional()
}
pub fn highest_mapped_legacy_message_id(conn: &Connection, room_id: &str) -> rusqlite::Result<Option<i64>> {
conn.query_row(
"SELECT MAX(m.legacy_message_id) FROM legacy_dm_message_map m
JOIN events e ON e.event_id = m.event_id
WHERE e.room_id = ?1",
params![room_id],
|row| row.get(0),
)
}
#[derive(Debug, Clone, PartialEq)]
pub struct LegacyDmMessageImport {
pub legacy_message_id: i64,
pub event_id: String,
pub sender_user_id: i64,
pub content: String,
pub origin_server_ts: i64,
}
#[derive(Debug, Clone, PartialEq)]
pub struct LegacyDmReadReceipt {
pub reader_user_id: i64,
pub up_to_legacy_message_id: i64,
pub ts_ms: i64,
}
#[derive(Debug, Clone, PartialEq)]
pub struct LegacyDmDirectHint {
pub user_id: i64,
pub peer_mxid: String,
}
#[derive(Debug, Clone, Copy)]
pub struct DmMigrationExtras<'a> {
pub messages: &'a [LegacyDmMessageImport],
pub receipts: &'a [LegacyDmReadReceipt],
pub direct_hints: &'a [LegacyDmDirectHint],
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct DmMigrationCounts {
pub messages_imported: usize,
pub receipts_set: usize,
}
fn merge_m_direct_in_tx(tx: &Transaction, user_id: i64, peer_mxid: &str, room_id: &str) -> Result<(), MatrixStoreError> {
let existing: Option<String> = tx
.query_row(
"SELECT content FROM account_data WHERE user_id = ?1 AND room_id = ?2 AND data_type = 'm.direct'",
params![user_id, GLOBAL_ACCOUNT_DATA_ROOM],
|row| row.get(0),
)
.optional()?;
let mut direct: serde_json::Map<String, serde_json::Value> = match existing {
Some(content) => serde_json::from_str(&content)?,
None => serde_json::Map::new(),
};
let rooms_entry = direct.entry(peer_mxid.to_string()).or_insert_with(|| serde_json::Value::Array(Vec::new()));
if !rooms_entry.is_array() {
*rooms_entry = serde_json::Value::Array(Vec::new());
}
if let serde_json::Value::Array(list) = rooms_entry {
if !list.iter().any(|v| v.as_str() == Some(room_id)) {
list.push(serde_json::Value::String(room_id.to_string()));
}
}
let stream_id = next_stream_id(tx)?;
tx.execute(
"INSERT INTO account_data (user_id, room_id, data_type, content, stream_id) VALUES (?1, ?2, 'm.direct', ?3, ?4)
ON CONFLICT(user_id, room_id, data_type) DO UPDATE SET content = excluded.content, stream_id = excluded.stream_id",
params![user_id, GLOBAL_ACCOUNT_DATA_ROOM, serde_json::Value::Object(direct).to_string(), stream_id],
)?;
Ok(())
}
pub fn migrate_dm_conversation(
conn: &mut Connection,
bootstrap: RoomBootstrap<'_>,
state_events: &[NewStateEvent],
bootstrap_origin_server_ts: i64,
extras: DmMigrationExtras<'_>,
) -> Result<(Room, DmMigrationCounts), MatrixStoreError> {
let tx = conn.transaction()?;
insert_room_row(
&tx,
bootstrap.room_id,
bootstrap.kind,
bootstrap.creator_user_id,
bootstrap.created_at,
bootstrap.is_encrypted,
bootstrap.join_rule,
bootstrap.history_visibility,
bootstrap.dm_pair_key,
bootstrap.legacy_dm_id,
)?;
for event in state_events {
apply_state_event_in_tx(
&tx,
&event.event_id,
bootstrap.room_id,
event.sender_user_id,
&event.event_type,
&event.state_key,
&event.content,
bootstrap_origin_server_ts,
bootstrap.created_at,
)?;
}
let mut event_id_by_legacy_id: std::collections::HashMap<i64, String> = std::collections::HashMap::new();
for message in extras.messages {
insert_timeline_event_in_tx(
&tx,
&TimelineEventRow {
event_id: &message.event_id,
room_id: bootstrap.room_id,
sender_user_id: message.sender_user_id,
event_type: "org.example.legacy_dm",
content: &message.content,
origin_server_ts: message.origin_server_ts,
txn_id: None,
},
)?;
insert_legacy_dm_message_map(&tx, message.legacy_message_id, &message.event_id)?;
event_id_by_legacy_id.insert(message.legacy_message_id, message.event_id.clone());
}
let mut receipts_set = 0usize;
for receipt in extras.receipts {
let Some(target_event_id) = event_id_by_legacy_id.get(&receipt.up_to_legacy_message_id) else { continue };
let target_stream_id: i64 =
tx.query_row("SELECT stream_id FROM events WHERE event_id = ?1", params![target_event_id], |row| row.get(0))?;
tx.execute(
"INSERT INTO receipts (room_id, user_id, receipt_type, event_id, ts, stream_id) VALUES (?1, ?2, 'm.read', ?3, ?4, ?5)",
params![bootstrap.room_id, receipt.reader_user_id, target_event_id, receipt.ts_ms, target_stream_id],
)?;
receipts_set += 1;
}
for hint in extras.direct_hints {
merge_m_direct_in_tx(&tx, hint.user_id, &hint.peer_mxid, bootstrap.room_id)?;
}
tx.commit()?;
Ok((
Room {
id: bootstrap.room_id.to_string(),
kind: bootstrap.kind,
room_version: MATRIX_ROOM_VERSION.to_string(),
creator_user_id: bootstrap.creator_user_id,
created_at: bootstrap.created_at.to_string(),
is_encrypted: bootstrap.is_encrypted,
join_rule: bootstrap.join_rule,
history_visibility: bootstrap.history_visibility,
dm_pair_key: bootstrap.dm_pair_key.map(str::to_string),
legacy_dm_id: bootstrap.legacy_dm_id,
},
DmMigrationCounts { messages_imported: extras.messages.len(), receipts_set },
))
}
pub fn catch_up_dm_conversation(
conn: &mut Connection,
room_id: &str,
messages: &[LegacyDmMessageImport],
receipts: &[LegacyDmReadReceipt],
) -> Result<DmMigrationCounts, MatrixStoreError> {
let tx = conn.transaction()?;
let counts = catch_up_dm_in_tx(&tx, room_id, messages, receipts)?;
tx.commit()?;
Ok(counts)
}
fn catch_up_dm_in_tx(
tx: &Transaction,
room_id: &str,
messages: &[LegacyDmMessageImport],
receipts: &[LegacyDmReadReceipt],
) -> Result<DmMigrationCounts, MatrixStoreError> {
for message in messages {
insert_timeline_event_in_tx(
tx,
&TimelineEventRow {
event_id: &message.event_id,
room_id,
sender_user_id: message.sender_user_id,
event_type: "org.example.legacy_dm",
content: &message.content,
origin_server_ts: message.origin_server_ts,
txn_id: None,
},
)?;
insert_legacy_dm_message_map(tx, message.legacy_message_id, &message.event_id)?;
}
let mut receipts_set = 0usize;
for receipt in receipts {
let Some(target_event_id) = legacy_dm_message_event_id(tx, receipt.up_to_legacy_message_id)? else { continue };
upsert_receipt_in_tx(tx, room_id, receipt.reader_user_id, ReceiptType::Read, &target_event_id, receipt.ts_ms)?;
receipts_set += 1;
}
Ok(DmMigrationCounts { messages_imported: messages.len(), receipts_set })
}
#[derive(Debug, Clone, Copy)]
pub struct DmAdoption<'a> {
pub room_id: &'a str,
pub legacy_dm_id: i64,
pub key_events: &'a [NewStateEvent],
pub key_events_origin_server_ts: i64,
pub now: &'a str,
}
pub fn adopt_dm_room_for_legacy(
conn: &mut Connection,
adoption: DmAdoption<'_>,
messages: &[LegacyDmMessageImport],
receipts: &[LegacyDmReadReceipt],
) -> Result<Option<DmMigrationCounts>, MatrixStoreError> {
let tx = conn.transaction()?;
let bound = tx.execute(
"UPDATE rooms SET legacy_dm_id = ?1 WHERE id = ?2 AND kind = 'dm' AND legacy_dm_id IS NULL",
params![adoption.legacy_dm_id, adoption.room_id],
)?;
if bound == 0 {
return Ok(None);
}
for event in adoption.key_events {
if current_state_event(&tx, adoption.room_id, &event.event_type, &event.state_key)?.is_some() {
continue;
}
apply_state_event_in_tx(
&tx,
&event.event_id,
adoption.room_id,
event.sender_user_id,
&event.event_type,
&event.state_key,
&event.content,
adoption.key_events_origin_server_ts,
adoption.now,
)?;
}
let counts = catch_up_dm_in_tx(&tx, adoption.room_id, messages, receipts)?;
tx.commit()?;
Ok(Some(counts))
}
#[derive(Debug, Clone, PartialEq)]
pub struct PublicRoomSummary {
pub room_id: String,
pub name: Option<String>,
pub topic: Option<String>,
pub num_joined_members: i64,
pub world_readable: bool,
}
pub fn public_rooms_page(
conn: &Connection,
after_room_id: Option<&str>,
limit: usize,
search_term: Option<&str>,
) -> Result<(Vec<PublicRoomSummary>, bool, i64), MatrixStoreError> {
let mut stmt = conn.prepare(
"SELECT r.id, r.history_visibility,
(SELECT COUNT(*) FROM room_members WHERE room_id = r.id AND membership = 'join')
FROM rooms r WHERE r.join_rule = 'public' ORDER BY r.id ASC",
)?;
let mut rows = stmt.query([])?;
let mut all = Vec::new();
while let Some(row) = rows.next()? {
let room_id: String = row.get(0)?;
let history_visibility_raw: String = row.get(1)?;
let num_joined_members: i64 = row.get(2)?;
let world_readable = history_visibility_raw == HistoryVisibility::WorldReadable.as_str();
let name = current_state_event(conn, &room_id, "m.room.name", "")?
.and_then(|e| serde_json::from_str::<serde_json::Value>(&e.content).ok())
.and_then(|v| v.get("name").and_then(|n| n.as_str()).map(str::to_string));
let topic = current_state_event(conn, &room_id, "m.room.topic", "")?
.and_then(|e| serde_json::from_str::<serde_json::Value>(&e.content).ok())
.and_then(|v| v.get("topic").and_then(|t| t.as_str()).map(str::to_string));
all.push(PublicRoomSummary { room_id, name, topic, num_joined_members, world_readable });
}
drop(rows);
drop(stmt);
let total_room_count_estimate = all.len() as i64;
let filtered: Vec<PublicRoomSummary> = match search_term {
Some(term) if !term.is_empty() => {
let term_lower = term.to_lowercase();
all.into_iter().filter(|room| room.name.as_deref().is_some_and(|n| n.to_lowercase().contains(&term_lower))).collect()
}
_ => all,
};
let start = match after_room_id {
Some(cursor) => filtered.iter().position(|r| r.room_id == cursor).map_or(0, |idx| idx + 1),
None => 0,
};
let has_more = filtered.len() > start + limit;
let page: Vec<PublicRoomSummary> = filtered.into_iter().skip(start).take(limit).collect();
Ok((page, has_more, total_room_count_estimate))
}
#[cfg(test)]
mod tests {
use super::*;
const T0: &str = "2026-09-24T00:00:00+00:00";
const ROOM: &str = "!testroom:example.org";
fn test_conn() -> Connection {
let conn = Connection::open_in_memory().expect("in-memory sqlite");
create_matrix_schema(&conn).expect("schema");
conn
}
fn ensure_legacy_dm_map_table(conn: &Connection) {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS legacy_dm_message_map (
legacy_message_id INTEGER PRIMARY KEY,
event_id TEXT NOT NULL UNIQUE REFERENCES events(event_id)
);",
)
.expect("legacy map table for remaining unit tests");
}
fn make_room(conn: &Connection) {
create_room(conn, ROOM, RoomKind::Group, 1, T0, false, JoinRule::Invite, HistoryVisibility::Shared, None, None).expect("create room");
}
#[test]
fn next_stream_id_is_strictly_monotonic_across_every_table() {
let mut conn = test_conn();
make_room(&conn);
let event = insert_timeline_event(&mut conn, "$event1", ROOM, 1, "m.room.message", "{}", 1000).expect("insert event");
let account_data_stream = upsert_account_data(&mut conn, 1, GLOBAL_ACCOUNT_DATA_ROOM, "m.direct", "{}").expect("account data");
let receipt_stream = upsert_receipt(&mut conn, ROOM, 1, ReceiptType::Read, "$event1", 1500).expect("receipt");
assert!(event.stream_id < account_data_stream, "event {} should precede account data {}", event.stream_id, account_data_stream);
assert!(
account_data_stream < receipt_stream,
"account data {account_data_stream} should precede receipt {receipt_stream}"
);
}
#[test]
fn apply_state_event_replaces_current_state_but_keeps_history_in_events() {
let mut conn = test_conn();
make_room(&conn);
apply_state_event(&mut conn, &StateEventWrite { event_id: "$e1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.name", state_key: "", content: r#"{"name":"first"}"#, origin_server_ts: 1000, now: T0 }).expect("apply first");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$e2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.name", state_key: "", content: r#"{"name":"second"}"#, origin_server_ts: 2000, now: T0 }).expect("apply second");
let current = current_state_event(&conn, ROOM, "m.room.name", "").expect("query").expect("row exists");
assert_eq!(current.event_id, "$e2");
assert_eq!(current.content, r#"{"name":"second"}"#);
let e1 = get_event(&conn, "$e1").expect("get e1").expect("row exists");
let e2 = get_event(&conn, "$e2").expect("get e2").expect("row exists");
assert_eq!(e1.content, r#"{"name":"first"}"#);
assert_eq!(e2.content, r#"{"name":"second"}"#);
}
#[test]
fn redact_event_strips_content_per_v11_allow_list_by_type() {
let cases: Vec<(&str, &str, serde_json::Value)> = vec![
(
"m.room.member",
r#"{"membership":"join","join_authorised_via_users_server":"@x:example.org","avatar_url":"mxc://x","displayname":"Bob","third_party_invite":{"display_name":"bob@example.com","signed":{"mxid":"@bob:example.org"}}}"#,
serde_json::json!({
"membership": "join",
"join_authorised_via_users_server": "@x:example.org",
"third_party_invite": {"signed": {"mxid": "@bob:example.org"}}
}),
),
(
"m.room.power_levels",
r#"{"ban":50,"events":{},"events_default":0,"invite":50,"kick":50,"redact":50,"state_default":50,"users":{"@a:x":100},"users_default":0,"extra":"drop me"}"#,
serde_json::json!({
"ban": 50, "events": {}, "events_default": 0, "invite": 50, "kick": 50,
"redact": 50, "state_default": 50, "users": {"@a:x": 100}, "users_default": 0
}),
),
(
"m.room.history_visibility",
r#"{"history_visibility":"shared","extra":"drop"}"#,
serde_json::json!({"history_visibility": "shared"}),
),
(
"m.room.message",
r#"{"body":"hi","msgtype":"m.text"}"#,
serde_json::json!({}),
),
];
for (event_type, content, expected) in cases {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$target", ROOM, 1, event_type, content, 1000).expect("insert target");
redact_event(&mut conn, ROOM, "$target", "$redaction", 1, None, 2000).expect("redact");
let target = get_event(&conn, "$target").expect("get target").expect("row exists");
let got: serde_json::Value = serde_json::from_str(&target.content).expect("parse stripped content");
assert_eq!(got, expected, "event_type={event_type}");
assert_eq!(target.redacted_by.as_deref(), Some("$redaction"), "event_type={event_type}");
}
}
#[test]
fn redaction_content_carries_the_target_as_redacts_for_every_write_path() {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$t1", ROOM, 1, "m.room.message", "{}", 1000).expect("insert t1");
insert_timeline_event(&mut conn, "$t2", ROOM, 1, "m.room.message", "{}", 1100).expect("insert t2");
insert_timeline_event(&mut conn, "$t3", ROOM, 1, "m.room.message", "{}", 1200).expect("insert t3");
let plain = redact_event(&mut conn, ROOM, "$t1", "$r1", 1, None, 2000).expect("plain redact");
assert_eq!(serde_json::from_str::<serde_json::Value>(&plain.content).expect("json"), serde_json::json!({ "redacts": "$t1" }));
assert_eq!(plain.redacts.as_deref(), Some("$t1"));
let marked = redact_event_marked(
&mut conn,
&Redaction { room_id: ROOM, target_event_id: "$t2", redaction_event_id: "$r2", sender_user_id: 1, reason: Some("spam"), origin_server_ts: 2100 },
&serde_json::json!({ "org.example.site_moderation": true }),
)
.expect("marked redact");
assert_eq!(
serde_json::from_str::<serde_json::Value>(&marked.content).expect("json"),
serde_json::json!({ "redacts": "$t2", "reason": "spam", "org.example.site_moderation": true })
);
let deduped = redact_event_deduped(&mut conn, "DEV1", "txn-1", ROOM, "$t3", "$r3", 1, None, 2200, T0).expect("deduped redact");
let DedupedWrite::New(event) = deduped else { panic!("first write is new") };
let stored = get_event(&conn, &event.event_id).expect("get").expect("row exists");
assert_eq!(serde_json::from_str::<serde_json::Value>(&stored.content).expect("json")["redacts"], "$t3");
}
#[test]
fn redact_content_per_v11_keeps_everything_for_m_room_create() {
let content = r#"{"room_version":"11","creator":"@a:example.org"}"#;
let stripped = redact_content_per_v11("m.room.create", content).expect("strip");
let got: serde_json::Value = serde_json::from_str(&stripped).expect("parse");
assert_eq!(got, serde_json::json!({"room_version": "11", "creator": "@a:example.org"}));
}
#[test]
fn txn_dedup_returns_the_same_event_id_on_a_repeated_txn_id() {
let mut conn = test_conn();
make_room(&conn);
assert_eq!(txn_dedup_lookup(&conn, 1, "DEV1", "txn-1").expect("lookup 1"), TxnDedupEntry::NotSeen);
let event = {
let tx = conn.transaction().expect("tx");
let row = TimelineEventRow {
event_id: "$e1",
room_id: ROOM,
sender_user_id: 1,
event_type: "m.room.message",
content: "{}",
origin_server_ts: 1000,
txn_id: Some("txn-1"),
};
let event = insert_timeline_event_in_tx(&tx, &row).expect("insert");
tx.commit().expect("commit");
event
};
txn_dedup_record(&conn, 1, "DEV1", "txn-1", Some(&event.event_id), T0).expect("record");
assert_eq!(
txn_dedup_lookup(&conn, 1, "DEV1", "txn-1").expect("lookup 2"),
TxnDedupEntry::Seen(Some(event.event_id.clone()))
);
let count: i64 = conn.query_row("SELECT COUNT(*) FROM events", [], |row| row.get(0)).expect("count");
assert_eq!(count, 1, "a repeated txn_id must never create a second events row");
}
#[test]
fn relations_index_populated_from_cleartext_relates_to_even_when_content_is_ciphertext() {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$target", ROOM, 1, "m.room.message", "{}", 1000).expect("target");
let content = r#"{"algorithm":"m.megolm.v1.aes-sha2","ciphertext":"opaque-base64==","sender_key":"opaque","m.relates_to":{"rel_type":"m.annotation","event_id":"$target","key":"a"}}"#;
let reaction = insert_timeline_event(&mut conn, "$reaction", ROOM, 2, "m.room.encrypted", content, 2000).expect("insert reaction");
let (rel_type, target_id, agg_key): (String, String, Option<String>) = conn
.query_row(
"SELECT rel_type, target_id, agg_key FROM relations WHERE event_id = ?1",
params![reaction.event_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.expect("relation row exists");
assert_eq!(rel_type, "m.annotation");
assert_eq!(target_id, "$target");
assert_eq!(agg_key.as_deref(), Some("a"));
}
#[test]
fn account_data_global_row_is_unique() {
let mut conn = test_conn();
upsert_account_data(&mut conn, 1, GLOBAL_ACCOUNT_DATA_ROOM, "m.direct", r#"{"v":1}"#).expect("first upsert");
upsert_account_data(&mut conn, 1, GLOBAL_ACCOUNT_DATA_ROOM, "m.direct", r#"{"v":2}"#).expect("second upsert");
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM account_data WHERE user_id = 1 AND room_id = ''",
[],
|row| row.get(0),
)
.expect("count");
assert_eq!(count, 1, "two global upserts of the same type must leave one row");
let row = get_account_data(&conn, 1, GLOBAL_ACCOUNT_DATA_ROOM, "m.direct").expect("get").expect("row exists");
assert_eq!(row.content, r#"{"v":2}"#);
}
#[test]
fn receipt_never_moves_backwards() {
let mut conn = test_conn();
make_room(&conn);
let e1 = insert_timeline_event(&mut conn, "$e1", ROOM, 1, "m.room.message", "{}", 1000).expect("e1");
let e2 = insert_timeline_event(&mut conn, "$e2", ROOM, 1, "m.room.message", "{}", 2000).expect("e2");
upsert_receipt(&mut conn, ROOM, 9, ReceiptType::Read, &e2.event_id, 5000).expect("advance to e2");
upsert_receipt(&mut conn, ROOM, 9, ReceiptType::Read, &e1.event_id, 6000).expect("attempted backward move is a no-op");
let receipt = get_receipt(&conn, ROOM, 9, ReceiptType::Read).expect("get").expect("row exists");
assert_eq!(receipt.event_id, e2.event_id, "a receipt must never move back to an earlier event");
}
#[test]
fn receipt_further_ahead_of_the_two_types_wins_for_notification_count() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000001", T0).expect("alice");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 900, now: T0 }).expect("alice joins");
insert_timeline_event(&mut conn, "$e1", ROOM, 2, "m.room.message", "{}", 1000).expect("e1");
let e2 = insert_timeline_event(&mut conn, "$e2", ROOM, 2, "m.room.message", "{}", 2000).expect("e2");
insert_timeline_event(&mut conn, "$e3", ROOM, 2, "m.room.message", "{}", 3000).expect("e3");
let e4 = insert_timeline_event(&mut conn, "$e4", ROOM, 2, "m.room.message", "{}", 4000).expect("e4");
let e5 = insert_timeline_event(&mut conn, "$e5", ROOM, 2, "m.room.message", "{}", 5000).expect("e5");
upsert_receipt(&mut conn, ROOM, 1, ReceiptType::Read, &e2.event_id, 2500).expect("read receipt");
upsert_receipt(&mut conn, ROOM, 1, ReceiptType::ReadPrivate, &e4.event_id, 4500).expect("private receipt further ahead");
let room = get_room(&conn, ROOM).expect("get room").expect("room exists");
assert_eq!(notification_count(&conn, &room, 1).expect("count"), 1, "only $e5 is after the further-ahead receipt ($e4)");
upsert_receipt(&mut conn, ROOM, 1, ReceiptType::Read, &e5.event_id, 5500).expect("read receipt advances past e5");
assert_eq!(notification_count(&conn, &room, 1).expect("count"), 0, "m.read now covers every message");
}
#[test]
fn notification_count_ignores_own_state_and_reactions() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000001", T0).expect("alice");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 900, now: T0 }).expect("alice joins");
insert_timeline_event(&mut conn, "$bob-msg", ROOM, 2, "m.room.message", "{}", 1000).expect("bob message");
insert_timeline_event(&mut conn, "$bob-enc", ROOM, 2, "m.room.encrypted", "{}", 1100).expect("bob encrypted");
insert_timeline_event(&mut conn, "$bob-legacy", ROOM, 2, "org.example.legacy_dm", "{}", 1200).expect("bob legacy dm");
insert_timeline_event(&mut conn, "$alice-msg", ROOM, 1, "m.room.message", "{}", 1300).expect("alice's own message");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$name", room_id: ROOM, sender_user_id: 2, event_type: "m.room.name", state_key: "", content: r#"{"name":"x"}"#, origin_server_ts: 1400, now: T0 }).expect("state event");
insert_timeline_event(&mut conn, "$reaction", ROOM, 2, "m.reaction", "{}", 1500).expect("reaction");
let room = get_room(&conn, ROOM).expect("get room").expect("room exists");
assert_eq!(notification_count(&conn, &room, 1).expect("count"), 3, "only bob's 3 message-like sends count");
}
#[test]
fn notification_count_is_zero_for_a_non_member() {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$e1", ROOM, 2, "m.room.message", "{}", 1000).expect("e1");
let room = get_room(&conn, ROOM).expect("get room").expect("room exists");
assert_eq!(notification_count(&conn, &room, 999).expect("count"), 0, "a caller with no membership row at all sees nothing");
}
#[test]
fn duplicate_annotation_is_refused() {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$target", ROOM, 1, "m.room.message", "{}", 1000).expect("target");
let content = r#"{"m.relates_to":{"rel_type":"m.annotation","event_id":"$target","key":"a"}}"#;
insert_timeline_event(&mut conn, "$react1", ROOM, 5, "m.reaction", content, 2000).expect("first reaction");
let err = insert_timeline_event(&mut conn, "$react2", ROOM, 5, "m.reaction", content, 3000).unwrap_err();
assert!(matches!(err, MatrixStoreError::DuplicateAnnotation));
assert_eq!(get_event(&conn, "$react2").expect("get"), None);
let other = insert_timeline_event(&mut conn, "$react3", ROOM, 6, "m.reaction", content, 4000).expect("different sender reacts");
assert_eq!(other.event_id, "$react3");
}
#[test]
fn reannotation_after_redaction_is_allowed() {
let mut conn = test_conn();
make_room(&conn);
insert_timeline_event(&mut conn, "$target", ROOM, 1, "m.room.message", "{}", 1000).expect("target");
let content = r#"{"m.relates_to":{"rel_type":"m.annotation","event_id":"$target","key":"a"}}"#;
let first = insert_timeline_event(&mut conn, "$react1", ROOM, 5, "m.reaction", content, 2000).expect("first reaction");
redact_event(&mut conn, ROOM, &first.event_id, "$redaction", 1, None, 2500).expect("redact the reaction");
let second = insert_timeline_event(&mut conn, "$react2", ROOM, 5, "m.reaction", content, 3000)
.expect("re-annotation after redaction must succeed");
assert_eq!(second.event_id, "$react2");
}
#[test]
fn relation_target_must_exist() {
let mut conn = test_conn();
make_room(&conn);
let content = r#"{"m.relates_to":{"rel_type":"m.annotation","event_id":"$missing","key":"a"}}"#;
let err = insert_timeline_event(&mut conn, "$react1", ROOM, 5, "m.reaction", content, 2000).unwrap_err();
assert!(matches!(err, MatrixStoreError::InvalidRelationTarget(ref id) if id == "$missing"));
}
#[test]
fn relation_target_must_be_in_the_same_room() {
let mut conn = test_conn();
make_room(&conn);
let other_room = format!("!other:{}", matrix_server_name());
create_room(&conn, &other_room, RoomKind::Group, 1, T0, false, JoinRule::Invite, HistoryVisibility::Shared, None, None)
.expect("other room");
insert_timeline_event(&mut conn, "$target", &other_room, 1, "m.room.message", "{}", 1000).expect("target in other room");
let content = r#"{"m.relates_to":{"rel_type":"m.annotation","event_id":"$target","key":"a"}}"#;
let err = insert_timeline_event(&mut conn, "$react1", ROOM, 5, "m.reaction", content, 2000).unwrap_err();
assert!(matches!(err, MatrixStoreError::InvalidRelationTarget(ref id) if id == "$target"));
}
#[test]
fn redact_event_refuses_a_target_in_a_different_room() {
let mut conn = test_conn();
make_room(&conn);
let other_room = format!("!other:{}", matrix_server_name());
create_room(&conn, &other_room, RoomKind::Group, 1, T0, false, JoinRule::Invite, HistoryVisibility::Shared, None, None)
.expect("other room");
insert_timeline_event(&mut conn, "$target", &other_room, 1, "m.room.message", "{}", 1000).expect("target in other room");
let err = redact_event(&mut conn, ROOM, "$target", "$redaction", 1, None, 2000).unwrap_err();
assert!(matches!(err, MatrixStoreError::WrongRoom(ref id) if id == "$target"));
}
#[test]
fn redact_event_refuses_create_and_encryption_events() {
for event_type in ["m.room.create", "m.room.encryption"] {
let mut conn = test_conn();
make_room(&conn);
apply_state_event(&mut conn, &StateEventWrite { event_id: "$target", room_id: ROOM, sender_user_id: 1, event_type, state_key: "", content: r#"{"a":1}"#, origin_server_ts: 1000, now: T0 }).expect("apply state event");
let err = redact_event(&mut conn, ROOM, "$target", "$redaction", 1, None, 2000).unwrap_err();
assert!(
matches!(err, MatrixStoreError::UnredactableEvent(ref t) if t == event_type),
"event_type={event_type}"
);
}
}
#[test]
fn upsert_receipt_refuses_a_target_in_a_different_room() {
let mut conn = test_conn();
make_room(&conn);
let other_room = format!("!other:{}", matrix_server_name());
create_room(&conn, &other_room, RoomKind::Group, 1, T0, false, JoinRule::Invite, HistoryVisibility::Shared, None, None)
.expect("other room");
let event = insert_timeline_event(&mut conn, "$e1", &other_room, 1, "m.room.message", "{}", 1000).expect("event in other room");
let err = upsert_receipt(&mut conn, ROOM, 9, ReceiptType::Read, &event.event_id, 5000).unwrap_err();
assert!(matches!(err, MatrixStoreError::WrongRoom(ref id) if id == &event.event_id));
}
#[test]
fn mxid_parse_refuses_foreign_server() {
assert_eq!(public_id_from_mxid("@abc123:example.org"), Ok("abc123"));
assert_eq!(public_id_from_mxid("@abc123:otherserver.example"), Err(MatrixIdError::ForeignServerName));
let aliases = ["chat.example", "m4a.example.net", "m4a.example.org"];
set_local_aliases(aliases.iter().map(|s| s.to_string()));
for name in aliases {
assert_eq!(public_id_from_mxid(&format!("@abc123:{name}")), Ok("abc123"));
}
assert_eq!(public_id_from_mxid("@abc123:evil.example"), Err(MatrixIdError::ForeignServerName));
assert_eq!(mxid_for_public_id("abc123"), format!("@abc123:{}", matrix_server_name()), "minting never uses an alias");
assert_eq!(public_id_from_mxid("abc123:example.org"), Err(MatrixIdError::MissingSigil));
assert_eq!(public_id_from_mxid("@abc123"), Err(MatrixIdError::MissingServerName));
}
#[test]
fn ensure_matrix_user_is_idempotent_and_refuses_a_reserved_localpart() {
let conn = test_conn();
let mxid = ensure_matrix_user(&conn, 1, "abc123", T0).expect("first ensure");
assert_eq!(mxid, "@abc123:example.org");
let mxid_again = ensure_matrix_user(&conn, 1, "abc123", T0).expect("second ensure is a no-op");
assert_eq!(mxid_again, mxid);
assert_eq!(mxid_of(&conn, 1).expect("mxid_of"), Some(mxid.clone()));
assert_eq!(user_id_of(&conn, &mxid).expect("user_id_of"), Some(1));
let err = ensure_matrix_user(&conn, 2, "_bridge_evil", T0).unwrap_err();
assert!(matches!(err, MatrixStoreError::ReservedLocalpart));
}
#[test]
fn apply_state_event_member_refreshes_room_members_and_power_levels() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000001", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 2, "bob000000000000000000000000002", T0).expect("bob");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"invite"}"#, origin_server_ts: 1100, now: T0 }).expect("bob invited");
let joined = room_members(&conn, ROOM, Some(Membership::Join)).expect("joined members");
assert_eq!(joined.len(), 1);
assert_eq!(joined[0].user_id, 1);
assert_eq!(joined[0].power_level, None);
let power_levels_content = serde_json::json!({"users": {alice.clone(): 100}, "users_default": 0}).to_string();
apply_state_event(&mut conn, &StateEventWrite { event_id: "$pl1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.power_levels", state_key: "", content: &power_levels_content, origin_server_ts: 1200, now: T0 }).expect("power levels");
let alice_row = room_members(&conn, ROOM, None)
.expect("all members")
.into_iter()
.find(|m| m.user_id == 1)
.expect("alice row");
assert_eq!(alice_row.power_level, Some(100));
let bob_rooms = rooms_for_user(&conn, 2, Some(Membership::Invite)).expect("bob's invited rooms");
assert_eq!(bob_rooms, vec![ROOM.to_string()]);
}
#[test]
fn events_in_room_after_and_before_page_in_the_documented_order() {
let mut conn = test_conn();
make_room(&conn);
let e1 = insert_timeline_event(&mut conn, "$e1", ROOM, 1, "m.room.message", "{}", 1000).expect("e1");
let e2 = insert_timeline_event(&mut conn, "$e2", ROOM, 1, "m.room.message", "{}", 2000).expect("e2");
let e3 = insert_timeline_event(&mut conn, "$e3", ROOM, 1, "m.room.message", "{}", 3000).expect("e3");
let after = events_in_room_after(&conn, ROOM, e1.stream_id, 10).expect("after");
assert_eq!(after.iter().map(|e| e.event_id.clone()).collect::<Vec<_>>(), vec![e2.event_id.clone(), e3.event_id.clone()]);
let before = events_in_room_before(&conn, ROOM, e3.stream_id, 10).expect("before");
assert_eq!(before.iter().map(|e| e.event_id.clone()).collect::<Vec<_>>(), vec![e2.event_id.clone(), e1.event_id.clone()]);
assert_eq!(max_stream_id(&conn).expect("max"), e3.stream_id);
}
#[test]
fn txn_dedup_lookup_of_a_to_device_send_has_no_event_id() {
let conn = test_conn();
assert_eq!(txn_dedup_lookup(&conn, 1, "DEV1", "txn-td").expect("lookup"), TxnDedupEntry::NotSeen);
txn_dedup_record(&conn, 1, "DEV1", "txn-td", None, T0).expect("record to-device send");
assert_eq!(txn_dedup_lookup(&conn, 1, "DEV1", "txn-td").expect("lookup again"), TxnDedupEntry::Seen(None));
}
#[test]
fn filters_create_and_get_are_scoped_to_their_owner() {
let conn = test_conn();
let filter_id = create_filter(&conn, 1, r#"{"room":{"timeline":{"limit":20}}}"#).expect("create");
assert_eq!(get_filter(&conn, 1, filter_id).expect("owner reads it"), Some(r#"{"room":{"timeline":{"limit":20}}}"#.to_string()));
assert_eq!(get_filter(&conn, 2, filter_id).expect("a different user cannot"), None);
}
#[test]
fn legacy_dm_message_map_insert_and_get_round_trip() {
let mut conn = test_conn();
make_room(&conn);
ensure_legacy_dm_map_table(&conn);
let event = insert_timeline_event(&mut conn, "$legacy1", ROOM, 1, "org.example.legacy_dm", "{}", 1000).expect("insert");
insert_legacy_dm_message_map(&conn, 42, &event.event_id).expect("map insert");
assert_eq!(legacy_dm_message_event_id(&conn, 42).expect("map get"), Some(event.event_id));
assert_eq!(legacy_dm_message_event_id(&conn, 999).expect("missing"), None);
}
#[test]
fn power_level_defaults_apply_when_fields_missing() {
let pl = serde_json::json!({});
assert_eq!(user_level(&pl, "@nobody:example.org"), 0, "users_default defaults to 0");
assert_eq!(event_level(&pl, "m.room.message", false), 0, "events_default defaults to 0");
assert_eq!(event_level(&pl, "m.room.name", true), 50, "state_default defaults to 50");
for action in [PowerAction::Invite, PowerAction::Kick, PowerAction::Ban, PowerAction::Redact, PowerAction::StateDefault] {
assert!(!can(&pl, action, "@nobody:example.org"), "level 0 must not reach the default 50 threshold for {action:?}");
}
let pl_with_creator = serde_json::json!({ "users": { "@creator:example.org": 100 } });
assert!(can(&pl_with_creator, PowerAction::Ban, "@creator:example.org"));
assert!(can(&pl_with_creator, PowerAction::StateDefault, "@creator:example.org"));
}
#[test]
fn event_level_uses_the_events_type_override_before_falling_back_to_a_default() {
let pl = serde_json::json!({ "events": { "m.room.name": 60 }, "events_default": 0, "state_default": 50 });
assert_eq!(event_level(&pl, "m.room.name", true), 60, "an explicit events[type] override wins");
assert_eq!(event_level(&pl, "m.room.topic", true), 50, "an unlisted state type falls back to state_default");
assert_eq!(event_level(&pl, "m.room.message", false), 0, "an unlisted timeline type falls back to events_default");
}
#[test]
fn can_act_on_requires_strictly_greater_level_except_self_leave() {
let pl = serde_json::json!({
"users": { "@owner:example.org": 100, "@admin:example.org": 50, "@peer:example.org": 50 },
"kick": 50,
"ban": 50,
});
assert!(!can_act_on(&pl, PowerAction::Kick, "@admin:example.org", "@owner:example.org", false));
assert!(!can_act_on(&pl, PowerAction::Kick, "@admin:example.org", "@peer:example.org", false));
assert!(can_act_on(&pl, PowerAction::Kick, "@owner:example.org", "@admin:example.org", false));
assert!(can_act_on(&pl, PowerAction::Kick, "@admin:example.org", "@admin:example.org", true));
assert!(!can_act_on(&pl, PowerAction::Ban, "@admin:example.org", "@admin:example.org", false));
}
#[test]
fn validate_power_levels_change_refuses_raising_self_above_own_level() {
let old = serde_json::json!({ "users": { "@admin:example.org": 50 } });
let new = serde_json::json!({ "users": { "@admin:example.org": 100 } });
assert!(validate_power_levels_change(&old, &new, "@admin:example.org").is_err());
}
#[test]
fn validate_power_levels_change_refuses_demoting_a_peer_at_an_equal_level() {
let old = serde_json::json!({ "users": { "@a:example.org": 50, "@b:example.org": 50 } });
let new = serde_json::json!({ "users": { "@a:example.org": 50, "@b:example.org": 0 } });
assert!(validate_power_levels_change(&old, &new, "@a:example.org").is_err());
}
#[test]
fn validate_power_levels_change_allows_demoting_self() {
let old = serde_json::json!({ "users": { "@admin:example.org": 50 } });
let new = serde_json::json!({ "users": { "@admin:example.org": 10 } });
assert!(validate_power_levels_change(&old, &new, "@admin:example.org").is_ok());
}
#[test]
fn validate_power_levels_change_refuses_raising_events_default_above_own_level() {
let old = serde_json::json!({ "users": { "@admin:example.org": 50 }, "events_default": 0 });
let new = serde_json::json!({ "users": { "@admin:example.org": 50 }, "events_default": 60 });
assert!(validate_power_levels_change(&old, &new, "@admin:example.org").is_err());
}
#[test]
fn validate_power_levels_change_allows_the_owner_changing_anything_up_to_their_own_level() {
let old = serde_json::json!({ "users": { "@owner:example.org": 100, "@a:example.org": 50 } });
let new = serde_json::json!({
"users": { "@owner:example.org": 100, "@a:example.org": 90 },
"ban": 100,
"kick": 100,
"events_default": 100,
});
assert!(validate_power_levels_change(&old, &new, "@owner:example.org").is_ok());
}
#[test]
fn validate_power_levels_change_refuses_a_scalar_field_change_above_own_level() {
let old = serde_json::json!({ "users": { "@admin:example.org": 50 }, "ban": 50 });
let new = serde_json::json!({ "users": { "@admin:example.org": 50 }, "ban": 75 });
assert!(validate_power_levels_change(&old, &new, "@admin:example.org").is_err());
}
#[test]
fn validate_power_levels_change_ignores_unchanged_fields() {
let old = serde_json::json!({ "users": { "@admin:example.org": 50 }, "ban": 50, "events": { "m.room.name": 40 } });
let new = old.clone();
assert!(validate_power_levels_change(&old, &new, "@admin:example.org").is_ok());
}
fn bootstrap<'a>(room_id: &'a str, kind: RoomKind, creator: i64) -> RoomBootstrap<'a> {
RoomBootstrap {
room_id,
kind,
creator_user_id: creator,
created_at: T0,
is_encrypted: true,
join_rule: if kind == RoomKind::Channel { JoinRule::Public } else { JoinRule::Invite },
history_visibility: HistoryVisibility::Shared,
dm_pair_key: None,
legacy_dm_id: None,
}
}
fn state_event(event_id: &str, sender: i64, event_type: &str, state_key: &str, content: &str) -> NewStateEvent {
NewStateEvent {
event_id: event_id.to_string(),
sender_user_id: sender,
event_type: event_type.to_string(),
state_key: state_key.to_string(),
content: content.to_string(),
}
}
#[test]
fn create_room_with_state_inserts_the_room_and_every_bootstrap_event_atomically() {
let mut conn = test_conn();
let creator_mxid = ensure_matrix_user(&conn, 1, "creator0000000000000000000001", T0).expect("creator");
let room_id = "!batch:example.org";
let events = vec![
state_event("$create", 1, "m.room.create", "", r#"{"room_version":"11"}"#),
state_event("$m1", 1, "m.room.member", &creator_mxid, r#"{"membership":"join"}"#),
state_event(
"$pl",
1,
"m.room.power_levels",
"",
&serde_json::json!({"users": {creator_mxid.clone(): 100}, "users_default": 0}).to_string(),
),
];
let (room, applied) = create_room_with_state(&mut conn, bootstrap(room_id, RoomKind::Group, 1), &events, 1000).expect("create batch");
assert_eq!(room.id, room_id);
assert_eq!(applied.len(), 3);
assert!(get_room(&conn, room_id).expect("get room").is_some());
for event in &applied {
assert!(get_event(&conn, &event.event_id).expect("get event").is_some());
}
let creator_row = room_member(&conn, room_id, 1).expect("member row").expect("row exists");
assert_eq!(creator_row.power_level, Some(100), "power_levels applied after the member row existed");
}
#[test]
fn create_room_is_atomic_on_failure() {
let mut conn = test_conn();
let creator_mxid = ensure_matrix_user(&conn, 1, "creator0000000000000000000002", T0).expect("creator");
let room_id = "!atomic:example.org";
let events = vec![
state_event("$create", 1, "m.room.create", "", r#"{"room_version":"11"}"#),
state_event("$m1", 1, "m.room.member", &creator_mxid, r#"{"membership":"join"}"#),
state_event("$bad", 1, "m.room.member", "@ghost:example.org", r#"{"membership":"invite"}"#),
];
let err = create_room_with_state(&mut conn, bootstrap(room_id, RoomKind::Group, 1), &events, 1000).unwrap_err();
assert!(matches!(err, MatrixStoreError::UnknownMxid(ref m) if m == "@ghost:example.org"));
assert_eq!(get_room(&conn, room_id).expect("get room"), None, "a failed batch must leave no room row");
assert_eq!(get_event(&conn, "$create").expect("get"), None, "a failed batch must leave no event rows at all");
assert_eq!(get_event(&conn, "$m1").expect("get"), None);
}
#[test]
fn dm_pair_key_reuse_is_freed_once_the_room_is_not_reused() {
let conn = test_conn();
create_room(&conn, "!dm1:example.org", RoomKind::Dm, 1, T0, true, JoinRule::Invite, HistoryVisibility::Shared, Some("1:2"), None)
.expect("first dm room");
assert_eq!(room_by_dm_pair_key(&conn, "1:2").expect("lookup").map(|r| r.id), Some("!dm1:example.org".to_string()));
clear_dm_pair_key(&conn, "!dm1:example.org").expect("clear");
assert_eq!(room_by_dm_pair_key(&conn, "1:2").expect("lookup after clear"), None);
create_room(&conn, "!dm2:example.org", RoomKind::Dm, 1, T0, true, JoinRule::Invite, HistoryVisibility::Shared, Some("1:2"), None)
.expect("second dm room reuses the freed pair key");
assert_eq!(room_by_dm_pair_key(&conn, "1:2").expect("lookup").map(|r| r.id), Some("!dm2:example.org".to_string()));
}
#[test]
fn room_member_finds_the_one_row_forget_membership_deletes_only_when_left() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000099", T0).expect("alice");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("join");
assert_eq!(room_member(&conn, ROOM, 1).expect("member").map(|m| m.membership), Some(Membership::Join));
assert_eq!(room_member(&conn, ROOM, 999).expect("no such member"), None);
assert_eq!(forget_membership(&conn, ROOM, 1).expect("forget while joined"), 0);
assert!(room_member(&conn, ROOM, 1).expect("still a member").is_some());
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"leave"}"#, origin_server_ts: 2000, now: T0 }).expect("leave");
assert_eq!(forget_membership(&conn, ROOM, 1).expect("forget after leaving"), 1);
assert_eq!(room_member(&conn, ROOM, 1).expect("gone"), None);
}
#[test]
fn state_events_of_type_at_excludes_state_keys_created_after_the_cutoff() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000098", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 2, "bob0000000000000000000000000098", T0).expect("bob");
let e1 = apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins");
let cutoff = e1.stream_id;
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"invite"}"#, origin_server_ts: 1100, now: T0 }).expect("bob invited later");
let at_cutoff = state_events_of_type_at(&conn, ROOM, "m.room.member", cutoff).expect("at cutoff");
assert_eq!(at_cutoff.len(), 1, "bob's invite lands strictly after the cutoff and must be excluded");
assert_eq!(at_cutoff[0].event_id, "$m1");
let after_both = state_events_of_type_at(&conn, ROOM, "m.room.member", cutoff + 1).expect("after both");
assert_eq!(after_both.len(), 2);
}
#[test]
fn stripped_invite_state_includes_room_basics_and_the_inviters_own_member_event() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000097", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 2, "bob0000000000000000000000000097", T0).expect("bob");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$create", room_id: ROOM, sender_user_id: 1, event_type: "m.room.create", state_key: "", content: r#"{"room_version":"11"}"#, origin_server_ts: 900, now: T0 }).expect("create");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$jr", room_id: ROOM, sender_user_id: 1, event_type: "m.room.join_rules", state_key: "", content: r#"{"join_rule":"invite"}"#, origin_server_ts: 1100, now: T0 }).expect("join rules");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$m2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"invite"}"#, origin_server_ts: 1200, now: T0 }).expect("bob invited");
let stripped = stripped_invite_state(&conn, ROOM, 1).expect("stripped state");
let types: Vec<&str> = stripped.iter().map(|v| v["type"].as_str().expect("type")).collect();
assert!(types.contains(&"m.room.create"));
assert!(types.contains(&"m.room.join_rules"));
assert!(!types.contains(&"m.room.encryption"), "no encryption event exists in this room");
let inviter_member = stripped
.iter()
.find(|v| v["type"] == "m.room.member" && v["state_key"] == alice)
.expect("the inviter's own member event is included");
assert_eq!(inviter_member["sender"], alice);
assert_eq!(inviter_member["content"]["membership"], "join");
}
#[test]
fn user_ids_with_leave_transition_in_rooms_finds_only_leave_and_ban_inside_the_window() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 2, "alice00000000000000000000000097", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 3, "bob0000000000000000000000000097", T0).expect("bob");
let carol = ensure_matrix_user(&conn, 4, "carol0000000000000000000000097a", T0).expect("carol");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$a1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins");
let boundary = apply_state_event(&mut conn, &StateEventWrite { event_id: "$b1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"join"}"#, origin_server_ts: 1100, now: T0 })
.expect("bob joins")
.stream_id;
apply_state_event(&mut conn, &StateEventWrite { event_id: "$a2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"leave"}"#, origin_server_ts: 1200, now: T0 }).expect("alice leaves");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$b2", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"ban"}"#, origin_server_ts: 1300, now: T0 }).expect("bob banned");
apply_state_event(&mut conn, &StateEventWrite { event_id: "$c1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &carol, content: r#"{"membership":"join"}"#, origin_server_ts: 1400, now: T0 }).expect("carol joins (not a departure)");
let mut left = user_ids_with_leave_transition_in_rooms(&conn, &[ROOM.to_string()], boundary, i64::MAX).expect("query");
left.sort_unstable();
assert_eq!(left, vec![2, 3], "alice (leave) and bob (ban) both count; carol's join does not");
let empty = user_ids_with_leave_transition_in_rooms(&conn, &[], 0, i64::MAX).expect("empty room set");
assert!(empty.is_empty());
}
#[test]
fn membership_at_reads_the_state_as_of_a_stream_position() {
let mut conn = test_conn();
make_room(&conn);
let alice = ensure_matrix_user(&conn, 2, "alice00000000000000000000000081", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 3, "bob0000000000000000000000000081", T0).expect("bob");
let alice_joined = apply_state_event(&mut conn, &StateEventWrite { event_id: "$a1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins").stream_id;
let bob_invited = apply_state_event(&mut conn, &StateEventWrite { event_id: "$b1", room_id: ROOM, sender_user_id: 1, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"invite"}"#, origin_server_ts: 1100, now: T0 }).expect("bob invited").stream_id;
let bob_joined = apply_state_event(&mut conn, &StateEventWrite { event_id: "$b2", room_id: ROOM, sender_user_id: 3, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"join","displayname":"Bob"}"#, origin_server_ts: 1200, now: T0 }).expect("bob joins").stream_id;
assert_eq!(membership_at(&conn, ROOM, &bob, alice_joined).expect("query"), None, "bob had no member event yet");
assert_eq!(membership_at(&conn, ROOM, &bob, bob_invited).expect("query"), Some(Membership::Invite));
assert_eq!(membership_at(&conn, ROOM, &bob, bob_joined).expect("query"), Some(Membership::Join));
assert_eq!(membership_at(&conn, ROOM, &alice, bob_invited).expect("query"), Some(Membership::Join));
let mut window = member_state_keys_in_window(&conn, ROOM, bob_invited, bob_joined).expect("window");
window.sort();
assert_eq!(window, vec![bob.clone()], "only bob has a member event after `bob_invited`");
let mut all = member_state_keys_in_window(&conn, ROOM, 0, bob_joined).expect("whole history");
all.sort();
let mut expected = vec![alice, bob];
expected.sort();
assert_eq!(all, expected);
let rooms = [ROOM.to_string(), "!other:example.org".to_string()];
assert_eq!(rooms_with_member_events_in_window(&conn, &rooms, bob_invited, bob_joined).expect("rooms"), vec![ROOM.to_string()]);
assert!(rooms_with_member_events_in_window(&conn, &rooms, bob_joined, i64::MAX).expect("rooms after the last member event").is_empty());
assert!(rooms_with_member_events_in_window(&conn, &[], 0, i64::MAX).expect("empty room set").is_empty());
}
const ROOM_A: &str = "!roomA:example.org";
const ROOM_B: &str = "!roomB:example.org";
const ROOM_C: &str = "!roomC:example.org";
fn make_room_with_kind(conn: &Connection, room_id: &str, kind: RoomKind) {
create_room(conn, room_id, kind, 1, T0, false, JoinRule::Invite, HistoryVisibility::Shared, None, None).expect("create room");
}
fn member_content_in(conn: &Connection, room_id: &str, mxid: &str) -> serde_json::Value {
let event = current_state_event(conn, room_id, "m.room.member", mxid).expect("query").expect("member event exists");
serde_json::from_str(&event.content).expect("member content is json")
}
fn member_event_count(conn: &Connection, room_id: &str, mxid: &str) -> i64 {
conn.query_row(
"SELECT COUNT(*) FROM events WHERE room_id = ?1 AND event_type = 'm.room.member' AND state_key = ?2",
params![room_id, mxid],
|row| row.get(0),
)
.expect("count member events")
}
#[test]
fn refresh_member_displayname_restamps_every_joined_room_and_skips_a_room_the_user_left() {
let mut conn = test_conn();
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000091", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 2, "bob0000000000000000000000000091", T0).expect("bob");
for room in [ROOM_A, ROOM_B, ROOM_C] {
make_room_with_kind(&conn, room, RoomKind::Group);
apply_state_event(&mut conn, &StateEventWrite { event_id: &new_event_id(), room_id: room, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join","displayname":"old_alice"}"#, origin_server_ts: 1000, now: T0 })
.expect("alice joins");
apply_state_event(&mut conn, &StateEventWrite { event_id: &new_event_id(), room_id: room, sender_user_id: 2, event_type: "m.room.member", state_key: &bob, content: r#"{"membership":"join","displayname":"bob_nick"}"#, origin_server_ts: 1100, now: T0 })
.expect("bob joins");
}
apply_state_event(&mut conn, &StateEventWrite { event_id: &new_event_id(), room_id: ROOM_C, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"leave"}"#, origin_server_ts: 1200, now: T0 }).expect("alice leaves C");
let refresh = refresh_member_displayname(&mut conn, 1, "new_alice", T0, 5000).expect("refresh");
assert_eq!(refresh.rooms_updated, 2, "the two rooms alice is still joined in");
for room in [ROOM_A, ROOM_B] {
let content = member_content_in(&conn, room, &alice);
assert_eq!(content["membership"], "join");
assert_eq!(content["displayname"], "new_alice");
assert_eq!(member_event_count(&conn, room, &alice), 2, "a NEW member event lands; the old one stays in history");
let event = current_state_event(&conn, room, "m.room.member", &alice).expect("query").expect("exists");
assert_eq!(event.sender_user_id, 1, "a join refresh is sent by the user themself");
assert_eq!(event.origin_server_ts, 5000);
assert_eq!(member_content_in(&conn, room, &bob)["displayname"], "bob_nick", "another member's event is never touched");
}
assert_eq!(member_content_in(&conn, ROOM_C, &alice), serde_json::json!({ "membership": "leave" }), "the left room gets no new event");
assert_eq!(member_event_count(&conn, ROOM_C, &alice), 2, "join + leave, nothing more");
assert_eq!(refresh.affected_user_ids, HashSet::from([1, 2]), "alice and bob are woken; room C's members are not part of it");
let repeat = refresh_member_displayname(&mut conn, 1, "new_alice", T0, 6000).expect("second pass");
assert_eq!(repeat, DisplaynameRefresh::default(), "an up-to-date event is skipped, so a repeat pass writes nothing");
assert_eq!(member_event_count(&conn, ROOM_A, &alice), 2);
}
#[test]
fn refresh_member_displayname_keeps_an_invite_events_sender_and_is_direct() {
let mut conn = test_conn();
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000092", T0).expect("alice");
let bob = ensure_matrix_user(&conn, 2, "bob0000000000000000000000000092", T0).expect("bob");
make_room_with_kind(&conn, ROOM_A, RoomKind::Dm);
apply_state_event(&mut conn, &StateEventWrite { event_id: &new_event_id(), room_id: ROOM_A, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join","displayname":"alice_nick"}"#, origin_server_ts: 1000, now: T0 })
.expect("alice joins");
apply_state_event(
&mut conn,
&StateEventWrite {
event_id: &new_event_id(),
room_id: ROOM_A,
sender_user_id: 1,
event_type: "m.room.member",
state_key: &bob,
content: r#"{"membership":"invite","is_direct":true}"#,
origin_server_ts: 1100,
now: T0,
},
)
.expect("alice invites bob");
let refresh = refresh_member_displayname(&mut conn, 2, "bob_nick", T0, 5000).expect("refresh");
assert_eq!(refresh.rooms_updated, 1);
let content = member_content_in(&conn, ROOM_A, &bob);
assert_eq!(content, serde_json::json!({ "membership": "invite", "is_direct": true, "displayname": "bob_nick" }));
let event = current_state_event(&conn, ROOM_A, "m.room.member", &bob).expect("query").expect("exists");
assert_eq!(event.sender_user_id, 1, "the inviter stays the sender: stripped invite state reads the inviter off it");
assert_eq!(room_member(&conn, ROOM_A, 2).expect("query").expect("row").membership, Membership::Invite, "still an invitation");
assert_eq!(refresh.affected_user_ids, HashSet::from([1, 2]));
}
#[test]
fn refresh_member_displayname_is_a_noop_without_a_matrix_user_or_a_label() {
let mut conn = test_conn();
let alice = ensure_matrix_user(&conn, 1, "alice00000000000000000000000093", T0).expect("alice");
make_room_with_kind(&conn, ROOM_A, RoomKind::Group);
apply_state_event(&mut conn, &StateEventWrite { event_id: &new_event_id(), room_id: ROOM_A, sender_user_id: 1, event_type: "m.room.member", state_key: &alice, content: r#"{"membership":"join"}"#, origin_server_ts: 1000, now: T0 }).expect("alice joins");
assert_eq!(refresh_member_displayname(&mut conn, 99, "ghost", T0, 5000).expect("unknown user"), DisplaynameRefresh::default());
assert_eq!(refresh_member_displayname(&mut conn, 1, "", T0, 5000).expect("empty label"), DisplaynameRefresh::default());
assert_eq!(member_event_count(&conn, ROOM_A, &alice), 1);
assert_eq!(matrix_user_ids(&conn).expect("ids"), vec![1]);
}
fn native_dm_room(conn: &Connection) {
create_room(conn, ROOM, RoomKind::Dm, 1, T0, true, JoinRule::Invite, HistoryVisibility::Shared, Some("1:2"), None).expect("create dm room");
}
fn legacy_import(legacy_message_id: i64, event_id: &str) -> LegacyDmMessageImport {
LegacyDmMessageImport {
legacy_message_id,
event_id: event_id.to_string(),
sender_user_id: 1,
content: "{}".to_string(),
origin_server_ts: 1000 + legacy_message_id,
}
}
fn adoption_of(legacy_dm_id: i64) -> DmAdoption<'static> {
DmAdoption { room_id: ROOM, legacy_dm_id, key_events: &[], key_events_origin_server_ts: 500, now: T0 }
}
#[test]
fn adopt_dm_room_for_legacy_binds_the_room_and_imports_the_messages() {
let mut conn = test_conn();
ensure_legacy_dm_map_table(&conn);
native_dm_room(&conn);
let counts = adopt_dm_room_for_legacy(&mut conn, adoption_of(7), &[legacy_import(1, "$l1"), legacy_import(2, "$l2")], &[])
.expect("adopt")
.expect("the room is adoptable");
assert_eq!(counts.messages_imported, 2);
assert_eq!(room_by_legacy_dm_id(&conn, 7).expect("query").expect("bound").id, ROOM);
assert_eq!(highest_mapped_legacy_message_id(&conn, ROOM).expect("query"), Some(2));
assert_eq!(get_event(&conn, "$l1").expect("query").expect("imported").room_id, ROOM);
}
#[test]
fn adopt_dm_room_for_legacy_writes_a_key_event_only_where_the_room_has_none() {
let mut conn = test_conn();
ensure_legacy_dm_map_table(&conn);
native_dm_room(&conn);
let existing = apply_state_event(&mut conn, &StateEventWrite { event_id: "$k-existing", room_id: ROOM, sender_user_id: 1, event_type: "org.example.legacy_dm_key", state_key: "@a:example.org", content: r#"{"public_key_b64":"AAAA"}"#, origin_server_ts: 900, now: T0 })
.expect("existing key event");
let key_events = [
NewStateEvent {
event_id: "$k-a".to_string(),
sender_user_id: 1,
event_type: "org.example.legacy_dm_key".to_string(),
state_key: "@a:example.org".to_string(),
content: r#"{"public_key_b64":"BBBB"}"#.to_string(),
},
NewStateEvent {
event_id: "$k-b".to_string(),
sender_user_id: 2,
event_type: "org.example.legacy_dm_key".to_string(),
state_key: "@b:example.org".to_string(),
content: r#"{"public_key_b64":"CCCC"}"#.to_string(),
},
];
let adoption = DmAdoption { key_events: &key_events, ..adoption_of(7) };
adopt_dm_room_for_legacy(&mut conn, adoption, &[], &[]).expect("adopt").expect("adoptable");
let a = current_state_event(&conn, ROOM, "org.example.legacy_dm_key", "@a:example.org").expect("query").expect("still present");
assert_eq!(a.event_id, existing.event_id, "an existing key event is never overwritten");
let b = current_state_event(&conn, ROOM, "org.example.legacy_dm_key", "@b:example.org").expect("query").expect("added");
assert_eq!(b.event_id, "$k-b");
}
#[test]
fn adopt_dm_room_for_legacy_refuses_a_room_that_is_bound_or_not_a_dm() {
let mut conn = test_conn();
ensure_legacy_dm_map_table(&conn);
make_room(&conn);
assert!(adopt_dm_room_for_legacy(&mut conn, adoption_of(7), &[legacy_import(1, "$l1")], &[]).expect("adopt").is_none(), "a group room is never adopted");
assert!(room_by_legacy_dm_id(&conn, 7).expect("query").is_none());
let mut bound = test_conn();
create_room(&bound, ROOM, RoomKind::Dm, 1, T0, true, JoinRule::Invite, HistoryVisibility::Shared, Some("1:2"), Some(9)).expect("create bound room");
assert!(adopt_dm_room_for_legacy(&mut bound, adoption_of(7), &[legacy_import(1, "$l1")], &[]).expect("adopt").is_none(), "an already-bound room is never re-bound");
assert_eq!(room_by_legacy_dm_id(&bound, 9).expect("query").expect("still bound to 9").id, ROOM);
assert!(get_event(&bound, "$l1").expect("query").is_none(), "a refused adoption writes nothing");
}
#[test]
fn adopt_dm_room_for_legacy_rolls_back_the_binding_when_an_import_fails() {
let mut conn = test_conn();
ensure_legacy_dm_map_table(&conn);
native_dm_room(&conn);
let result = adopt_dm_room_for_legacy(&mut conn, adoption_of(7), &[legacy_import(1, "$dup"), legacy_import(2, "$dup")], &[]);
assert!(result.is_err());
assert!(room_by_legacy_dm_id(&conn, 7).expect("query").is_none(), "the binding must roll back with the failed import");
assert!(get_event(&conn, "$dup").expect("query").is_none(), "no imported event survives");
assert_eq!(highest_mapped_legacy_message_id(&conn, ROOM).expect("query"), None);
}
}