Skip to main content

max / goingson

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