max / makenotwork

2.6 KB · 85 lines History Blame Raw

1	//! tracking writes and read-position advance, which must never move backwards
2
3	use super::{PgPool, Uuid};
4
5	/// Track a thread (upsert).
6	#[tracing::instrument(skip_all)]
7	pub async fn track_thread(
8	pool: &PgPool,
9	user_id: Uuid,
10	thread_id: Uuid,
11	) -> Result<(), sqlx::Error> {
12	sqlx::query!(
13	"INSERT INTO tracked_threads (user_id, thread_id)
14	VALUES ($1, $2)
15	ON CONFLICT (user_id, thread_id) DO NOTHING",
16	user_id,
17	thread_id,
18	)
19	.execute(pool)
20	.await?;
21	Ok(())
22	}
23
24	/// Untrack a thread.
25	#[tracing::instrument(skip_all)]
26	pub async fn untrack_thread(
27	pool: &PgPool,
28	user_id: Uuid,
29	thread_id: Uuid,
30	) -> Result<(), sqlx::Error> {
31	sqlx::query!(
32	"DELETE FROM tracked_threads WHERE user_id = $1 AND thread_id = $2",
33	user_id,
34	thread_id,
35	)
36	.execute(pool)
37	.await?;
38	Ok(())
39	}
40
41	/// Stop tracking all threads for a user.
42	#[tracing::instrument(skip_all)]
43	pub async fn untrack_all(pool: &PgPool, user_id: Uuid) -> Result<(), sqlx::Error> {
44	sqlx::query!("DELETE FROM tracked_threads WHERE user_id = $1", user_id)
45	.execute(pool)
46	.await?;
47	Ok(())
48	}
49
50	/// Advance the read position for a tracked thread to `last_post_id`.
51	///
52	/// Monotonic: the position only ever moves forward in thread order. The caller
53	/// (`routes::forum::thread`) bumps to the last post on the page being viewed,
54	/// so an unguarded write would drag the position backward whenever a reader
55	/// revisits an earlier page, and `list_tracked_threads` would then re-count
56	/// already-read posts as unread. The two writes are also spawned off the
57	/// response path, so concurrent views of different pages race; keeping the
58	/// comparison in SQL means Postgres re-evaluates it against the winning row
59	/// under read-committed rather than letting the last writer win.
60	///
61	/// A `last_post_id` that names no row leaves the position untouched: the
62	/// scalar subquery yields NULL, the comparison is not true, and the guard
63	/// fails closed.
64	#[tracing::instrument(skip_all)]
65	pub async fn update_read_position(
66	pool: &PgPool,
67	user_id: Uuid,
68	thread_id: Uuid,
69	last_post_id: Uuid,
70	) -> Result<(), sqlx::Error> {
71	sqlx::query!(
72	"UPDATE tracked_threads SET last_read_post_id = $3
73	WHERE user_id = $1 AND thread_id = $2
74	AND (last_read_post_id IS NULL
75	OR (SELECT np.created_at FROM posts np WHERE np.id = $3)
76	> (SELECT cp.created_at FROM posts cp WHERE cp.id = last_read_post_id))",
77	user_id,
78	thread_id,
79	last_post_id,
80	)
81	.execute(pool)
82	.await?;
83	Ok(())
84	}
85