Skip to main content

max / makenotwork

12.8 KB · 452 lines History Blame Raw
1 //! Issue tracker queries: issues, comments, labels.
2
3 use sqlx::PgPool;
4
5 use super::models::{DbIssue, DbIssueComment, DbIssueCommentWithAuthor, DbIssueWithMeta};
6 use super::{GitRepoId, IssueId, IssueStatus, UserId};
7 use crate::error::Result;
8
9 // ── Issues ──
10
11 /// Create a new issue, assigning the next sequential number for the repo.
12 /// Retries once on unique constraint violation (concurrent insert race).
13 #[tracing::instrument(skip_all)]
14 pub async fn create_issue(
15 pool: &PgPool,
16 repo_id: GitRepoId,
17 author_id: UserId,
18 title: &str,
19 body_md: &str,
20 body_html: &str,
21 ) -> Result<DbIssue> {
22 let result = try_create_issue(pool, repo_id, author_id, title, body_md, body_html).await;
23 match result {
24 Ok(issue) => Ok(issue),
25 // Retry once ONLY on a unique violation: concurrent `MAX(number)+1`
26 // assignment can collide on the `(repo_id, number)` index. Any other
27 // error (FK violation, timeout, pool exhaustion) is not helped by an
28 // immediate retry and should surface (audit Run 13 Conc).
29 Err(e) if crate::helpers::is_unique_violation(&e) => {
30 try_create_issue(pool, repo_id, author_id, title, body_md, body_html).await
31 }
32 Err(e) => Err(e),
33 }
34 }
35
36 async fn try_create_issue(
37 pool: &PgPool,
38 repo_id: GitRepoId,
39 author_id: UserId,
40 title: &str,
41 body_md: &str,
42 body_html: &str,
43 ) -> Result<DbIssue> {
44 let issue = sqlx::query_as::<_, DbIssue>(
45 r"
46 INSERT INTO issues (repo_id, number, author_user_id, title, body_markdown, body_html)
47 VALUES ($1, (SELECT COALESCE(MAX(number), 0) + 1 FROM issues WHERE repo_id = $1), $2, $3, $4, $5)
48 RETURNING *
49 ",
50 )
51 .bind(repo_id)
52 .bind(author_id)
53 .bind(title)
54 .bind(body_md)
55 .bind(body_html)
56 .fetch_one(pool)
57 .await?;
58
59 Ok(issue)
60 }
61
62 /// Get a single issue by its primary key.
63 #[tracing::instrument(skip_all)]
64 pub async fn get_issue_by_id(pool: &PgPool, issue_id: IssueId) -> Result<Option<DbIssue>> {
65 let issue = sqlx::query_as::<_, DbIssue>("SELECT * FROM issues WHERE id = $1")
66 .bind(issue_id)
67 .fetch_optional(pool)
68 .await?;
69 Ok(issue)
70 }
71
72 /// Get a single issue by its repo-scoped number.
73 #[tracing::instrument(skip_all)]
74 pub async fn get_issue_by_number(
75 pool: &PgPool,
76 repo_id: GitRepoId,
77 number: i32,
78 ) -> Result<Option<DbIssue>> {
79 let issue =
80 sqlx::query_as::<_, DbIssue>("SELECT * FROM issues WHERE repo_id = $1 AND number = $2")
81 .bind(repo_id)
82 .bind(number)
83 .fetch_optional(pool)
84 .await?;
85
86 Ok(issue)
87 }
88
89 /// Fetch multiple issues by number in a single query, keyed by number. Used by
90 /// the git push-refs handler so a push touching many commits doesn't run one
91 /// `get_issue_by_number` per commit-reference (N+1).
92 #[tracing::instrument(skip_all)]
93 pub async fn get_issues_by_numbers(
94 pool: &PgPool,
95 repo_id: GitRepoId,
96 numbers: &[i32],
97 ) -> Result<std::collections::HashMap<i32, DbIssue>> {
98 if numbers.is_empty() {
99 return Ok(std::collections::HashMap::new());
100 }
101 let issues = sqlx::query_as::<_, DbIssue>(
102 "SELECT * FROM issues WHERE repo_id = $1 AND number = ANY($2)",
103 )
104 .bind(repo_id)
105 .bind(numbers)
106 .fetch_all(pool)
107 .await?;
108 Ok(issues.into_iter().map(|i| (i.number, i)).collect())
109 }
110
111 /// List issues with author username and comment count. Returns (issues, total_count).
112 #[tracing::instrument(skip_all)]
113 pub async fn list_issues(
114 pool: &PgPool,
115 repo_id: GitRepoId,
116 status: Option<IssueStatus>,
117 search: Option<&str>,
118 page: i64,
119 per_page: i64,
120 ) -> Result<(Vec<DbIssueWithMeta>, i64)> {
121 let offset = (page - 1) * per_page;
122 let status_str = status.map(|s| s.to_string());
123 let search_pattern = search.map(|s| {
124 format!(
125 "%{}%",
126 s.replace('\\', "\\\\")
127 .replace('%', "\\%")
128 .replace('_', "\\_")
129 )
130 });
131
132 let issues = sqlx::query_as::<_, DbIssueWithMeta>(
133 r"
134 SELECT
135 i.id, i.repo_id, i.number, i.author_user_id, i.title,
136 i.status, i.created_at, i.updated_at,
137 u.username AS author_username,
138 (SELECT COUNT(*) FROM issue_comments WHERE issue_id = i.id) AS comment_count
139 FROM issues i
140 JOIN users u ON u.id = i.author_user_id
141 WHERE i.repo_id = $1
142 AND ($2::TEXT IS NULL OR i.status = $2)
143 AND ($3::TEXT IS NULL OR i.title ILIKE $3 OR i.body_markdown ILIKE $3)
144 ORDER BY i.created_at DESC
145 LIMIT $4 OFFSET $5
146 ",
147 )
148 .bind(repo_id)
149 .bind(&status_str)
150 .bind(&search_pattern)
151 .bind(per_page)
152 .bind(offset)
153 .fetch_all(pool)
154 .await?;
155
156 let total: i64 = sqlx::query_scalar(
157 r"
158 SELECT COUNT(*)
159 FROM issues
160 WHERE repo_id = $1
161 AND ($2::TEXT IS NULL OR status = $2)
162 AND ($3::TEXT IS NULL OR title ILIKE $3 OR body_markdown ILIKE $3)
163 ",
164 )
165 .bind(repo_id)
166 .bind(&status_str)
167 .bind(&search_pattern)
168 .fetch_one(pool)
169 .await?;
170
171 Ok((issues, total))
172 }
173
174 /// Update an issue's title and body.
175 #[tracing::instrument(skip_all)]
176 pub async fn update_issue(
177 pool: &PgPool,
178 issue_id: IssueId,
179 title: &str,
180 body_md: &str,
181 body_html: &str,
182 ) -> Result<()> {
183 sqlx::query(
184 "UPDATE issues SET title = $2, body_markdown = $3, body_html = $4, updated_at = NOW() WHERE id = $1",
185 )
186 .bind(issue_id)
187 .bind(title)
188 .bind(body_md)
189 .bind(body_html)
190 .execute(pool)
191 .await?;
192
193 Ok(())
194 }
195
196 /// Update an issue's status (open/closed).
197 #[tracing::instrument(skip_all)]
198 pub async fn update_issue_status(
199 pool: &PgPool,
200 issue_id: IssueId,
201 status: IssueStatus,
202 ) -> Result<()> {
203 sqlx::query("UPDATE issues SET status = $2, updated_at = NOW() WHERE id = $1")
204 .bind(issue_id)
205 .bind(status)
206 .execute(pool)
207 .await?;
208
209 Ok(())
210 }
211
212 /// Get (open_count, closed_count) for a repo.
213 #[tracing::instrument(skip_all)]
214 pub async fn get_issue_counts(pool: &PgPool, repo_id: GitRepoId) -> Result<(i64, i64)> {
215 let open: i64 =
216 sqlx::query_scalar("SELECT COUNT(*) FROM issues WHERE repo_id = $1 AND status = 'open'")
217 .bind(repo_id)
218 .fetch_one(pool)
219 .await?;
220
221 let closed: i64 =
222 sqlx::query_scalar("SELECT COUNT(*) FROM issues WHERE repo_id = $1 AND status = 'closed'")
223 .bind(repo_id)
224 .fetch_one(pool)
225 .await?;
226
227 Ok((open, closed))
228 }
229
230 // ── Comments ──
231
232 /// Add a comment to an issue.
233 #[tracing::instrument(skip_all)]
234 pub async fn create_comment(
235 pool: &PgPool,
236 issue_id: IssueId,
237 author_id: UserId,
238 body_md: &str,
239 body_html: &str,
240 ) -> Result<DbIssueComment> {
241 let comment = sqlx::query_as::<_, DbIssueComment>(
242 r"
243 INSERT INTO issue_comments (issue_id, author_user_id, body_markdown, body_html)
244 VALUES ($1, $2, $3, $4)
245 RETURNING *
246 ",
247 )
248 .bind(issue_id)
249 .bind(author_id)
250 .bind(body_md)
251 .bind(body_html)
252 .fetch_one(pool)
253 .await?;
254
255 // Touch the issue's updated_at
256 sqlx::query("UPDATE issues SET updated_at = NOW() WHERE id = $1")
257 .bind(issue_id)
258 .execute(pool)
259 .await?;
260
261 Ok(comment)
262 }
263
264 /// Bulk-insert issue comments in a single statement, then touch each affected
265 /// issue's `updated_at` once. The git-push processor would otherwise issue one
266 /// INSERT (and one updated_at UPDATE) per referenced commit, ~100-150 sequential
267 /// writes per push. All comments share one author.
268 #[tracing::instrument(skip_all)]
269 pub async fn create_comments(
270 pool: &PgPool,
271 author_id: UserId,
272 comments: &[(IssueId, String, String)],
273 ) -> Result<()> {
274 if comments.is_empty() {
275 return Ok(());
276 }
277 let issue_ids: Vec<sqlx::types::Uuid> =
278 comments.iter().map(|(id, _, _)| *id.as_uuid()).collect();
279 let bodies_md: Vec<&str> = comments.iter().map(|(_, md, _)| md.as_str()).collect();
280 let bodies_html: Vec<&str> = comments.iter().map(|(_, _, html)| html.as_str()).collect();
281
282 sqlx::query(
283 r"
284 INSERT INTO issue_comments (issue_id, author_user_id, body_markdown, body_html)
285 SELECT u.issue_id, $1, u.body_md, u.body_html
286 FROM UNNEST($2::uuid[], $3::text[], $4::text[]) AS u(issue_id, body_md, body_html)
287 ",
288 )
289 .bind(author_id)
290 .bind(&issue_ids)
291 .bind(&bodies_md)
292 .bind(&bodies_html)
293 .execute(pool)
294 .await?;
295
296 // Touch updated_at once per distinct affected issue.
297 let mut distinct = issue_ids;
298 distinct.sort_unstable();
299 distinct.dedup();
300 sqlx::query("UPDATE issues SET updated_at = NOW() WHERE id = ANY($1)")
301 .bind(&distinct)
302 .execute(pool)
303 .await?;
304
305 Ok(())
306 }
307
308 /// Bulk status update for the git-push processor. Each entry is `(issue, final
309 /// status)`; the caller resolves repeated references to one final state so each
310 /// issue appears at most once.
311 #[tracing::instrument(skip_all)]
312 pub async fn update_issue_statuses(
313 pool: &PgPool,
314 updates: &[(IssueId, IssueStatus)],
315 ) -> Result<()> {
316 if updates.is_empty() {
317 return Ok(());
318 }
319 let ids: Vec<sqlx::types::Uuid> = updates.iter().map(|(id, _)| *id.as_uuid()).collect();
320 let statuses: Vec<String> = updates.iter().map(|(_, s)| s.to_string()).collect();
321
322 sqlx::query(
323 r"
324 UPDATE issues AS i SET status = u.status, updated_at = NOW()
325 FROM UNNEST($1::uuid[], $2::text[]) AS u(id, status)
326 WHERE i.id = u.id
327 ",
328 )
329 .bind(&ids)
330 .bind(&statuses)
331 .execute(pool)
332 .await?;
333
334 Ok(())
335 }
336
337 /// List all comments on an issue with author usernames.
338 #[tracing::instrument(skip_all)]
339 pub async fn list_comments(
340 pool: &PgPool,
341 issue_id: IssueId,
342 ) -> Result<Vec<DbIssueCommentWithAuthor>> {
343 let comments = sqlx::query_as::<_, DbIssueCommentWithAuthor>(
344 r"
345 SELECT c.id, c.issue_id, c.author_user_id, c.body_markdown, c.body_html, c.created_at,
346 u.username AS author_username
347 FROM issue_comments c
348 JOIN users u ON u.id = c.author_user_id
349 WHERE c.issue_id = $1
350 ORDER BY c.created_at ASC
351 ",
352 )
353 .bind(issue_id)
354 .fetch_all(pool)
355 .await?;
356
357 Ok(comments)
358 }
359
360 // ── Issue participants (for email notifications) ──
361
362 /// Get all distinct participant user IDs for an issue (author + all comment authors).
363 #[tracing::instrument(skip_all)]
364 pub async fn get_issue_participants(pool: &PgPool, issue_id: IssueId) -> Result<Vec<UserId>> {
365 let ids = sqlx::query_scalar::<_, UserId>(
366 r"
367 SELECT DISTINCT author_user_id
368 FROM (
369 SELECT author_user_id FROM issues WHERE id = $1
370 UNION
371 SELECT author_user_id FROM issue_comments WHERE issue_id = $1
372 ) AS participants
373 ",
374 )
375 .bind(issue_id)
376 .fetch_all(pool)
377 .await?;
378
379 Ok(ids)
380 }
381
382 /// Record the Multithreaded forum thread that mirrors this issue. Best-effort:
383 /// log and proceed on DB error so the issue itself isn't lost.
384 #[tracing::instrument(skip_all)]
385 pub async fn set_mt_thread_id(
386 pool: &PgPool,
387 issue_id: IssueId,
388 mt_thread_id: uuid::Uuid,
389 ) -> Result<()> {
390 sqlx::query("UPDATE issues SET mt_thread_id = $2 WHERE id = $1")
391 .bind(issue_id)
392 .bind(mt_thread_id)
393 .execute(pool)
394 .await?;
395 Ok(())
396 }
397
398 /// Fetch the MT thread linked to an issue, if any.
399 #[tracing::instrument(skip_all)]
400 pub async fn get_mt_thread_id(pool: &PgPool, issue_id: IssueId) -> Result<Option<uuid::Uuid>> {
401 let row: Option<(Option<uuid::Uuid>,)> =
402 sqlx::query_as("SELECT mt_thread_id FROM issues WHERE id = $1")
403 .bind(issue_id)
404 .fetch_optional(pool)
405 .await?;
406 Ok(row.and_then(|r| r.0))
407 }
408
409 // ── Issue message ID mapping (for email threading) ──
410
411 /// Store a mapping from an email Message-ID to an issue.
412 #[tracing::instrument(skip_all)]
413 pub async fn insert_issue_message_id(
414 pool: &PgPool,
415 message_id: &str,
416 issue_id: IssueId,
417 ) -> Result<()> {
418 sqlx::query(
419 "INSERT INTO issue_message_ids (message_id, issue_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
420 )
421 .bind(message_id)
422 .bind(issue_id)
423 .execute(pool)
424 .await?;
425
426 Ok(())
427 }
428
429 /// Look up an issue ID by any of the given email Message-IDs.
430 #[tracing::instrument(skip_all)]
431 pub async fn get_issue_id_by_any_message_id(
432 pool: &PgPool,
433 message_ids: &[&str],
434 ) -> Result<Option<IssueId>> {
435 if message_ids.is_empty() {
436 return Ok(None);
437 }
438
439 let ids: Vec<String> = message_ids
440 .iter()
441 .map(std::string::ToString::to_string)
442 .collect();
443 let issue_id = sqlx::query_scalar::<_, IssueId>(
444 "SELECT issue_id FROM issue_message_ids WHERE message_id = ANY($1) LIMIT 1",
445 )
446 .bind(&ids)
447 .fetch_optional(pool)
448 .await?;
449
450 Ok(issue_id)
451 }
452