mail4agent_server/
public_channels.rs

1//! Public plaintext store (channels; a forum could join later).
2//!
3//! BOUNDARY. This module owns the tables `pub_events` and `pub_txn`. Nothing
4//! here writes `events`, `relations`, `receipts` or `txn_dedup` (the closed
5//! store), and nothing in [`crate::retention`] ever reads or deletes these
6//! tables. Bodies stay: a public channel is plaintext by design (MLC
7//! decision), so posts are kept until a moderator redacts them.
8//!
9//! What stays in the closed store for a channel room: the room row, its
10//! membership and its *state* events (create, member, join rules, name,
11//! power levels). Only timeline events (`m.room.message`, `m.reaction`,
12//! `m.room.redaction`, ...) of a plaintext channel live here. DMs, groups and
13//! already-encrypted rooms never touch this module.
14//!
15//! The store shares the global stream counter with the closed store so
16//! `/sync` and `/messages` can merge both by stream id; that counter is the
17//! only shared state. Reactions/edits keep `m.relates_to` inside the content
18//! (clients fold them); there is no server-side `relations` aggregation here.
19//!
20//! Whether this store runs in the same process as the closed one is open
21//! (owner decision): the boundary is the table set plus this module's API.
22
23use rusqlite::{params, Connection, OptionalExtension};
24
25use crate::store::MatrixEvent;
26
27/// DDL for the public store. Called from `init_messenger_db`.
28pub fn create_public_schema(conn: &Connection) -> rusqlite::Result<()> {
29    conn.execute_batch(
30        r#"
31        CREATE TABLE IF NOT EXISTS pub_events (
32            stream_id        INTEGER PRIMARY KEY,
33            event_id         TEXT NOT NULL UNIQUE,
34            room_id          TEXT NOT NULL,
35            sender_user_id   INTEGER NOT NULL,
36            event_type       TEXT NOT NULL,
37            content          TEXT NOT NULL,
38            origin_server_ts INTEGER NOT NULL,
39            txn_id           TEXT,
40            redacts          TEXT,
41            redacted_by      TEXT
42        );
43        CREATE INDEX IF NOT EXISTS idx_pub_events_room_stream ON pub_events(room_id, stream_id);
44        CREATE TABLE IF NOT EXISTS pub_txn (
45            user_id    INTEGER NOT NULL,
46            device_id  TEXT NOT NULL,
47            txn_id     TEXT NOT NULL,
48            event_id   TEXT NOT NULL,
49            PRIMARY KEY (user_id, device_id, txn_id)
50        );
51        "#,
52    )
53}
54
55/// SQL predicate (on `rooms`) selecting plaintext public channels. The one
56/// definition of "public store room", also used by retention to stay away.
57pub const PUBLIC_ROOM_PREDICATE: &str = "kind = 'channel' AND is_encrypted = 0";
58
59/// Whether `room_id` is a plaintext public channel (its timeline lives here).
60pub fn is_public_room(conn: &Connection, room_id: &str) -> rusqlite::Result<bool> {
61    let found: Option<i64> = conn
62        .query_row(&format!("SELECT 1 FROM rooms WHERE id = ?1 AND {PUBLIC_ROOM_PREDICATE}"), params![room_id], |r| r.get(0))
63        .optional()?;
64    Ok(found.is_some())
65}
66
67const COLS: &str = "stream_id, event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id, redacts, redacted_by";
68
69fn from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<MatrixEvent> {
70    Ok(MatrixEvent {
71        stream_id: row.get(0)?,
72        event_id: row.get(1)?,
73        room_id: row.get(2)?,
74        sender_user_id: row.get(3)?,
75        event_type: row.get(4)?,
76        state_key: None,
77        content: row.get(5)?,
78        origin_server_ts: row.get(6)?,
79        txn_id: row.get(7)?,
80        redacts: row.get(8)?,
81        redacted_by: row.get(9)?,
82    })
83}
84
85/// One public event by id.
86pub fn get_event(conn: &Connection, event_id: &str) -> rusqlite::Result<Option<MatrixEvent>> {
87    conn.query_row(&format!("SELECT {COLS} FROM pub_events WHERE event_id = ?1"), params![event_id], from_row).optional()
88}
89
90/// Events strictly after `since_stream`, oldest first.
91pub fn events_after(conn: &Connection, room_id: &str, since_stream: i64, limit: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
92    let mut stmt = conn.prepare(&format!("SELECT {COLS} FROM pub_events WHERE room_id = ?1 AND stream_id > ?2 ORDER BY stream_id ASC LIMIT ?3"))?;
93    let rows = stmt.query_map(params![room_id, since_stream, limit], from_row)?;
94    rows.collect()
95}
96
97/// Events strictly before `before_stream`, newest first.
98pub fn events_before(conn: &Connection, room_id: &str, before_stream: i64, limit: i64) -> rusqlite::Result<Vec<MatrixEvent>> {
99    let mut stmt = conn.prepare(&format!("SELECT {COLS} FROM pub_events WHERE room_id = ?1 AND stream_id < ?2 ORDER BY stream_id DESC LIMIT ?3"))?;
100    let rows = stmt.query_map(params![room_id, before_stream, limit], from_row)?;
101    rows.collect()
102}
103
104/// The public event a `(user, device, txn)` already produced, if any.
105pub fn seen_txn(conn: &Connection, user_id: i64, device_id: &str, txn_id: &str) -> rusqlite::Result<Option<MatrixEvent>> {
106    let id: Option<String> = conn
107        .query_row("SELECT event_id FROM pub_txn WHERE user_id = ?1 AND device_id = ?2 AND txn_id = ?3", params![user_id, device_id, txn_id], |r| r.get(0))
108        .optional()?;
109    match id {
110        Some(id) => get_event(conn, &id),
111        None => Ok(None),
112    }
113}
114
115/// Result of a deduped public write.
116pub enum PublicWrite {
117    /// New event.
118    New(MatrixEvent),
119    /// Same `(user, device, txn)` seen before.
120    Existing(MatrixEvent),
121}
122
123/// Insert one timeline event (deduped per user/device/txn) into the public store.
124#[allow(clippy::too_many_arguments)]
125pub fn insert_event_deduped(
126    conn: &mut Connection,
127    device_id: &str,
128    txn_id: &str,
129    event_id: &str,
130    room_id: &str,
131    sender_user_id: i64,
132    event_type: &str,
133    content: &str,
134    origin_server_ts: i64,
135) -> rusqlite::Result<PublicWrite> {
136    let tx = conn.transaction()?;
137    let seen: Option<String> = tx
138        .query_row("SELECT event_id FROM pub_txn WHERE user_id = ?1 AND device_id = ?2 AND txn_id = ?3", params![sender_user_id, device_id, txn_id], |r| r.get(0))
139        .optional()?;
140    if let Some(existing) = seen {
141        let event = get_event(&tx, &existing)?.ok_or(rusqlite::Error::QueryReturnedNoRows)?;
142        tx.commit()?;
143        return Ok(PublicWrite::Existing(event));
144    }
145    let stream_id = crate::store::next_stream_id(&tx)?;
146    tx.execute(
147        "INSERT INTO pub_events (stream_id, event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id) VALUES (?1,?2,?3,?4,?5,?6,?7,?8)",
148        params![stream_id, event_id, room_id, sender_user_id, event_type, content, origin_server_ts, txn_id],
149    )?;
150    tx.execute("INSERT INTO pub_txn (user_id, device_id, txn_id, event_id) VALUES (?1,?2,?3,?4)", params![sender_user_id, device_id, txn_id, event_id])?;
151    tx.commit()?;
152    Ok(PublicWrite::New(MatrixEvent {
153        stream_id,
154        event_id: event_id.to_string(),
155        room_id: room_id.to_string(),
156        sender_user_id,
157        event_type: event_type.to_string(),
158        state_key: None,
159        content: content.to_string(),
160        origin_server_ts,
161        txn_id: Some(txn_id.to_string()),
162        redacts: None,
163        redacted_by: None,
164    }))
165}
166
167/// Redact a public event: strip its content, record the redaction event.
168#[allow(clippy::too_many_arguments)]
169pub fn redact_deduped(
170    conn: &mut Connection,
171    device_id: &str,
172    txn_id: &str,
173    room_id: &str,
174    target_event_id: &str,
175    redaction_event_id: &str,
176    sender_user_id: i64,
177    reason: Option<&str>,
178    origin_server_ts: i64,
179) -> rusqlite::Result<PublicWrite> {
180    let mut content = serde_json::json!({ "redacts": target_event_id });
181    if let Some(r) = reason {
182        content["reason"] = serde_json::Value::String(r.to_string());
183    }
184    let content = content.to_string();
185    let written = insert_event_deduped(conn, device_id, txn_id, redaction_event_id, room_id, sender_user_id, "m.room.redaction", &content, origin_server_ts)?;
186    if let PublicWrite::New(mut event) = written {
187        conn.execute("UPDATE pub_events SET redacts = ?1 WHERE event_id = ?2", params![target_event_id, redaction_event_id])?;
188        conn.execute("UPDATE pub_events SET content = '{}', redacted_by = ?1 WHERE event_id = ?2", params![redaction_event_id, target_event_id])?;
189        event.redacts = Some(target_event_id.to_string());
190        return Ok(PublicWrite::New(event));
191    }
192    Ok(written)
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198
199    fn db() -> Connection {
200        let conn = Connection::open_in_memory().unwrap();
201        crate::store::create_matrix_schema(&conn).unwrap();
202        crate::keys::create_matrix_keys_schema(&conn).unwrap();
203        crate::retention::create_retention_schema(&conn).unwrap();
204        create_public_schema(&conn).unwrap();
205        conn.execute("INSERT INTO rooms (id, kind, creator_user_id, created_at, is_encrypted) VALUES ('!c:x','channel',1,'t',0)", []).unwrap();
206        conn.execute("INSERT INTO rooms (id, kind, creator_user_id, created_at, is_encrypted) VALUES ('!g:x','group',1,'t',1)", []).unwrap();
207        conn
208    }
209
210    #[test]
211    fn only_plaintext_channels_are_public_rooms() {
212        let conn = db();
213        assert!(is_public_room(&conn, "!c:x").unwrap());
214        assert!(!is_public_room(&conn, "!g:x").unwrap());
215    }
216
217    #[test]
218    fn public_writes_never_touch_the_closed_tables_and_dedup() {
219        let mut conn = db();
220        let a = insert_event_deduped(&mut conn, "D", "t1", "$a", "!c:x", 1, "m.room.message", r#"{"body":"hi"}"#, 10).unwrap();
221        assert!(matches!(a, PublicWrite::New(_)));
222        let again = insert_event_deduped(&mut conn, "D", "t1", "$zzz", "!c:x", 1, "m.room.message", "{}", 11).unwrap();
223        assert!(matches!(again, PublicWrite::Existing(ref e) if e.event_id == "$a"));
224        for table in ["events", "relations", "receipts", "txn_dedup"] {
225            let n: i64 = conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0)).unwrap();
226            assert_eq!(n, 0, "{table} must stay untouched by the public store");
227        }
228        assert_eq!(events_after(&conn, "!c:x", 0, 10).unwrap().len(), 1);
229    }
230
231    #[test]
232    fn retention_never_deletes_public_posts() {
233        let mut conn = db();
234        for i in 0..5 {
235            insert_event_deduped(&mut conn, "D", &format!("t{i}"), &format!("$e{i}"), "!c:x", 1, "m.room.message", r#"{"body":"x"}"#, 1).unwrap();
236        }
237        let policy = crate::retention::RetentionPolicy { ttl_ms: 1, ack_grace_ms: 0, keep_last: 0, stale_device_ms: 1 };
238        crate::retention::purge_delivered_events(&mut conn, 10_000_000, &policy).unwrap();
239        let n: i64 = conn.query_row("SELECT COUNT(*) FROM pub_events", [], |r| r.get(0)).unwrap();
240        assert_eq!(n, 5);
241    }
242
243    #[test]
244    fn redaction_strips_content() {
245        let mut conn = db();
246        insert_event_deduped(&mut conn, "D", "t1", "$a", "!c:x", 1, "m.room.message", r#"{"body":"hi"}"#, 10).unwrap();
247        redact_deduped(&mut conn, "D", "t2", "!c:x", "$a", "$r", 1, Some("spam"), 11).unwrap();
248        let ev = get_event(&conn, "$a").unwrap().unwrap();
249        assert_eq!(ev.content, "{}");
250        assert_eq!(ev.redacted_by.as_deref(), Some("$r"));
251    }
252}
mail4agent_server/public_channels.rs

mail4agent_server/
public_channels.rs
