Skip to main content

max / goingson

6.4 KB · 205 lines History Blame Raw
1 //! Status-token repository methods for SqliteTaskRepository.
2 //!
3 //! A task carries an ordered list of status tokens (typed at-a-glance markers); at
4 //! most one is the `is_primary`, the token that resolved it. Token ids are
5 //! deterministic in `(task_id, kind, reference)` (see
6 //! [`StatusToken::deterministic_id`]), so recording a token is idempotent, locally
7 //! via `ON CONFLICT(id)` and across devices via the sync changelog's id-keyed upsert.
8
9 use sqlx::SqlitePool;
10 use std::collections::HashMap;
11
12 use goingson_core::{
13 CoreError, DbValue, ParseableEnum, Result, StatusToken, StatusTokenId, TaskId, TokenState,
14 UserId,
15 };
16
17 use crate::utils::{bind_placeholders, parse_uuid};
18
19 /// Row struct for task_status_tokens from SQLite.
20 #[derive(Debug, Clone, sqlx::FromRow)]
21 pub(crate) struct StatusTokenRow {
22 pub id: String,
23 pub task_id: String,
24 pub kind: String,
25 pub reference: String,
26 pub state: String,
27 pub is_primary: i32,
28 pub position: i32,
29 }
30
31 impl TryFrom<StatusTokenRow> for StatusToken {
32 type Error = CoreError;
33
34 fn try_from(row: StatusTokenRow) -> std::result::Result<Self, Self::Error> {
35 Ok(StatusToken {
36 id: parse_uuid(&row.id)?.into(),
37 task_id: parse_uuid(&row.task_id)?.into(),
38 kind: row.kind,
39 reference: row.reference,
40 state: TokenState::from_str_or_default(&row.state),
41 is_primary: row.is_primary != 0,
42 position: row.position,
43 })
44 }
45 }
46
47 /// Batch-fetch status tokens for multiple tasks by their IDs.
48 pub(crate) async fn get_tokens_for_tasks(
49 pool: &SqlitePool,
50 task_ids: &[String],
51 ) -> Result<HashMap<TaskId, Vec<StatusToken>>> {
52 if task_ids.is_empty() {
53 return Ok(HashMap::new());
54 }
55
56 let query = format!(
57 r"
58 SELECT id, task_id, kind, reference, state, is_primary, position
59 FROM task_status_tokens
60 WHERE task_id IN ({})
61 ORDER BY position ASC, created_at ASC
62 ",
63 bind_placeholders(task_ids.len())
64 );
65
66 let mut q = sqlx::query_as::<_, StatusTokenRow>(sqlx::AssertSqlSafe(query.as_str()));
67 for id in task_ids {
68 q = q.bind(id);
69 }
70
71 let rows = q.fetch_all(pool).await.map_err(CoreError::database)?;
72
73 let mut map: HashMap<TaskId, Vec<StatusToken>> = HashMap::new();
74 for row in rows {
75 let token = StatusToken::try_from(row)?;
76 map.entry(token.task_id).or_default().push(token);
77 }
78
79 Ok(map)
80 }
81
82 /// Get all status tokens for a single task, in append order.
83 pub(crate) async fn get_tokens_for_task(
84 pool: &SqlitePool,
85 task_id: TaskId,
86 ) -> Result<Vec<StatusToken>> {
87 let rows = sqlx::query_as::<_, StatusTokenRow>(
88 r"
89 SELECT id, task_id, kind, reference, state, is_primary, position
90 FROM task_status_tokens
91 WHERE task_id = ?
92 ORDER BY position ASC, created_at ASC
93 ",
94 )
95 .bind(task_id.to_string())
96 .fetch_all(pool)
97 .await
98 .map_err(CoreError::database)?;
99
100 rows.into_iter().map(StatusToken::try_from).collect()
101 }
102
103 /// Record (upsert) a status token on a task (verifies task ownership).
104 ///
105 /// Idempotent in `(task_id, kind, reference)` via the deterministic id: a re-record
106 /// updates `state`/`is_primary` in place. When `is_primary` is set, any prior primary
107 /// flag on the task is cleared first so at most one token is ever primary. Returns
108 /// `None` when the task is missing or not owned by `user_id`.
109 pub(crate) async fn record_token(
110 pool: &SqlitePool,
111 task_id: TaskId,
112 user_id: UserId,
113 kind: &str,
114 reference: &str,
115 state: TokenState,
116 is_primary: bool,
117 ) -> Result<Option<StatusToken>> {
118 let mut tx = pool.begin().await.map_err(CoreError::database)?;
119
120 let task_exists: (i64,) =
121 sqlx::query_as("SELECT COUNT(*) FROM tasks WHERE id = ? AND user_id = ?")
122 .bind(task_id.to_string())
123 .bind(user_id.to_string())
124 .fetch_one(&mut *tx)
125 .await
126 .map_err(CoreError::database)?;
127 if task_exists.0 == 0 {
128 return Ok(None);
129 }
130
131 // At most one primary per task: clear the flag on every other token first. The
132 // deterministic id below means the current token, if it already exists, is the
133 // one row this WHERE would also touch; harmless, its flag is set by the upsert.
134 if is_primary {
135 sqlx::query(
136 "UPDATE task_status_tokens SET is_primary = 0 WHERE task_id = ? AND is_primary = 1",
137 )
138 .bind(task_id.to_string())
139 .execute(&mut *tx)
140 .await
141 .map_err(CoreError::database)?;
142 }
143
144 let id = StatusToken::deterministic_id(task_id, kind, reference);
145 let primary_int = i32::from(is_primary);
146
147 // Insert with the next position, or, if this token was already recorded, keep its
148 // position and update state/is_primary. Position is assigned inside the statement
149 // so a concurrent record can't race the MAX (SQLite serializes writers).
150 let (position,): (i32,) = sqlx::query_as(
151 r"
152 INSERT INTO task_status_tokens (id, task_id, kind, reference, state, is_primary, position, group_id)
153 SELECT ?, ?, ?, ?, ?, ?, COALESCE(MAX(position), -1) + 1, (SELECT group_id FROM tasks WHERE id = ?) FROM task_status_tokens WHERE task_id = ?
154 ON CONFLICT(id) DO UPDATE SET state = excluded.state, is_primary = excluded.is_primary
155 RETURNING position
156 ",
157 )
158 .bind(id.to_string())
159 .bind(task_id.to_string())
160 .bind(kind)
161 .bind(reference)
162 .bind(state.db_value())
163 .bind(primary_int)
164 // Inherit the parent task's group scope.
165 .bind(task_id.to_string())
166 .bind(task_id.to_string())
167 .fetch_one(&mut *tx)
168 .await
169 .map_err(CoreError::database)?;
170
171 tx.commit().await.map_err(CoreError::database)?;
172
173 Ok(Some(StatusToken {
174 id,
175 task_id,
176 kind: kind.to_string(),
177 reference: reference.to_string(),
178 state,
179 is_primary,
180 position,
181 }))
182 }
183
184 /// Delete a status token by id, scoped to the user's own tasks.
185 pub(crate) async fn delete_token(
186 pool: &SqlitePool,
187 token_id: StatusTokenId,
188 user_id: UserId,
189 ) -> Result<bool> {
190 let result = sqlx::query(
191 r"
192 DELETE FROM task_status_tokens
193 WHERE id = ?
194 AND task_id IN (SELECT id FROM tasks WHERE user_id = ?)
195 ",
196 )
197 .bind(token_id.to_string())
198 .bind(user_id.to_string())
199 .execute(pool)
200 .await
201 .map_err(CoreError::database)?;
202
203 Ok(result.rows_affected() > 0)
204 }
205