Skip to main content

max / makenotwork

8.8 KB · 298 lines History Blame Raw
1 //! thread writes. `auto_hide_if_threshold_met` is the concurrency-sensitive one: it must hide exactly once no matter how many flags land together
2
3 use super::{PgPool, Uuid};
4
5 /// Atomically create an externally-referenced thread and its opening post in
6 /// one transaction. Returns `(thread_id, op_post_id)`.
7 ///
8 /// The atomicity matters most here: the MNW→MT internal path retries, and a
9 /// committed-but-post-less thread would make the retry collide with the UNIQUE
10 /// `external_ref` index and 500 while leaking an empty thread.
11 #[tracing::instrument(skip_all)]
12 pub async fn create_thread_with_op_external_ref(
13 pool: &PgPool,
14 category_id: Uuid,
15 author_id: Uuid,
16 title: &str,
17 external_ref: &str,
18 body_markdown: &str,
19 body_html: &str,
20 ) -> Result<(Uuid, Uuid), sqlx::Error> {
21 let mut tx = pool.begin().await?;
22
23 let thread_id = sqlx::query_scalar!(
24 "INSERT INTO threads (category_id, author_id, title, external_ref)
25 VALUES ($1, $2, $3, $4)
26 RETURNING id",
27 category_id,
28 author_id,
29 title,
30 external_ref,
31 )
32 .fetch_one(&mut *tx)
33 .await?;
34
35 let post_id = sqlx::query_scalar!(
36 "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
37 VALUES ($1, $2, $3, $4)
38 RETURNING id",
39 thread_id,
40 author_id,
41 body_markdown,
42 body_html,
43 )
44 .fetch_one(&mut *tx)
45 .await?;
46
47 tx.commit().await?;
48 Ok((thread_id, post_id))
49 }
50
51 /// Low-level: insert a bare thread row (no opening post). Returns the thread ID.
52 ///
53 /// Handlers must NOT pair this with a separate `create_post` for the OP, that
54 /// non-atomic sequence can leak a post-less thread. Use [`create_thread_with_op`]
55 /// for the real "start a thread" operation. This primitive exists for tests and
56 /// data construction that build posts explicitly.
57 #[tracing::instrument(skip_all)]
58 pub async fn create_thread(
59 pool: &PgPool,
60 category_id: Uuid,
61 author_id: Uuid,
62 title: &str,
63 ) -> Result<Uuid, sqlx::Error> {
64 sqlx::query_scalar!(
65 "INSERT INTO threads (category_id, author_id, title)
66 VALUES ($1, $2, $3)
67 RETURNING id",
68 category_id,
69 author_id,
70 title,
71 )
72 .fetch_one(pool)
73 .await
74 }
75
76 /// Atomically create a thread and its opening post in one transaction.
77 ///
78 /// Returns `(thread_id, op_post_id)`. Creating the thread and OP separately
79 /// could leave a titled, post-less thread if the second insert failed; this
80 /// makes that state unreachable. `threads.last_activity_at` defaults to now().
81 #[tracing::instrument(skip_all)]
82 pub async fn create_thread_with_op(
83 pool: &PgPool,
84 category_id: Uuid,
85 author_id: Uuid,
86 title: &str,
87 body_markdown: &str,
88 body_html: &str,
89 ) -> Result<(Uuid, Uuid), sqlx::Error> {
90 let mut tx = pool.begin().await?;
91 let ids = create_thread_with_op_tx(
92 &mut tx,
93 category_id,
94 author_id,
95 title,
96 body_markdown,
97 body_html,
98 )
99 .await?;
100 tx.commit().await?;
101 Ok(ids)
102 }
103
104 /// Create a thread and its opening post on the caller's transaction.
105 ///
106 /// Lets a handler fold the thread, its OP, its mentions, and its tags into one
107 /// atomic unit so a mid-sequence failure can't leave a thread whose @mentions or
108 /// tags silently vanished (ultra-fuzz M-St1). The pool-taking
109 /// [`create_thread_with_op`] wraps this in its own transaction.
110 #[tracing::instrument(skip_all)]
111 pub async fn create_thread_with_op_tx(
112 conn: &mut sqlx::PgConnection,
113 category_id: Uuid,
114 author_id: Uuid,
115 title: &str,
116 body_markdown: &str,
117 body_html: &str,
118 ) -> Result<(Uuid, Uuid), sqlx::Error> {
119 let thread_id = sqlx::query_scalar!(
120 "INSERT INTO threads (category_id, author_id, title)
121 VALUES ($1, $2, $3)
122 RETURNING id",
123 category_id,
124 author_id,
125 title,
126 )
127 .fetch_one(&mut *conn)
128 .await?;
129
130 let post_id = sqlx::query_scalar!(
131 "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
132 VALUES ($1, $2, $3, $4)
133 RETURNING id",
134 thread_id,
135 author_id,
136 body_markdown,
137 body_html,
138 )
139 .fetch_one(&mut *conn)
140 .await?;
141
142 Ok((thread_id, post_id))
143 }
144
145 /// Atomically auto-hide a post if pending flag count meets the threshold.
146 /// Combines count check and removal in a single query to avoid race conditions.
147 /// Returns true if the post was actually removed.
148 /// Sets removed_by to NULL (system action), the mod log records the event.
149 #[tracing::instrument(skip_all)]
150 pub async fn auto_hide_if_threshold_met<'e, E: sqlx::PgExecutor<'e>>(
151 executor: E,
152 post_id: Uuid,
153 threshold: i32,
154 ) -> Result<bool, sqlx::Error> {
155 let result = sqlx::query!(
156 "UPDATE posts SET removed_by = NULL, removed_at = now()
157 WHERE id = $1 AND removed_at IS NULL
158 AND (SELECT COUNT(*) FROM post_flags WHERE post_id = $1 AND resolved_at IS NULL) >= $2",
159 post_id,
160 threshold as i64,
161 )
162 .execute(executor)
163 .await?;
164 Ok(result.rows_affected() > 0)
165 }
166
167 /// Update a thread's title.
168 #[tracing::instrument(skip_all)]
169 pub async fn update_thread_title(
170 pool: &PgPool,
171 thread_id: Uuid,
172 title: &str,
173 ) -> Result<(), sqlx::Error> {
174 sqlx::query!(
175 "UPDATE threads SET title = $2 WHERE id = $1",
176 thread_id,
177 title
178 )
179 .execute(pool)
180 .await?;
181 Ok(())
182 }
183
184 /// Soft-delete a thread: set deleted_at (hides from listings).
185 #[tracing::instrument(skip_all)]
186 pub async fn soft_delete_thread<'e, E: sqlx::PgExecutor<'e>>(
187 executor: E,
188 thread_id: Uuid,
189 ) -> Result<(), sqlx::Error> {
190 sqlx::query!(
191 "UPDATE threads SET deleted_at = now() WHERE id = $1",
192 thread_id
193 )
194 .execute(executor)
195 .await?;
196 Ok(())
197 }
198
199 /// Outcome of [`restore_thread_cascade`].
200 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
201 pub struct ThreadRestore {
202 /// The thread was restored by this call (false if it was not deleted).
203 pub thread_restored: bool,
204 /// The opening post had been removed with the thread, so it was restored
205 /// too. False when the thread was deleted directly and its opening post was
206 /// never removed.
207 pub op_restored: bool,
208 }
209
210 /// Un-delete a thread, and bring its opening post back if the removal of that
211 /// post is what deleted the thread.
212 ///
213 /// A thread reaches `deleted_at` two ways, and the reversal has to cover both.
214 /// `mod_remove_post_cascade` sets it as a consequence of removing the opening
215 /// post; `delete_thread_handler` sets it directly and leaves every post alone.
216 /// Restoring only the thread row would leave the first case headless, showing a
217 /// tombstone where the opening post should be, which is the state the delete
218 /// cascade exists to prevent in the first place.
219 ///
220 /// The opening post is identified the same way the removal identifies it
221 /// (earliest by `created_at`, `id` as tiebreak, regardless of removal state), so
222 /// the two stay in step. A post removed on its own merits that merely happens to
223 /// be first is indistinguishable from one removed by the cascade, and both want
224 /// restoring here: the thread is coming back, so its first post should be
225 /// readable.
226 #[tracing::instrument(skip_all)]
227 pub async fn restore_thread_cascade(
228 conn: &mut sqlx::PgConnection,
229 thread_id: Uuid,
230 ) -> Result<ThreadRestore, sqlx::Error> {
231 let thread_restored = sqlx::query!(
232 "UPDATE threads SET deleted_at = NULL WHERE id = $1 AND deleted_at IS NOT NULL",
233 thread_id,
234 )
235 .execute(&mut *conn)
236 .await?
237 .rows_affected()
238 > 0;
239
240 let mut op_restored = false;
241 if thread_restored {
242 op_restored = sqlx::query!(
243 "UPDATE posts SET removed_by = NULL, removed_at = NULL
244 WHERE removed_at IS NOT NULL
245 AND id = (
246 SELECT id FROM posts
247 WHERE thread_id = $1
248 ORDER BY created_at ASC, id ASC
249 LIMIT 1
250 )",
251 thread_id,
252 )
253 .execute(&mut *conn)
254 .await?
255 .rows_affected()
256 > 0;
257 }
258
259 Ok(ThreadRestore {
260 thread_restored,
261 op_restored,
262 })
263 }
264
265 /// Set or unset the pinned flag on a thread.
266 #[tracing::instrument(skip_all)]
267 pub async fn set_thread_pinned<'e, E: sqlx::PgExecutor<'e>>(
268 executor: E,
269 thread_id: Uuid,
270 pinned: bool,
271 ) -> Result<(), sqlx::Error> {
272 sqlx::query!(
273 "UPDATE threads SET pinned = $2 WHERE id = $1",
274 thread_id,
275 pinned
276 )
277 .execute(executor)
278 .await?;
279 Ok(())
280 }
281
282 /// Set or unset the locked flag on a thread.
283 #[tracing::instrument(skip_all)]
284 pub async fn set_thread_locked<'e, E: sqlx::PgExecutor<'e>>(
285 executor: E,
286 thread_id: Uuid,
287 locked: bool,
288 ) -> Result<(), sqlx::Error> {
289 sqlx::query!(
290 "UPDATE threads SET locked = $2 WHERE id = $1",
291 thread_id,
292 locked
293 )
294 .execute(executor)
295 .await?;
296 Ok(())
297 }
298