Skip to main content

max / makenotwork

5.8 KB · 162 lines History Blame Raw
1 //! chat message inserts, removals and the retention sweep
2 //!
3 //! Expiry is computed in SQL as `now() + make_interval(hours => …)` rather than
4 //! bound as a `chrono::DateTime`. That is not a style choice: the full server
5 //! build unifies sqlx's `time` and `chrono` features (the session store pulls
6 //! in `time`), and with both on, the compile-time macro infers a TIMESTAMPTZ
7 //! bind parameter as `time::OffsetDateTime`. An output column's type can be
8 //! overridden in the macro; a bind parameter's cannot. Computing the timestamp
9 //! server-side sidesteps the bind entirely and keeps every statement here
10 //! compile-time checked, which is what `mutations/moderation.rs` had to give up
11 //! to bind a ban's `expires_at`.
12
13 use super::{PgPool, Uuid};
14
15 /// A freshly inserted message, as the database assigned it.
16 pub struct InsertedChatMessage {
17 pub id: i64,
18 /// Unix seconds.
19 pub created_at: i64,
20 }
21
22 /// Store one message, stamping its expiry from the room's retention.
23 ///
24 /// `body_html` must already be rendered through docengine's chat preset. This
25 /// layer does not sanitize and must never be handed raw user input.
26 #[tracing::instrument(skip_all)]
27 pub async fn insert_chat_message(
28 pool: &PgPool,
29 community_id: Uuid,
30 author_id: Uuid,
31 body_html: &str,
32 retention_hours: i32,
33 ) -> Result<InsertedChatMessage, sqlx::Error> {
34 sqlx::query_as!(
35 InsertedChatMessage,
36 r#"INSERT INTO chat_messages (community_id, author_id, body_html, expires_at)
37 VALUES ($1, $2, $3, now() + make_interval(hours => $4))
38 RETURNING id, EXTRACT(EPOCH FROM created_at)::BIGINT AS "created_at!""#,
39 community_id,
40 author_id,
41 body_html,
42 retention_hours,
43 )
44 .fetch_one(pool)
45 .await
46 }
47
48 /// Remove one message from a room. Returns how many rows went, so a caller can
49 /// tell a real deletion from a message that had already expired.
50 #[tracing::instrument(skip_all)]
51 pub async fn delete_chat_message<'e, E: sqlx::PgExecutor<'e>>(
52 executor: E,
53 community_id: Uuid,
54 message_id: i64,
55 ) -> Result<u64, sqlx::Error> {
56 let result = sqlx::query!(
57 "DELETE FROM chat_messages WHERE community_id = $1 AND id = $2",
58 community_id,
59 message_id,
60 )
61 .execute(executor)
62 .await?;
63 Ok(result.rows_affected())
64 }
65
66 /// Remove every message one author wrote in a room, in one statement.
67 ///
68 /// This is the ban purge. It is a single statement keyed by (room, author)
69 /// because a ban is issued on a hot path, at the moment the room is busiest,
70 /// and the caller broadcasts one `Purge` event rather than one `Delete` per
71 /// row.
72 #[tracing::instrument(skip_all)]
73 pub async fn purge_author_messages<'e, E: sqlx::PgExecutor<'e>>(
74 executor: E,
75 community_id: Uuid,
76 author_id: Uuid,
77 ) -> Result<u64, sqlx::Error> {
78 let result = sqlx::query!(
79 "DELETE FROM chat_messages WHERE community_id = $1 AND author_id = $2",
80 community_id,
81 author_id,
82 )
83 .execute(executor)
84 .await?;
85 Ok(result.rows_affected())
86 }
87
88 /// Delete every expired message across every room.
89 ///
90 /// One indexed statement on `idx_chat_messages_expiry`, deliberately not keyed
91 /// by community: per-room would turn a single sweep into one statement per
92 /// community, and expiry is already stamped on the row.
93 #[tracing::instrument(skip_all)]
94 pub async fn sweep_expired_chat_messages(pool: &PgPool) -> Result<u64, sqlx::Error> {
95 let result = sqlx::query!("DELETE FROM chat_messages WHERE expires_at < now()")
96 .execute(pool)
97 .await?;
98 Ok(result.rows_affected())
99 }
100
101 /// Trim every room back to its `chat_max_messages` cap.
102 ///
103 /// The count half of retention: age expires a quiet room, this bounds a busy
104 /// one that would otherwise reach its age limit holding far more than the cap.
105 ///
106 /// One statement for the whole table rather than one per community. It does
107 /// walk every live chat row, which is affordable precisely because the other
108 /// half of retention already bounds that set, and because this runs on the
109 /// scheduler and never on a request. Per-community would be a query per
110 /// community on every tick, most of them finding nothing to do.
111 #[tracing::instrument(skip_all)]
112 pub async fn trim_chat_rooms_to_cap(pool: &PgPool) -> Result<u64, sqlx::Error> {
113 let result = sqlx::query!(
114 "DELETE FROM chat_messages c
115 USING (
116 SELECT ranked.id
117 FROM (
118 SELECT id,
119 community_id,
120 row_number() OVER (
121 PARTITION BY community_id ORDER BY id DESC
122 ) AS rn
123 FROM chat_messages
124 ) ranked
125 JOIN communities co ON co.id = ranked.community_id
126 WHERE ranked.rn > co.chat_max_messages
127 ) over_cap
128 WHERE c.id = over_cap.id"
129 )
130 .execute(pool)
131 .await?;
132 Ok(result.rows_affected())
133 }
134
135 /// Restamp a room's messages after its retention policy changed.
136 ///
137 /// Expiry is stored on the row at insert, which is what makes the sweep one
138 /// indexed delete. The cost of that choice is exactly this: shortening a
139 /// window has to reach back over messages already sent, or the change would
140 /// only apply to future ones and an owner who just cut retention from 30 days
141 /// to 1 would still be holding 30 days of chat.
142 ///
143 /// Recomputed from `created_at`, not from `now()`, so the new window means the
144 /// same thing for an old message as for a new one.
145 #[tracing::instrument(skip_all)]
146 pub async fn recompute_chat_expiry(
147 pool: &PgPool,
148 community_id: Uuid,
149 retention_hours: i32,
150 ) -> Result<u64, sqlx::Error> {
151 let result = sqlx::query!(
152 "UPDATE chat_messages
153 SET expires_at = created_at + make_interval(hours => $2)
154 WHERE community_id = $1",
155 community_id,
156 retention_hours,
157 )
158 .execute(pool)
159 .await?;
160 Ok(result.rows_affected())
161 }
162