Skip to main content

max / makenotwork

6.3 KB · 230 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 /// Set or unset the pinned flag on a thread.
198 #[tracing::instrument(skip_all)]
199 pub async fn set_thread_pinned<'e, E: sqlx::PgExecutor<'e>>(
200 executor: E,
201 thread_id: Uuid,
202 pinned: bool,
203 ) -> Result<(), sqlx::Error> {
204 sqlx::query!(
205 "UPDATE threads SET pinned = $2 WHERE id = $1",
206 thread_id,
207 pinned
208 )
209 .execute(executor)
210 .await?;
211 Ok(())
212 }
213
214 /// Set or unset the locked flag on a thread.
215 #[tracing::instrument(skip_all)]
216 pub async fn set_thread_locked<'e, E: sqlx::PgExecutor<'e>>(
217 executor: E,
218 thread_id: Uuid,
219 locked: bool,
220 ) -> Result<(), sqlx::Error> {
221 sqlx::query!(
222 "UPDATE threads SET locked = $2 WHERE id = $1",
223 thread_id,
224 locked
225 )
226 .execute(executor)
227 .await?;
228 Ok(())
229 }
230