fix: use link header to get next issue page

GH pagination doesn't work anymore with large datasets

Author: McPatate

issue-bot/src/github.rs

@@ -202,21 +202,24 @@ impl GithubApi {
 
     pub(crate) fn get_issues(
         &self,
-        from_page: i32,
+        from_url: Option<String>,
         repo_data: RepositoryData,
-    ) -> impl Stream<Item = Result<(IssueWithComments, Option<i32>), GithubApiError>> + use<'_>
+    ) -> impl Stream<Item = Result<(IssueWithComments, Option<String>), GithubApiError>> + use<'_>
     {
         try_stream! {
-            let url = format!("https://api.github.com/repos/{}/issues", repo_data.full_name);
             let client = self.client.clone();
-            let mut page = from_page;
+            let mut url = if let Some(from_url) = from_url {
+                info!("resuming fetching issues from repo {} at {}", repo_data.full_name, from_url);
+                from_url
+            } else {
+                format!("https://api.github.com/repos/{}/issues", repo_data.full_name)
+            };
             loop {
                 let res = client
                     .get(&url)
                     .query(&[
                         ("state", "all"),
                         ("direction", "desc"),
-                        ("page", &page.to_string()),
                         ("per_page", "100"),
                     ])
                 .send()
@@ -236,8 +239,11 @@ impl GithubApi {
                         break;
                     }
                 };
-                info!("fetched {} issues from page {}, getting comments for each issue next", issues.len(), page);
+                info!("fetched {} issues from {}, getting comments for each issue next", issues.len(), url);
                 let page_issue_count = issues.len();
+                if let Some(next_url) = get_next_page(link_header.clone())? {
+                    url = next_url;
+                };
                 for (i, issue) in issues.into_iter().enumerate() {
                     loop {
                         let res = client
@@ -259,14 +265,13 @@ impl GithubApi {
                                 break;
                             }
                         };
-                        yield (IssueWithComments::new(issue, comments), (i + 1 == page_issue_count).then_some(page));
+                        yield (IssueWithComments::new(issue, comments), (i + 1 == page_issue_count).then_some(url.clone()));
                         break;
                     }
                 }
                 if get_next_page(link_header)?.is_none() {
                     break;
                 }
-                page += 1;
             }
         }
     }

issue-bot/src/main.rs

@@ -268,13 +268,15 @@ struct ClosestIssue {
 
 #[derive(Debug, Deserialize, Serialize)]
 enum JobData {
-    IssueIndexation { issues_page: i32 },
+    // FIXME: naming is a bit confusing, this means "repository issue indexation"
+    IssueIndexation { next_url: String },
     EmbeddingsRegeneration { current_issue: i32 },
 }
 
 #[derive(Debug, sqlx::Type)]
 #[sqlx(type_name = "job_type", rename_all = "snake_case")]
 enum JobType {
+    // FIXME: naming is a bit confusing, this means "repository issue indexation"
     IssueIndexation,
     EmbeddingsRegeneration,
 }
@@ -576,11 +578,11 @@ async fn handle_webhooks(
                         }
                     };
                     let from_issues_page =
-                        job.as_ref().and_then(|j| match j.data.0 { JobData::IssueIndexation { issues_page } => Some(issues_page + 1), _ => None}).unwrap_or(1);
+                        job.and_then(|j| match j.data.0 { JobData::IssueIndexation { next_url } => Some(next_url), _ => None});
                     let issues = github_api.get_issues(from_issues_page, repo_data.clone());
                     pin_mut!(issues);
                     while let Some(issue) = issues.next().await {
-                        let (issue, page) = match issue {
+                        let (issue, next_url) = match issue {
                             Ok(issue) => issue,
                             Err(err) => {
                                 error!(err = err.to_string(), "error fetching next item from issues stream");
@@ -664,7 +666,7 @@ async fn handle_webhooks(
                                 error!(issue_number = issue.number, err = err.to_string(), "error inserting comments");
                             }
                         }
-                        if let Some(page) = page {
+                        if let Some(next_url) = next_url {
                             if let Err(err) = sqlx::query(
                                 r#"insert into jobs (data, job_type, repository_full_name)
                                values ($1, $2, $3)
@@ -675,7 +677,7 @@ async fn handle_webhooks(
                                    updated_at = current_timestamp"#,
                             )
                             .bind(Json(JobData::IssueIndexation {
-                                issues_page: page,
+                                next_url,
                             }))
                             .bind(JobType::IssueIndexation)
                             .bind(&repo_data.full_name)