Skip to main content

max / makenotwork

5.6 KB · 174 lines History Blame Raw
1 //! chat backlog reads for one community room
2 //!
3 //! Timestamps come back as Unix seconds rather than `DateTime<Utc>` because
4 //! that is what `livechat::Message::created_at` is, and converting at the query
5 //! boundary keeps the chrono type out of the hot replay path entirely.
6
7 use super::{ChatPolicy, CommunityState, PgPool, Uuid};
8
9 /// Everything resolving a community slug to a `livechat::Room` needs.
10 ///
11 /// A dedicated query rather than widening `CommunityRow`: that struct is loaded
12 /// on effectively every request in the app, and chat is on almost none of them.
13 pub struct ChatRoomRow {
14 pub id: Uuid,
15 pub policy: ChatPolicy,
16 pub state: CommunityState,
17 pub suspended: bool,
18 pub retention_hours: i32,
19 pub max_messages: i32,
20 }
21
22 /// Resolve a community slug to its chat room.
23 ///
24 /// `Ok(None)` means no such community. A community whose chat is off still
25 /// comes back as a row: the caller maps `ChatPolicy::Off` to a closed room, and
26 /// keeping the two cases distinct here is what lets it decide whether "disabled"
27 /// and "absent" should look the same from outside (they should, and do).
28 #[tracing::instrument(skip_all)]
29 pub async fn get_chat_room_by_slug(
30 pool: &PgPool,
31 slug: &str,
32 ) -> Result<Option<ChatRoomRow>, sqlx::Error> {
33 sqlx::query_as!(
34 ChatRoomRow,
35 r#"SELECT id,
36 chat_policy AS "policy: ChatPolicy",
37 state AS "state: CommunityState",
38 (suspended_at IS NOT NULL) AS "suspended!",
39 chat_retention_hours AS retention_hours,
40 chat_max_messages AS max_messages
41 FROM communities
42 WHERE slug = $1"#,
43 slug,
44 )
45 .fetch_optional(pool)
46 .await
47 }
48
49 /// One stored chat message, shaped for `livechat::Message`.
50 ///
51 /// `author` is absent here on purpose: display name, avatar and flair are
52 /// resolved on the way out through `ChatIdentity::attach` and never stored, so
53 /// a rename takes effect everywhere rather than only on messages sent after it.
54 pub struct ChatMessageRow {
55 pub id: i64,
56 pub author_id: Uuid,
57 pub body_html: String,
58 /// Unix seconds.
59 pub created_at: i64,
60 }
61
62 /// Messages after `cursor`, oldest first, for a client resuming a stream.
63 ///
64 /// `limit` is a hard ceiling on one reply, not a page size: a client that has
65 /// been away longer than the retention window asks for everything and must not
66 /// be able to make the server materialize an unbounded row set. It receives the
67 /// oldest `limit` messages past its cursor and its next cursor advances, so a
68 /// long absence resolves in a few round trips instead of one large one.
69 #[tracing::instrument(skip_all)]
70 pub async fn backlog_after(
71 pool: &PgPool,
72 community_id: Uuid,
73 cursor: i64,
74 limit: i64,
75 ) -> Result<Vec<ChatMessageRow>, sqlx::Error> {
76 sqlx::query_as!(
77 ChatMessageRow,
78 r#"SELECT id,
79 author_id,
80 body_html,
81 EXTRACT(EPOCH FROM created_at)::BIGINT AS "created_at!"
82 FROM chat_messages
83 WHERE community_id = $1 AND id > $2
84 ORDER BY id
85 LIMIT $3"#,
86 community_id,
87 cursor,
88 limit,
89 )
90 .fetch_all(pool)
91 .await
92 }
93
94 /// The newest `limit` messages, oldest first, for a client with no cursor.
95 ///
96 /// Selected newest-first so the index gives the tail directly, then reversed to
97 /// the ascending order the room renders in. Doing that in SQL rather than in
98 /// Rust would mean sorting the whole room.
99 #[tracing::instrument(skip_all)]
100 pub async fn recent_backlog(
101 pool: &PgPool,
102 community_id: Uuid,
103 limit: i64,
104 ) -> Result<Vec<ChatMessageRow>, sqlx::Error> {
105 let mut rows = sqlx::query_as!(
106 ChatMessageRow,
107 r#"SELECT id,
108 author_id,
109 body_html,
110 EXTRACT(EPOCH FROM created_at)::BIGINT AS "created_at!"
111 FROM chat_messages
112 WHERE community_id = $1
113 ORDER BY id DESC
114 LIMIT $2"#,
115 community_id,
116 limit,
117 )
118 .fetch_all(pool)
119 .await?;
120
121 rows.reverse();
122 Ok(rows)
123 }
124
125 /// Who wrote a message, if it exists in this room.
126 ///
127 /// Scoped by `community_id` rather than looked up by id alone: a message id is
128 /// a guessable integer, and answering "who wrote 4172" without asking which
129 /// room it is in would confirm the existence and authorship of a message in a
130 /// community the caller cannot read. The self-delete entitlement check is the
131 /// caller for this.
132 #[tracing::instrument(skip_all)]
133 pub async fn chat_message_author(
134 pool: &PgPool,
135 community_id: Uuid,
136 message_id: i64,
137 ) -> Result<Option<Uuid>, sqlx::Error> {
138 sqlx::query_scalar!(
139 "SELECT author_id FROM chat_messages WHERE community_id = $1 AND id = $2",
140 community_id,
141 message_id,
142 )
143 .fetch_optional(pool)
144 .await
145 }
146
147 /// Display fields for many users at once, for `ChatIdentity`.
148 ///
149 /// One query for the whole batch: backlog replay needs every author in the
150 /// window, and doing that per author is an N+1 on the reconnect path, which is
151 /// the exact moment every client in the room arrives at once.
152 pub struct ChatIdentityRow {
153 pub id: Uuid,
154 pub username: String,
155 pub display_name: Option<String>,
156 pub avatar_url: Option<String>,
157 }
158
159 #[tracing::instrument(skip_all)]
160 pub async fn chat_identities(
161 pool: &PgPool,
162 users: &[Uuid],
163 ) -> Result<Vec<ChatIdentityRow>, sqlx::Error> {
164 sqlx::query_as!(
165 ChatIdentityRow,
166 "SELECT mnw_account_id AS id, username, display_name, avatar_url
167 FROM users
168 WHERE mnw_account_id = ANY($1)",
169 users,
170 )
171 .fetch_all(pool)
172 .await
173 }
174