refactor: jobs table to contain jsonb data col

Author: McPatate

init_db.sql

@@ -36,7 +36,7 @@ CREATE INDEX IF NOT EXISTS issues_embedding_hnsw_idx ON issues USING hnsw (embed
 CREATE TABLE IF NOT EXISTS jobs (
   id SERIAL PRIMARY KEY,
   repository_id VARCHAR NOT NULL UNIQUE,
-  page INT NOT NULL,
+  data JSONB NOT NULL,
   created_at timestamp with time zone NOT NULL DEFAULT (current_timestamp AT TIME ZONE 'UTC'),
   updated_at timestamp with time zone NOT NULL DEFAULT (current_timestamp AT TIME ZONE 'UTC')
 );

issue-bot/Cargo.toml

@@ -24,7 +24,7 @@ once_cell = "1.20"
 pgvector = { version = "0.4", features = ["sqlx"] }
 reqwest = { version = "0.12", features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }
-serde_json = "1"
+serde_json = { version = "1", features = ["raw_value"] }
 sha2 = "0.10"
 sqlx = { version = "0.8", features = ["chrono", "postgres", "runtime-tokio"] }
 thiserror = "2"

issue-bot/src/github.rs

@@ -42,8 +42,9 @@ pub enum GithubApiError {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(unused)]
 struct PullRequest {
-    #[allow(unused)]
+    html_url: String,
     url: String,
 }
 
@@ -180,13 +181,13 @@ impl GithubApi {
 
     pub(crate) fn get_issues(
         &self,
-        from_page: Option<i32>,
+        from_page: i32,
         repository: RepositoryData,
     ) -> impl Stream<Item = Result<(IssueWithComments, Option<i32>), GithubApiError>> + use<'_> {
         try_stream! {
             let url = format!("https://api.github.com/repos/{}/issues", repository.repo_id);
             let client = self.client.clone();
-            let mut page = from_page.unwrap_or(1);
+            let mut page = from_page;
             loop {
                 let res = client
                     .get(&url)
@@ -206,7 +207,6 @@ impl GithubApi {
                 if get_next_page(link_header)?.is_none() {
                     break;
                 }
-                let issues: Vec<Issue> = issues.into_iter().filter(|i| i.pull_request.is_none()).collect();
                 let page_issue_count = issues.len();
                 for (i, issue) in issues.into_iter().enumerate() {
                     let res = client
@@ -226,14 +226,6 @@ impl GithubApi {
             }
         }
     }
-
-    // pub(crate) async fn get_prs(
-    //     &self,
-    //     repository: &RepositoryData,
-    // ) -> Result<Vec<IssueWithComments>, GithubApiError> {
-    //     let issues = Vec::new();
-    //     Ok(issues)
-    // }
 }
 
 async fn handle_ratelimit(remaining: Option<HeaderValue>, reset: Option<HeaderValue>) -> Result<(), GithubApiError> {

issue-bot/src/main.rs

@@ -17,7 +17,7 @@ use axum::{
 };
 use config::{load_config, IssueBotConfig, ServerConfig};
 use embeddings::inference_endpoints::EmbeddingApi;
-use futures::{pin_mut, StreamExt};
+use futures::{pin_mut, TryStreamExt};
 use github::GithubApi;
 use huggingface::HuggingfaceApi;
 use metrics::start_metrics_server;
@@ -30,6 +30,7 @@ use slack::Slack;
 use sqlx::{
     postgres::{PgConnectOptions, PgPoolOptions},
     prelude::FromRow,
+    types::Json,
     Pool, Postgres, QueryBuilder,
 };
 use tokio::{
@@ -244,11 +245,20 @@ struct ClosestIssue {
     title: String,
     number: i32,
     html_url: String,
-    body: String,
     #[allow(unused)]
     cosine_similarity: f64,
 }
 
+#[derive(Debug, Deserialize, FromRow)]
+struct JobData {
+    issues_page: i32,
+}
+
+#[derive(Debug, FromRow)]
+struct Job {
+    data: Json<JobData>,
+}
+
 async fn handle_webhooks_wrapper(
     rx: Receiver<EventData>,
     embedding_api: EmbeddingApi,
@@ -282,7 +292,7 @@ async fn handle_webhooks(
                             Vector::from(embedding_api.generate_embedding(issue_text).await?);
 
                         let closest_issues: Vec<ClosestIssue> = sqlx::query_as(
-                            "select title, number, html_url, body, 1 - (embedding <=> $1) as cosine_similarity from issues order by embedding <=> $1 LIMIT 3",
+                            "select title, number, html_url, 1 - (embedding <=> $1) as cosine_similarity from issues order by embedding <=> $1 LIMIT 3",
                         )
                             .bind(embedding.clone())
                             .fetch_all(&pool)
@@ -395,17 +405,17 @@ async fn handle_webhooks(
                     source = repository.source.to_string()
                 );
                 info!(parent: &span, "indexing started");
-                let from_page = sqlx::query!(
-                    "select page from jobs where repository_id = $1",
+                let job = sqlx::query_as!(
+                    Job,
+                    r#"select data as "data: Json<JobData>" from jobs where repository_id = $1"#,
                     repository.repo_id
                 )
-                .map(|row| row.page + 1)
                 .fetch_optional(&pool)
                 .await?;
-                let issues = github_api.get_issues(from_page, repository.clone());
+                let from_issues_page = job.as_ref().map(|j| j.data.issues_page + 1).unwrap_or(1);
+                let issues = github_api.get_issues(from_issues_page, repository.clone());
                 pin_mut!(issues);
-                while let Some(issue) = issues.next().await {
-                    let (issue, page) = issue?;
+                while let Some((issue, page)) = issues.try_next().await? {
                     let embedding_api = embedding_api.clone();
                     let pool = pool.clone();
                     let source = repository.source.to_string();
@@ -461,19 +471,19 @@ async fn handle_webhooks(
                         qb.build().execute(&pool).await?;
                     }
                     if let Some(page) = page {
-                        sqlx::query!(
-                            r#"insert into jobs (repository_id, page)
-                               values ($1, $2)
+                        sqlx::query(
+                                r#"insert into jobs (repository_id, data)
+                               values ($1, jsonb_build_object('issues_page', $2))
                                on conflict (repository_id)
                                do update
                                set
-                                   page = excluded.page,
+                                   data = jsonb_set(jobs.data, '{issues_page}', to_jsonb($2::int), true),
                                    updated_at = current_timestamp"#,
-                            repository.repo_id,
-                            page,
-                        )
-                        .execute(&pool)
-                        .await?;
+                            )
+                            .bind(&repository.repo_id)
+                            .bind(page)
+                            .execute(&pool)
+                            .await?;
                     }
                 }
                 sqlx::query!(

issue-bot/src/slack.rs

@@ -59,14 +59,11 @@ impl Slack {
         closest_issues: &[ClosestIssue],
     ) -> Result<(), SlackError> {
         let mut msg = vec![format!(
-            "Closest issues for {} <{}|#{}>:\n```{}```",
+            "Closest issues for {} (<{}|#{}>):\n```{}```",
             issue.title, issue.number, issue.html_url, issue.body
         )];
         for ci in closest_issues {
-            msg.push(format!(
-                "- {} (<{}|#{}>):\n```{}```",
-                ci.title, ci.html_url, ci.number, ci.body
-            ));
+            msg.push(format!("- {} (<{}|#{}>)", ci.title, ci.html_url, ci.number));
         }
         let body = SlackBody::new(&self.channel, msg.join("\n"));
         self.client