Skip to main content

max / makenotwork

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