Skip to main content

max / goingson

12.3 KB · 356 lines History Blame Raw
1 //! SQLite implementation of the ProblemRepository.
2 //!
3 //! Backs the Problems inbox: a mirror of problems pulled from external sources
4 //! (wam tickets, /audit and /fuzz findings), ranked by painhours and promoted by
5 //! hand into tasks.
6 //!
7 //! Two rules shape the SQL here. Ingest upserts on the provenance key and
8 //! touches only the columns the source owns, so a re-pull never undoes triage.
9 //! And ordering happens in Rust, not `ORDER BY`, because painhours depends on
10 //! the current time.
11
12 use async_trait::async_trait;
13 use chrono::{DateTime, Utc};
14 use goingson_core::{
15 CoreError, DbValue, NewProblem, ParseableEnum, Problem, ProblemFilter, ProblemId,
16 ProblemRepository, ProblemStatus, ProjectId, Result, TaskId, UserId,
17 };
18 use sqlx::SqlitePool;
19
20 use crate::utils::{
21 format_datetime, format_datetime_now, parse_datetime, parse_tags, parse_uuid, parse_uuid_opt,
22 };
23
24 const COLUMNS: &str = "id, source, source_ref, title, body, pain, scale, status, project_id, \
25 tags, promoted_task_id, created_at, updated_at, settled_at, last_seen_at";
26
27 /// Database row struct for Problem
28 #[derive(Debug, Clone, sqlx::FromRow)]
29 struct ProblemRow {
30 pub id: String,
31 pub source: String,
32 pub source_ref: String,
33 pub title: String,
34 pub body: String,
35 pub pain: i64,
36 pub scale: i64,
37 pub status: String,
38 pub project_id: Option<String>,
39 pub tags: String,
40 pub promoted_task_id: Option<String>,
41 pub created_at: String,
42 pub updated_at: String,
43 pub settled_at: Option<String>,
44 pub last_seen_at: String,
45 }
46
47 impl TryFrom<ProblemRow> for Problem {
48 type Error = CoreError;
49
50 fn try_from(row: ProblemRow) -> std::result::Result<Self, Self::Error> {
51 Ok(Problem {
52 id: parse_uuid(&row.id)?.into(),
53 source: row.source,
54 source_ref: row.source_ref,
55 title: row.title,
56 body: row.body,
57 // The 1-5 factors are clamped by the painhours crate on use, so a
58 // source that wrote something wild ranks as the nearest valid
59 // factor instead of failing the whole fetch.
60 pain: row.pain.clamp(0, 255) as u8,
61 scale: row.scale.clamp(0, 255) as u8,
62 status: ProblemStatus::from_str_or_default(&row.status),
63 project_id: parse_uuid_opt(row.project_id.as_deref())?.map(Into::into),
64 tags: parse_tags(&row.tags),
65 promoted_task_id: parse_uuid_opt(row.promoted_task_id.as_deref())?.map(Into::into),
66 created_at: parse_datetime(&row.created_at)?,
67 updated_at: parse_datetime(&row.updated_at)?,
68 settled_at: row.settled_at.as_deref().map(parse_datetime).transpose()?,
69 last_seen_at: parse_datetime(&row.last_seen_at)?,
70 })
71 }
72 }
73
74 /// SQLite-backed implementation of [`ProblemRepository`].
75 pub struct SqliteProblemRepository {
76 pool: SqlitePool,
77 }
78
79 impl SqliteProblemRepository {
80 /// Creates a new repository instance with the given connection pool.
81 #[tracing::instrument(skip_all)]
82 pub fn new(pool: SqlitePool) -> Self {
83 Self { pool }
84 }
85
86 /// Applies a status transition and returns the updated problem.
87 ///
88 /// Shared by [`ProblemRepository::promote`] and
89 /// [`ProblemRepository::set_status`], because both have to keep `status`,
90 /// `settled_at`, and `promoted_task_id` consistent: settling stamps the
91 /// freeze instant, reopening clears both it and the backlink.
92 async fn transition(
93 &self,
94 id: ProblemId,
95 user_id: UserId,
96 status: ProblemStatus,
97 task_id: Option<TaskId>,
98 ) -> Result<Option<Problem>> {
99 let settled_at = status.is_settled().then(format_datetime_now);
100
101 let result = sqlx::query(
102 r"
103 UPDATE problems
104 SET status = ?, settled_at = ?, promoted_task_id = ?, updated_at = ?
105 WHERE id = ? AND user_id = ?
106 ",
107 )
108 .bind(status.db_value())
109 .bind(&settled_at)
110 .bind(task_id.map(|t| t.to_string()))
111 .bind(format_datetime_now())
112 .bind(id.to_string())
113 .bind(user_id.to_string())
114 .execute(&self.pool)
115 .await
116 .map_err(CoreError::database)?;
117
118 if result.rows_affected() > 0 {
119 self.get_by_id(id, user_id).await
120 } else {
121 Ok(None)
122 }
123 }
124 }
125
126 #[async_trait]
127 impl ProblemRepository for SqliteProblemRepository {
128 #[tracing::instrument(skip_all)]
129 async fn list(&self, user_id: UserId, filter: &ProblemFilter) -> Result<Vec<Problem>> {
130 let mut sql = format!("SELECT {COLUMNS} FROM problems WHERE user_id = ?");
131 if filter.source.is_some() {
132 sql.push_str(" AND source = ?");
133 }
134 if filter.status.is_some() {
135 sql.push_str(" AND status = ?");
136 }
137 if filter.project_id.is_some() {
138 sql.push_str(" AND project_id = ?");
139 }
140
141 let mut query =
142 sqlx::query_as::<_, ProblemRow>(sqlx::AssertSqlSafe(sql)).bind(user_id.to_string());
143 if let Some(source) = &filter.source {
144 query = query.bind(source.clone());
145 }
146 if let Some(status) = filter.status {
147 query = query.bind(status.db_value());
148 }
149 if let Some(project_id) = filter.project_id {
150 query = query.bind(project_id.to_string());
151 }
152
153 let rows = query
154 .fetch_all(&self.pool)
155 .await
156 .map_err(CoreError::database)?;
157
158 let mut problems: Vec<Problem> = rows
159 .into_iter()
160 .map(Problem::try_from)
161 .collect::<Result<_>>()?;
162
163 // Highest painhours first, ties broken by newest. Sorted here rather
164 // than in SQL because the score depends on the current time.
165 problems.sort_by(|a, b| {
166 b.painhours()
167 .cmp(&a.painhours())
168 .then_with(|| b.created_at.cmp(&a.created_at))
169 });
170
171 Ok(problems)
172 }
173
174 #[tracing::instrument(skip_all)]
175 async fn get_by_id(&self, id: ProblemId, user_id: UserId) -> Result<Option<Problem>> {
176 let row = sqlx::query_as::<_, ProblemRow>(sqlx::AssertSqlSafe(format!(
177 "SELECT {COLUMNS} FROM problems WHERE id = ? AND user_id = ?"
178 )))
179 .bind(id.to_string())
180 .bind(user_id.to_string())
181 .fetch_optional(&self.pool)
182 .await
183 .map_err(CoreError::database)?;
184
185 row.map(Problem::try_from).transpose()
186 }
187
188 #[tracing::instrument(skip_all)]
189 async fn find_by_source_ref(
190 &self,
191 user_id: UserId,
192 source: &str,
193 source_ref: &str,
194 ) -> Result<Option<Problem>> {
195 let row = sqlx::query_as::<_, ProblemRow>(sqlx::AssertSqlSafe(format!(
196 "SELECT {COLUMNS} FROM problems WHERE user_id = ? AND source = ? AND source_ref = ?"
197 )))
198 .bind(user_id.to_string())
199 .bind(source)
200 .bind(source_ref)
201 .fetch_optional(&self.pool)
202 .await
203 .map_err(CoreError::database)?;
204
205 row.map(Problem::try_from).transpose()
206 }
207
208 #[tracing::instrument(skip_all)]
209 async fn ingest(&self, user_id: UserId, problem: NewProblem) -> Result<Problem> {
210 let tags_json = serde_json::to_string(&problem.tags).unwrap_or_else(|_| "[]".to_string());
211 let now = format_datetime_now();
212
213 // A source reporting the problem as fixed settles it, but only if the
214 // human has not already ruled on it: `status = 'Open'` in the CASE
215 // guards promoted and dismissed rows from being overwritten.
216 let (status, settled_at) = if problem.resolved_upstream {
217 (ProblemStatus::Resolved, Some(now.clone()))
218 } else {
219 (ProblemStatus::Open, None)
220 };
221
222 sqlx::query(
223 r"
224 INSERT INTO problems (
225 id, user_id, source, source_ref, title, body, pain, scale, status,
226 project_id, tags, created_at, updated_at, settled_at, last_seen_at
227 )
228 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
229 ON CONFLICT(user_id, source, source_ref) DO UPDATE SET
230 title = excluded.title,
231 body = excluded.body,
232 pain = excluded.pain,
233 scale = excluded.scale,
234 tags = excluded.tags,
235 created_at = excluded.created_at,
236 updated_at = excluded.updated_at,
237 last_seen_at = excluded.last_seen_at,
238 project_id = COALESCE(problems.project_id, excluded.project_id),
239 status = CASE
240 WHEN excluded.status = 'Resolved' AND problems.status = 'Open'
241 THEN 'Resolved'
242 ELSE problems.status
243 END,
244 settled_at = CASE
245 WHEN excluded.status = 'Resolved' AND problems.status = 'Open'
246 THEN excluded.settled_at
247 ELSE problems.settled_at
248 END
249 ",
250 )
251 .bind(ProblemId::new().to_string())
252 .bind(user_id.to_string())
253 .bind(&problem.source)
254 .bind(&problem.source_ref)
255 .bind(&problem.title)
256 .bind(&problem.body)
257 .bind(i64::from(problem.pain))
258 .bind(i64::from(problem.scale))
259 .bind(status.db_value())
260 .bind(problem.project_id.map(|p| p.to_string()))
261 .bind(&tags_json)
262 .bind(format_datetime(&problem.created_at))
263 .bind(format_datetime(&problem.updated_at))
264 .bind(&settled_at)
265 .bind(&now)
266 .execute(&self.pool)
267 .await
268 .map_err(CoreError::database)?;
269
270 // Re-read by provenance key rather than by the id above: on a conflict
271 // that id was discarded and the existing row kept its own.
272 self.find_by_source_ref(user_id, &problem.source, &problem.source_ref)
273 .await?
274 .ok_or_else(|| CoreError::internal("Failed to retrieve ingested problem"))
275 }
276
277 #[tracing::instrument(skip_all)]
278 async fn promote(
279 &self,
280 id: ProblemId,
281 user_id: UserId,
282 task_id: TaskId,
283 ) -> Result<Option<Problem>> {
284 self.transition(id, user_id, ProblemStatus::Promoted, Some(task_id))
285 .await
286 }
287
288 #[tracing::instrument(skip_all)]
289 async fn set_status(
290 &self,
291 id: ProblemId,
292 user_id: UserId,
293 status: ProblemStatus,
294 ) -> Result<Option<Problem>> {
295 // Reopening drops the backlink; a problem sent back to triage has no
296 // solution attached. Promoting through this path is a caller error
297 // (there is no task to link), so it lands as Promoted with no backlink
298 // rather than silently inventing one.
299 self.transition(id, user_id, status, None).await
300 }
301
302 #[tracing::instrument(skip_all)]
303 async fn set_project(
304 &self,
305 id: ProblemId,
306 user_id: UserId,
307 project_id: Option<ProjectId>,
308 ) -> Result<Option<Problem>> {
309 let result = sqlx::query(
310 r"
311 UPDATE problems SET project_id = ?, updated_at = ?
312 WHERE id = ? AND user_id = ?
313 ",
314 )
315 .bind(project_id.map(|p| p.to_string()))
316 .bind(format_datetime_now())
317 .bind(id.to_string())
318 .bind(user_id.to_string())
319 .execute(&self.pool)
320 .await
321 .map_err(CoreError::database)?;
322
323 if result.rows_affected() > 0 {
324 self.get_by_id(id, user_id).await
325 } else {
326 Ok(None)
327 }
328 }
329
330 #[tracing::instrument(skip_all)]
331 async fn delete(&self, id: ProblemId, user_id: UserId) -> Result<bool> {
332 let result = sqlx::query("DELETE FROM problems WHERE id = ? AND user_id = ?")
333 .bind(id.to_string())
334 .bind(user_id.to_string())
335 .execute(&self.pool)
336 .await
337 .map_err(CoreError::database)?;
338
339 Ok(result.rows_affected() > 0)
340 }
341
342 #[tracing::instrument(skip_all)]
343 async fn last_pulled_at(&self, user_id: UserId, source: &str) -> Result<Option<DateTime<Utc>>> {
344 let raw: Option<String> = sqlx::query_scalar(
345 "SELECT MAX(last_seen_at) FROM problems WHERE user_id = ? AND source = ?",
346 )
347 .bind(user_id.to_string())
348 .bind(source)
349 .fetch_one(&self.pool)
350 .await
351 .map_err(CoreError::database)?;
352
353 raw.as_deref().map(parse_datetime).transpose()
354 }
355 }
356