fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
24478 bytes
Raw
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Issues and pull requests (shared `issues` table), comments, and issue labels.
6
7use model::{Comment, Issue, IssueState, Label, PullRequest};
8use sqlx::any::AnyRow;
9use sqlx::{AssertSqlSafe, Row};
10
11use crate::{Store, StoreError, new_id, now_ms};
12
13/// The columns of `issues` in a fixed order.
14const ISSUE_COLUMNS: &str = "id, repo_id, number, author_id, title, body, state, is_pull, \
15 assignee_id, created_at, updated_at, closed_at, locked_at";
16
17/// Open/closed counts for a repo's issues or PRs.
18#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
19pub struct StateCounts {
20 /// Number of open items.
21 pub open: i64,
22 /// Number of closed items.
23 pub closed: i64,
24}
25
26impl Store {
27 /// Create an issue or PR, allocating the next per-repo number atomically.
28 ///
29 /// # Errors
30 ///
31 /// Returns [`StoreError::Query`] on a database failure.
32 pub async fn create_issue(
33 &self,
34 repo_id: &str,
35 author_id: &str,
36 title: &str,
37 body: &str,
38 is_pull: bool,
39 ) -> Result<Issue, StoreError> {
40 let mut tx = self.pool.begin().await?;
41 // Allocate the shared number: read then bump inside the transaction.
42 let row = sqlx::query("SELECT next_iid FROM repos WHERE id = $1")
43 .bind(repo_id)
44 .fetch_one(&mut *tx)
45 .await?;
46 let number: i64 = row.try_get("next_iid")?;
47 sqlx::query("UPDATE repos SET next_iid = $1 WHERE id = $2")
48 .bind(number + 1)
49 .bind(repo_id)
50 .execute(&mut *tx)
51 .await?;
52
53 let now = now_ms();
54 let issue = Issue {
55 id: new_id(),
56 repo_id: repo_id.to_string(),
57 number,
58 author_id: author_id.to_string(),
59 title: title.to_string(),
60 body: body.to_string(),
61 state: IssueState::Open,
62 is_pull,
63 assignee_id: None,
64 created_at: now,
65 updated_at: now,
66 closed_at: None,
67 locked_at: None,
68 };
69 let sql = "INSERT INTO issues \
70 (id, repo_id, number, author_id, title, body, state, is_pull, assignee_id, \
71 created_at, updated_at, closed_at) \
72 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)";
73 sqlx::query(sql)
74 .bind(&issue.id)
75 .bind(&issue.repo_id)
76 .bind(issue.number)
77 .bind(&issue.author_id)
78 .bind(&issue.title)
79 .bind(&issue.body)
80 .bind(issue.state.as_str())
81 .bind(issue.is_pull)
82 .bind(&issue.assignee_id)
83 .bind(issue.created_at)
84 .bind(issue.updated_at)
85 .bind(issue.closed_at)
86 .execute(&mut *tx)
87 .await?;
88 tx.commit().await?;
89 Ok(issue)
90 }
91
92 /// Fetch an issue/PR by repo and number, filtered by kind.
93 ///
94 /// # Errors
95 ///
96 /// Returns [`StoreError::Query`] on a database failure.
97 pub async fn issue_by_number(
98 &self,
99 repo_id: &str,
100 number: i64,
101 is_pull: bool,
102 ) -> Result<Option<Issue>, StoreError> {
103 let sql = format!(
104 "SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = $1 AND number = $2 AND is_pull = $3"
105 );
106 let row = sqlx::query(AssertSqlSafe(sql))
107 .bind(repo_id)
108 .bind(number)
109 .bind(is_pull)
110 .fetch_optional(&self.pool)
111 .await?;
112 Ok(row.as_ref().map(|r| self.map_issue(r)).transpose()?)
113 }
114
115 /// Fetch an issue or PR by repo and number, of either kind (for dependency
116 /// references, which cross issues and PRs).
117 ///
118 /// # Errors
119 ///
120 /// Returns [`StoreError::Query`] on a database failure.
121 pub async fn issue_by_number_any(
122 &self,
123 repo_id: &str,
124 number: i64,
125 ) -> Result<Option<Issue>, StoreError> {
126 let sql = format!("SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = $1 AND number = $2");
127 let row = sqlx::query(AssertSqlSafe(sql))
128 .bind(repo_id)
129 .bind(number)
130 .fetch_optional(&self.pool)
131 .await?;
132 Ok(row.as_ref().map(|r| self.map_issue(r)).transpose()?)
133 }
134
135 /// Fetch an issue/PR by id.
136 ///
137 /// # Errors
138 ///
139 /// Returns [`StoreError::Query`] on a database failure.
140 pub async fn issue_by_id(&self, id: &str) -> Result<Option<Issue>, StoreError> {
141 let sql = format!("SELECT {ISSUE_COLUMNS} FROM issues WHERE id = $1");
142 let row = sqlx::query(AssertSqlSafe(sql))
143 .bind(id)
144 .fetch_optional(&self.pool)
145 .await?;
146 Ok(row.as_ref().map(|r| self.map_issue(r)).transpose()?)
147 }
148
149 /// List a repo's issues or PRs, optionally filtered by state, newest first,
150 /// bounded to `limit` rows starting at `offset`.
151 ///
152 /// # Errors
153 ///
154 /// Returns [`StoreError::Query`] on a database failure.
155 pub async fn list_issues(
156 &self,
157 repo_id: &str,
158 is_pull: bool,
159 state: Option<IssueState>,
160 limit: i64,
161 offset: i64,
162 ) -> Result<Vec<Issue>, StoreError> {
163 // Placeholders after the fixed ones depend on whether a state filter is set.
164 let (state_clause, limit_ph, offset_ph) = if state.is_some() {
165 (" AND state = $3", "$4", "$5")
166 } else {
167 ("", "$3", "$4")
168 };
169 let sql = format!(
170 "SELECT {ISSUE_COLUMNS} FROM issues WHERE repo_id = $1 AND is_pull = $2{state_clause} \
171 ORDER BY number DESC LIMIT {limit_ph} OFFSET {offset_ph}"
172 );
173 let mut query = sqlx::query(AssertSqlSafe(sql)).bind(repo_id).bind(is_pull);
174 if let Some(s) = state {
175 query = query.bind(s.as_str());
176 }
177 let rows = query.bind(limit).bind(offset).fetch_all(&self.pool).await?;
178 rows.iter()
179 .map(|r| self.map_issue(r))
180 .collect::<Result<_, _>>()
181 .map_err(Into::into)
182 }
183
184 /// Open/closed counts for a repo's issues or PRs.
185 ///
186 /// # Errors
187 ///
188 /// Returns [`StoreError::Query`] on a database failure.
189 pub async fn issue_counts(
190 &self,
191 repo_id: &str,
192 is_pull: bool,
193 ) -> Result<StateCounts, StoreError> {
194 let sql = "SELECT state, COUNT(*) AS n FROM issues \
195 WHERE repo_id = $1 AND is_pull = $2 GROUP BY state";
196 let rows = sqlx::query(sql)
197 .bind(repo_id)
198 .bind(is_pull)
199 .fetch_all(&self.pool)
200 .await?;
201 let mut counts = StateCounts::default();
202 for row in &rows {
203 let state: String = row.try_get("state")?;
204 let n: i64 = row.try_get("n")?;
205 if state == "closed" {
206 counts.closed = n;
207 } else {
208 counts.open = n;
209 }
210 }
211 Ok(counts)
212 }
213
214 /// Set an issue/PR's open/closed state (stamping/clearing `closed_at`).
215 ///
216 /// # Errors
217 ///
218 /// Returns [`StoreError::Query`] on a database failure.
219 pub async fn set_issue_state(&self, id: &str, state: IssueState) -> Result<bool, StoreError> {
220 let now = now_ms();
221 let closed_at = matches!(state, IssueState::Closed).then_some(now);
222 let updated = sqlx::query(
223 "UPDATE issues SET state = $1, closed_at = $2, updated_at = $3 WHERE id = $4",
224 )
225 .bind(state.as_str())
226 .bind(closed_at)
227 .bind(now)
228 .bind(id)
229 .execute(&self.pool)
230 .await?
231 .rows_affected();
232 Ok(updated > 0)
233 }
234
235 /// Lock or unlock an issue/PR conversation.
236 ///
237 /// # Errors
238 ///
239 /// Returns [`StoreError::Query`] on a database failure.
240 pub async fn set_issue_locked(&self, id: &str, locked: bool) -> Result<bool, StoreError> {
241 let now = now_ms();
242 let locked_at = locked.then_some(now);
243 let updated =
244 sqlx::query("UPDATE issues SET locked_at = $1, updated_at = $2 WHERE id = $3")
245 .bind(locked_at)
246 .bind(now)
247 .bind(id)
248 .execute(&self.pool)
249 .await?
250 .rows_affected();
251 Ok(updated > 0)
252 }
253
254 /// Set (or clear) an issue/PR's assignee.
255 ///
256 /// # Errors
257 ///
258 /// Returns [`StoreError::Query`] on a database failure.
259 pub async fn set_issue_assignee(
260 &self,
261 id: &str,
262 assignee_id: Option<&str>,
263 ) -> Result<bool, StoreError> {
264 let updated =
265 sqlx::query("UPDATE issues SET assignee_id = $1, updated_at = $2 WHERE id = $3")
266 .bind(assignee_id)
267 .bind(now_ms())
268 .bind(id)
269 .execute(&self.pool)
270 .await?
271 .rows_affected();
272 Ok(updated > 0)
273 }
274
275 /// Edit an issue/PR's title and body.
276 ///
277 /// # Errors
278 ///
279 /// Returns [`StoreError::Query`] on a database failure.
280 pub async fn update_issue(
281 &self,
282 id: &str,
283 title: &str,
284 body: &str,
285 ) -> Result<bool, StoreError> {
286 let updated =
287 sqlx::query("UPDATE issues SET title = $1, body = $2, updated_at = $3 WHERE id = $4")
288 .bind(title)
289 .bind(body)
290 .bind(now_ms())
291 .bind(id)
292 .execute(&self.pool)
293 .await?
294 .rows_affected();
295 Ok(updated > 0)
296 }
297
298 // ---- Comments ----
299
300 /// Add a comment to an issue/PR.
301 ///
302 /// # Errors
303 ///
304 /// Returns [`StoreError::Query`] on a database failure.
305 pub async fn add_comment(
306 &self,
307 issue_id: &str,
308 author_id: &str,
309 body: &str,
310 ) -> Result<Comment, StoreError> {
311 let now = now_ms();
312 let comment = Comment {
313 id: new_id(),
314 issue_id: issue_id.to_string(),
315 author_id: author_id.to_string(),
316 body: body.to_string(),
317 created_at: now,
318 updated_at: now,
319 };
320 sqlx::query(
321 "INSERT INTO comments (id, issue_id, author_id, body, created_at, updated_at) \
322 VALUES ($1, $2, $3, $4, $5, $6)",
323 )
324 .bind(&comment.id)
325 .bind(&comment.issue_id)
326 .bind(&comment.author_id)
327 .bind(&comment.body)
328 .bind(comment.created_at)
329 .bind(comment.updated_at)
330 .execute(&self.pool)
331 .await?;
332 // Touch the issue so its "updated" reflects the new activity.
333 let _ = sqlx::query("UPDATE issues SET updated_at = $1 WHERE id = $2")
334 .bind(now)
335 .bind(issue_id)
336 .execute(&self.pool)
337 .await;
338 Ok(comment)
339 }
340
341 /// List an issue/PR's comments oldest-first.
342 ///
343 /// # Errors
344 ///
345 /// Returns [`StoreError::Query`] on a database failure.
346 pub async fn list_comments(&self, issue_id: &str) -> Result<Vec<Comment>, StoreError> {
347 let rows = sqlx::query(
348 "SELECT id, issue_id, author_id, body, created_at, updated_at FROM comments \
349 WHERE issue_id = $1 ORDER BY created_at ASC",
350 )
351 .bind(issue_id)
352 .fetch_all(&self.pool)
353 .await?;
354 rows.iter()
355 .map(|r| {
356 Ok(Comment {
357 id: r.try_get("id")?,
358 issue_id: r.try_get("issue_id")?,
359 author_id: r.try_get("author_id")?,
360 body: r.try_get("body")?,
361 created_at: r.try_get("created_at")?,
362 updated_at: r.try_get("updated_at")?,
363 })
364 })
365 .collect::<Result<_, sqlx::Error>>()
366 .map_err(Into::into)
367 }
368
369 /// Fetch a single comment by id (for edit authorization).
370 ///
371 /// # Errors
372 ///
373 /// Returns [`StoreError::Query`] on a database failure.
374 pub async fn comment_by_id(&self, id: &str) -> Result<Option<Comment>, StoreError> {
375 let row = sqlx::query(
376 "SELECT id, issue_id, author_id, body, created_at, updated_at FROM comments \
377 WHERE id = $1",
378 )
379 .bind(id)
380 .fetch_optional(&self.pool)
381 .await?;
382 row.map(|r| {
383 Ok(Comment {
384 id: r.try_get("id")?,
385 issue_id: r.try_get("issue_id")?,
386 author_id: r.try_get("author_id")?,
387 body: r.try_get("body")?,
388 created_at: r.try_get("created_at")?,
389 updated_at: r.try_get("updated_at")?,
390 })
391 })
392 .transpose()
393 .map_err(|e: sqlx::Error| e.into())
394 }
395
396 /// Edit a comment's body.
397 ///
398 /// # Errors
399 ///
400 /// Returns [`StoreError::Query`] on a database failure.
401 pub async fn update_comment(&self, id: &str, body: &str) -> Result<bool, StoreError> {
402 let updated = sqlx::query("UPDATE comments SET body = $1, updated_at = $2 WHERE id = $3")
403 .bind(body)
404 .bind(now_ms())
405 .bind(id)
406 .execute(&self.pool)
407 .await?
408 .rows_affected();
409 Ok(updated > 0)
410 }
411
412 // ---- Issue labels ----
413
414 /// The labels applied to an issue/PR.
415 ///
416 /// # Errors
417 ///
418 /// Returns [`StoreError::Query`] on a database failure.
419 pub async fn issue_labels(&self, issue_id: &str) -> Result<Vec<Label>, StoreError> {
420 let sql = "SELECT l.id, l.repo_id, l.name, l.color, l.description, l.created_at \
421 FROM issue_labels il JOIN labels l ON l.id = il.label_id \
422 WHERE il.issue_id = $1 ORDER BY l.name_lower ASC";
423 let rows = sqlx::query(sql)
424 .bind(issue_id)
425 .fetch_all(&self.pool)
426 .await?;
427 rows.iter()
428 .map(|r| {
429 Ok(Label {
430 id: r.try_get("id")?,
431 repo_id: r.try_get("repo_id")?,
432 name: r.try_get("name")?,
433 color: r.try_get("color")?,
434 description: r.try_get("description")?,
435 created_at: r.try_get("created_at")?,
436 })
437 })
438 .collect::<Result<_, sqlx::Error>>()
439 .map_err(Into::into)
440 }
441
442 /// Add a label to an issue/PR (idempotent).
443 ///
444 /// # Errors
445 ///
446 /// Returns [`StoreError::Query`] on a database failure.
447 pub async fn add_issue_label(&self, issue_id: &str, label_id: &str) -> Result<(), StoreError> {
448 sqlx::query(
449 "INSERT INTO issue_labels (issue_id, label_id) VALUES ($1, $2) \
450 ON CONFLICT (issue_id, label_id) DO NOTHING",
451 )
452 .bind(issue_id)
453 .bind(label_id)
454 .execute(&self.pool)
455 .await?;
456 Ok(())
457 }
458
459 /// Remove a label from an issue/PR.
460 ///
461 /// # Errors
462 ///
463 /// Returns [`StoreError::Query`] on a database failure.
464 pub async fn remove_issue_label(
465 &self,
466 issue_id: &str,
467 label_id: &str,
468 ) -> Result<(), StoreError> {
469 sqlx::query("DELETE FROM issue_labels WHERE issue_id = $1 AND label_id = $2")
470 .bind(issue_id)
471 .bind(label_id)
472 .execute(&self.pool)
473 .await?;
474 Ok(())
475 }
476
477 // ---- Pull requests ----
478
479 /// Open a pull request: allocate the shared number, insert the `issues` row
480 /// (`is_pull = true`), and record its base/head refs, all in one transaction.
481 ///
482 /// # Errors
483 ///
484 /// Returns [`StoreError::Query`] on a database failure.
485 pub async fn create_pull(
486 &self,
487 repo_id: &str,
488 author_id: &str,
489 title: &str,
490 body: &str,
491 base_ref: &str,
492 head_ref: &str,
493 ) -> Result<Issue, StoreError> {
494 let mut tx = self.pool.begin().await?;
495 let row = sqlx::query("SELECT next_iid FROM repos WHERE id = $1")
496 .bind(repo_id)
497 .fetch_one(&mut *tx)
498 .await?;
499 let number: i64 = row.try_get("next_iid")?;
500 sqlx::query("UPDATE repos SET next_iid = $1 WHERE id = $2")
501 .bind(number + 1)
502 .bind(repo_id)
503 .execute(&mut *tx)
504 .await?;
505
506 let now = now_ms();
507 let issue = Issue {
508 id: new_id(),
509 repo_id: repo_id.to_string(),
510 number,
511 author_id: author_id.to_string(),
512 title: title.to_string(),
513 body: body.to_string(),
514 state: IssueState::Open,
515 is_pull: true,
516 assignee_id: None,
517 created_at: now,
518 updated_at: now,
519 closed_at: None,
520 locked_at: None,
521 };
522 sqlx::query(
523 "INSERT INTO issues \
524 (id, repo_id, number, author_id, title, body, state, is_pull, assignee_id, \
525 created_at, updated_at, closed_at) \
526 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
527 )
528 .bind(&issue.id)
529 .bind(&issue.repo_id)
530 .bind(issue.number)
531 .bind(&issue.author_id)
532 .bind(&issue.title)
533 .bind(&issue.body)
534 .bind(issue.state.as_str())
535 .bind(issue.is_pull)
536 .bind(&issue.assignee_id)
537 .bind(issue.created_at)
538 .bind(issue.updated_at)
539 .bind(issue.closed_at)
540 .execute(&mut *tx)
541 .await?;
542 sqlx::query("INSERT INTO pull_requests (issue_id, base_ref, head_ref) VALUES ($1, $2, $3)")
543 .bind(&issue.id)
544 .bind(base_ref)
545 .bind(head_ref)
546 .execute(&mut *tx)
547 .await?;
548 tx.commit().await?;
549 Ok(issue)
550 }
551
552 /// Fetch the pull-request row for an issue id, if it is a PR.
553 ///
554 /// # Errors
555 ///
556 /// Returns [`StoreError::Query`] on a database failure.
557 pub async fn pull_by_issue(&self, issue_id: &str) -> Result<Option<PullRequest>, StoreError> {
558 let row = sqlx::query(
559 "SELECT issue_id, base_ref, head_ref, merged_at, merged_by, merge_sha, merge_strategy \
560 FROM pull_requests WHERE issue_id = $1",
561 )
562 .bind(issue_id)
563 .fetch_optional(&self.pool)
564 .await?;
565 row.map(|r| {
566 Ok(PullRequest {
567 issue_id: r.try_get("issue_id")?,
568 base_ref: r.try_get("base_ref")?,
569 head_ref: r.try_get("head_ref")?,
570 merged_at: r.try_get("merged_at")?,
571 merged_by: r.try_get("merged_by")?,
572 merge_sha: r.try_get("merge_sha")?,
573 merge_strategy: r.try_get("merge_strategy")?,
574 })
575 })
576 .transpose()
577 .map_err(|e: sqlx::Error| e.into())
578 }
579
580 /// Record a merge: stamp the pull-request row and close the issue, atomically.
581 ///
582 /// # Errors
583 ///
584 /// Returns [`StoreError::Query`] on a database failure.
585 pub async fn mark_pull_merged(
586 &self,
587 issue_id: &str,
588 merged_by: &str,
589 merge_sha: &str,
590 strategy: &str,
591 ) -> Result<(), StoreError> {
592 let now = now_ms();
593 let mut tx = self.pool.begin().await?;
594 sqlx::query(
595 "UPDATE pull_requests \
596 SET merged_at = $1, merged_by = $2, merge_sha = $3, merge_strategy = $4 \
597 WHERE issue_id = $5",
598 )
599 .bind(now)
600 .bind(merged_by)
601 .bind(merge_sha)
602 .bind(strategy)
603 .bind(issue_id)
604 .execute(&mut *tx)
605 .await?;
606 sqlx::query(
607 "UPDATE issues SET state = 'closed', closed_at = $1, updated_at = $1 WHERE id = $2",
608 )
609 .bind(now)
610 .bind(issue_id)
611 .execute(&mut *tx)
612 .await?;
613 tx.commit().await?;
614 Ok(())
615 }
616
617 // ---- Dependencies ----
618
619 /// Record that `issue_id` is blocked by `depends_on_id` (idempotent).
620 ///
621 /// # Errors
622 ///
623 /// Returns [`StoreError::Query`] on a database failure.
624 pub async fn add_dependency(
625 &self,
626 issue_id: &str,
627 depends_on_id: &str,
628 ) -> Result<(), StoreError> {
629 sqlx::query(
630 "INSERT INTO issue_dependencies (issue_id, depends_on_id, created_at) \
631 VALUES ($1, $2, $3) ON CONFLICT (issue_id, depends_on_id) DO NOTHING",
632 )
633 .bind(issue_id)
634 .bind(depends_on_id)
635 .bind(now_ms())
636 .execute(&self.pool)
637 .await?;
638 Ok(())
639 }
640
641 /// Remove a dependency edge.
642 ///
643 /// # Errors
644 ///
645 /// Returns [`StoreError::Query`] on a database failure.
646 pub async fn remove_dependency(
647 &self,
648 issue_id: &str,
649 depends_on_id: &str,
650 ) -> Result<(), StoreError> {
651 sqlx::query("DELETE FROM issue_dependencies WHERE issue_id = $1 AND depends_on_id = $2")
652 .bind(issue_id)
653 .bind(depends_on_id)
654 .execute(&self.pool)
655 .await?;
656 Ok(())
657 }
658
659 /// The issues/PRs that `issue_id` is blocked by (its dependencies).
660 ///
661 /// # Errors
662 ///
663 /// Returns [`StoreError::Query`] on a database failure.
664 pub async fn dependencies_of(&self, issue_id: &str) -> Result<Vec<Issue>, StoreError> {
665 self.related_issues(
666 "SELECT i.id, i.repo_id, i.number, i.author_id, i.title, i.body, i.state, \
667 i.is_pull, i.assignee_id, i.created_at, i.updated_at, i.closed_at, i.locked_at \
668 FROM issue_dependencies d JOIN issues i ON i.id = d.depends_on_id \
669 WHERE d.issue_id = $1 ORDER BY i.number ASC",
670 issue_id,
671 )
672 .await
673 }
674
675 /// The issues/PRs blocked by `issue_id` (its dependents).
676 ///
677 /// # Errors
678 ///
679 /// Returns [`StoreError::Query`] on a database failure.
680 pub async fn dependents_of(&self, issue_id: &str) -> Result<Vec<Issue>, StoreError> {
681 self.related_issues(
682 "SELECT i.id, i.repo_id, i.number, i.author_id, i.title, i.body, i.state, \
683 i.is_pull, i.assignee_id, i.created_at, i.updated_at, i.closed_at, i.locked_at \
684 FROM issue_dependencies d JOIN issues i ON i.id = d.issue_id \
685 WHERE d.depends_on_id = $1 ORDER BY i.number ASC",
686 issue_id,
687 )
688 .await
689 }
690
691 /// Count `issue_id`'s dependencies that are still open (i.e. blocking it).
692 ///
693 /// # Errors
694 ///
695 /// Returns [`StoreError::Query`] on a database failure.
696 pub async fn open_dependency_count(&self, issue_id: &str) -> Result<i64, StoreError> {
697 let row = sqlx::query(
698 "SELECT COUNT(*) AS n FROM issue_dependencies d JOIN issues i ON i.id = d.depends_on_id \
699 WHERE d.issue_id = $1 AND i.state = 'open'",
700 )
701 .bind(issue_id)
702 .fetch_one(&self.pool)
703 .await?;
704 Ok(row.try_get("n")?)
705 }
706
707 /// Run a dependency join query bound to one issue id and map the rows.
708 async fn related_issues(&self, sql: &str, issue_id: &str) -> Result<Vec<Issue>, StoreError> {
709 let rows = sqlx::query(AssertSqlSafe(sql.to_string()))
710 .bind(issue_id)
711 .fetch_all(&self.pool)
712 .await?;
713 rows.iter()
714 .map(|r| self.map_issue(r))
715 .collect::<Result<_, _>>()
716 .map_err(Into::into)
717 }
718
719 /// Map an `issues` row (selected via [`ISSUE_COLUMNS`]) to an [`Issue`].
720 fn map_issue(&self, row: &AnyRow) -> Result<Issue, sqlx::Error> {
721 Ok(Issue {
722 id: row.try_get("id")?,
723 repo_id: row.try_get("repo_id")?,
724 number: row.try_get("number")?,
725 author_id: row.try_get("author_id")?,
726 title: row.try_get("title")?,
727 body: row.try_get("body")?,
728 state: IssueState::from_token(&row.try_get::<String, _>("state")?),
729 is_pull: self.get_bool(row, "is_pull")?,
730 assignee_id: row.try_get("assignee_id")?,
731 created_at: row.try_get("created_at")?,
732 updated_at: row.try_get("updated_at")?,
733 closed_at: row.try_get("closed_at")?,
734 locked_at: row.try_get("locked_at")?,
735 })
736 }
737}