max / makenotwork

7.9 KB · 224 lines History Blame Raw

1	use super::{PgPool, Uuid};
2
3	/// Insert a reply and bump the thread's last_activity_at atomically.
4	///
5	/// Opening posts are created with the thread via [`create_thread_with_op`];
6	/// this is for replies only. The display reply count derives from the
7	/// denormalized `threads.post_count`, which the m029 trigger maintains on every
8	/// post INSERT/DELETE/removed_at change, no counter to update by hand here.
9	#[tracing::instrument(skip_all)]
10	pub async fn create_post(
11	pool: &PgPool,
12	thread_id: Uuid,
13	author_id: Uuid,
14	body_markdown: &str,
15	body_html: &str,
16	) -> Result<Uuid, sqlx::Error> {
17	let mut tx = pool.begin().await?;
18	let post_id = create_post_tx(&mut tx, thread_id, author_id, body_markdown, body_html)
19	.await?
20	// The pool-taking wrapper is for callers that have already established the
21	// thread is live (tests, simple internal paths); a locked/deleted thread
22	// here is a "row not inserted" error rather than a silent no-op.
23	.ok_or(sqlx::Error::RowNotFound)?;
24	tx.commit().await?;
25	Ok(post_id)
26	}
27
28	/// Insert a reply and bump `last_activity_at` on the caller's transaction.
29	///
30	/// Lets a handler fold the reply and its mentions into one atomic unit so a
31	/// mid-sequence failure can't leave a committed reply whose @mentions silently
32	/// vanished (ultra-fuzz M-St1). The pool-taking [`create_post`] wraps this in
33	/// its own transaction for the simple, mention-free callers (and tests).
34	#[tracing::instrument(skip_all)]
35	pub async fn create_post_tx(
36	conn: &mut sqlx::PgConnection,
37	thread_id: Uuid,
38	author_id: Uuid,
39	body_markdown: &str,
40	body_html: &str,
41	) -> Result<Option<Uuid>, sqlx::Error> {
42	// Insert only while the thread is still live and unlocked, in the same
43	// statement that reads that state, closing the TOCTOU between a handler's
44	// snapshot-time `locked`/`deleted_at` check and this insert. Without the
45	// guard a concurrent lock or soft-delete in the gap would let the reply
46	// commit (and the m029 post_count trigger increment a deleted thread). None
47	// = the thread was locked/deleted meanwhile; the caller surfaces it. This
48	// mirrors the conditional-write pattern `mod_remove_post_cascade` uses.
49	let post_id = sqlx::query_scalar!(
50	"INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
51	SELECT $1, $2, $3, $4
52	WHERE EXISTS (
53	SELECT 1 FROM threads
54	WHERE id = $1 AND deleted_at IS NULL AND locked = false
55	)
56	RETURNING id",
57	thread_id,
58	author_id,
59	body_markdown,
60	body_html,
61	)
62	.fetch_optional(&mut *conn)
63	.await?;
64
65	// Only bump activity if the reply actually landed.
66	if post_id.is_some() {
67	sqlx::query!(
68	"UPDATE threads SET last_activity_at = now() WHERE id = $1",
69	thread_id
70	)
71	.execute(&mut *conn)
72	.await?;
73	}
74
75	Ok(post_id)
76	}
77
78	/// Idempotent reply insert keyed on `external_ref`, for the internal API.
79	///
80	/// Mirrors [`create_thread_with_op_external_ref`]: a retried or replayed
81	/// server→server reply carrying the same `external_ref` inserts at most once.
82	/// `ON CONFLICT` on the partial unique index (m030) makes the repeat a no-op,
83	/// so the m029 post_count trigger fires exactly once and `last_activity_at` is
84	/// bumped only on the fresh insert. Returns `(post_id, created)`, `created` is
85	/// false when an existing post was returned. User-facing replies use the plain
86	/// [`create_post`] (no ref); this path is internal-only.
87	#[tracing::instrument(skip_all)]
88	pub async fn create_post_external_ref(
89	pool: &PgPool,
90	thread_id: Uuid,
91	author_id: Uuid,
92	external_ref: &str,
93	body_markdown: &str,
94	body_html: &str,
95	) -> Result<(Uuid, bool), sqlx::Error> {
96	let mut tx = pool.begin().await?;
97
98	let inserted = sqlx::query_scalar!(
99	"INSERT INTO posts (thread_id, author_id, body_markdown, body_html, external_ref)
100	VALUES ($1, $2, $3, $4, $5)
101	ON CONFLICT (external_ref) WHERE external_ref IS NOT NULL DO NOTHING
102	RETURNING id",
103	thread_id,
104	author_id,
105	body_markdown,
106	body_html,
107	external_ref,
108	)
109	.fetch_optional(&mut *tx)
110	.await?;
111
112	let result = match inserted {
113	Some(id) => {
114	// Fresh reply: bump thread activity (the trigger already counted it).
115	sqlx::query!(
116	"UPDATE threads SET last_activity_at = now() WHERE id = $1",
117	thread_id
118	)
119	.execute(&mut *tx)
120	.await?;
121	(id, true)
122	}
123	None => {
124	// Replay: the reply already exists. Return its id, don't re-bump.
125	let existing =
126	sqlx::query_scalar!("SELECT id FROM posts WHERE external_ref = $1", external_ref,)
127	.fetch_one(&mut *tx)
128	.await?;
129	(existing, false)
130	}
131	};
132
133	tx.commit().await?;
134	Ok(result)
135	}
136
137	/// Test-only: mod-remove a single post without the OP-cascade.
138	///
139	/// Production code must use [`mod_remove_post_cascade`], which also soft-deletes
140	/// the thread when the post is its opening post, removing an OP without the
141	/// cascade leaves a headless, still-repliable thread. This non-cascade variant
142	/// exists solely so the integration suite can stage a removed post directly; it
143	/// is gated behind the `test-support` feature so handler code cannot reach it.
144	#[cfg(feature = "test-support")]
145	#[tracing::instrument(skip_all)]
146	pub async fn mod_remove_post(
147	pool: &PgPool,
148	post_id: Uuid,
149	removed_by_id: Uuid,
150	) -> Result<bool, sqlx::Error> {
151	let result = sqlx::query!(
152	"UPDATE posts SET removed_by = $2, removed_at = now()
153	WHERE id = $1 AND removed_at IS NULL",
154	post_id,
155	removed_by_id,
156	)
157	.execute(pool)
158	.await?;
159	Ok(result.rows_affected() > 0)
160	}
161
162	/// Outcome of [`mod_remove_post_cascade`].
163	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
164	pub struct PostRemoval {
165	/// The post was removed by this call (false if it was already removed).
166	pub post_removed: bool,
167	/// The post was the thread's opening post, so the whole thread was
168	/// soft-deleted as well.
169	pub thread_removed: bool,
170	}
171
172	/// Mod-remove a post and, if it is the thread's opening post, soft-delete the
173	/// whole thread in the same transaction.
174	///
175	/// The opening post is the earliest post in the thread (by `created_at`, `id`
176	/// as tiebreak), counted regardless of removal state so the identity is stable.
177	/// Removing an OP without this cascade would leave a headless, still-repliable
178	/// thread; cascading deletes it so listings, the thread view, and the reply
179	/// path all treat it as gone. Returns false flags when the post was already
180	/// removed or was not the OP, callers can log a thread-deletion accordingly.
181	#[tracing::instrument(skip_all)]
182	pub async fn mod_remove_post_cascade(
183	conn: &mut sqlx::PgConnection,
184	post_id: Uuid,
185	removed_by_id: Uuid,
186	) -> Result<PostRemoval, sqlx::Error> {
187	let post_removed = sqlx::query!(
188	"UPDATE posts SET removed_by = $2, removed_at = now()
189	WHERE id = $1 AND removed_at IS NULL",
190	post_id,
191	removed_by_id,
192	)
193	.execute(&mut *conn)
194	.await?
195	.rows_affected()
196	> 0;
197
198	let mut thread_removed = false;
199	if post_removed {
200	// Soft-delete the thread only when this post is its opening post.
201	thread_removed = sqlx::query!(
202	"UPDATE threads SET deleted_at = now()
203	WHERE deleted_at IS NULL
204	AND id = (SELECT thread_id FROM posts WHERE id = $1)
205	AND $1 = (
206	SELECT id FROM posts
207	WHERE thread_id = (SELECT thread_id FROM posts WHERE id = $1)
208	ORDER BY created_at ASC, id ASC
209	LIMIT 1
210	)",
211	post_id,
212	)
213	.execute(&mut *conn)
214	.await?
215	.rows_affected()
216	> 0;
217	}
218
219	Ok(PostRemoval {
220	post_removed,
221	thread_removed,
222	})
223	}
224