fix(index): delete skipped documents for StructuredDoc

crates/tabby-index/src/structured_doc/public.rs

@@ -44,6 +44,15 @@
         true
     }
 
+    pub async fn delete(&self, id: &str) -> bool {
+        if self.indexer.is_indexed(id) {
+            self.indexer.delete(id);
+            true
+        } else {
+            false
+        }
+    }
+
     pub fn commit(self) {
         self.indexer.commit();
     }

ee/tabby-webserver/src/service/background_job/third_party_integration.rs

@@ -140,18 +140,23 @@
         stream! {
             let mut count = 0;
             let mut num_updated = 0;
-            for await (updated_at, doc) in issue_stream.chain(pull_stream) {
-                if index.add(updated_at, doc).await {
+            let mut num_deleted = 0;
+            for await (state, doc) in issue_stream.chain(pull_stream) {
+                let id = &doc.id().to_owned();
+                if !state.should_clean && index.add(state.updated_at, doc).await {
                     num_updated += 1
                 }
+                if state.should_clean && index.delete(id).await {
+                    num_deleted += 1;
+                }
 
                 count += 1;
                 if count % 100 == 0 {
-                    logkit::info!("{} docs seen, {} docs updated", count, num_updated);
+                    logkit::info!("{} docs seen, {} docs updated, {} docs deleted", count, num_updated, num_deleted);
                 };
             }
 
-            logkit::info!("{} docs seen, {} docs updated", count, num_updated);
+            logkit::info!("{} docs seen, {} docs updated, {} docs deleted", count, num_updated, num_deleted);
             index.commit();
         }
         .count()
@@ -179,11 +184,16 @@
     }
 }
 
+pub struct FetchState {
+    updated_at: DateTime<Utc>,
+    should_clean: bool,
+}
+
 async fn fetch_all_issues(
     integration: &Integration,
     repository: &ProvidedRepository,
-) -> tabby_schema::Result<BoxStream<'static, (DateTime<Utc>, StructuredDoc)>> {
-    let s: BoxStream<(DateTime<Utc>, StructuredDoc)> = match &integration.kind {
+) -> tabby_schema::Result<BoxStream<'static, (FetchState, StructuredDoc)>> {
+    let s: BoxStream<(FetchState, StructuredDoc)> = match &integration.kind {
         IntegrationKind::Github | IntegrationKind::GithubSelfHosted => list_github_issues(
             &repository.source_id(),
             integration.api_base(),
@@ -207,8 +217,8 @@
 async fn fetch_all_pulls(
     integration: &Integration,
     repository: &ProvidedRepository,
-) -> tabby_schema::Result<BoxStream<'static, (DateTime<Utc>, StructuredDoc)>> {
-    let s: BoxStream<(DateTime<Utc>, StructuredDoc)> = list_github_pulls(
+) -> tabby_schema::Result<BoxStream<'static, (FetchState, StructuredDoc)>> {
+    let s: BoxStream<(FetchState, StructuredDoc)> = list_github_pulls(
         &repository.source_id(),
         integration.api_base(),
         &repository.display_name,

ee/tabby-webserver/src/service/background_job/third_party_integration/issues.rs

@@ -7,14 +7,15 @@
 use serde::Deserialize;
 use tabby_index::public::{StructuredDoc, StructuredDocFields, StructuredDocIssueFields};
 
+use super::FetchState;
 use crate::service::create_gitlab_client;
 
 pub async fn list_github_issues(
     source_id: &str,
     api_base: &str,
     full_name: &str,
     access_token: &str,
-) -> Result<impl Stream<Item = (DateTime<Utc>, StructuredDoc)>> {
+) -> Result<impl Stream<Item = (FetchState, StructuredDoc)>> {
     let octocrab = Octocrab::builder()
         .personal_token(access_token.to_string())
         .base_uri(api_base)?
@@ -62,7 +63,10 @@
                         closed: issue.state == octocrab::models::IssueState::Closed,
                     })
                 };
-                yield (issue.updated_at, doc);
+                yield (FetchState {
+                    updated_at: issue.updated_at,
+                    should_clean: false,
+                }, doc);
             }
 
             page += 1;
@@ -89,7 +93,7 @@
     api_base: &str,
     full_name: &str,
     access_token: &str,
-) -> Result<impl Stream<Item = (DateTime<Utc>, StructuredDoc)>> {
+) -> Result<impl Stream<Item = (FetchState, StructuredDoc)>> {
     let gitlab = create_gitlab_client(api_base, access_token).await?;
 
     let source_id = source_id.to_owned();
@@ -118,7 +122,10 @@
                 body: issue.description.unwrap_or_default(),
                 closed: issue.state == "closed",
             })};
-            yield (issue.updated_at, doc);
+            yield (FetchState {
+                updated_at: issue.updated_at,
+                should_clean: false,
+            }, doc);
         }
     };
 

ee/tabby-webserver/src/service/background_job/third_party_integration/pulls.rs

@@ -1,16 +1,17 @@
 use anyhow::{anyhow, Result};
 use async_stream::stream;
-use chrono::{DateTime, Utc};
 use futures::Stream;
 use octocrab::{models::IssueState, Octocrab};
 use tabby_index::public::{StructuredDoc, StructuredDocFields, StructuredDocPullDocumentFields};
 
+use super::FetchState;
+
 pub async fn list_github_pulls(
     source_id: &str,
     api_base: &str,
     full_name: &str,
     access_token: &str,
-) -> Result<impl Stream<Item = (DateTime<Utc>, StructuredDoc)>> {
+) -> Result<impl Stream<Item = (FetchState, StructuredDoc)>> {
     let octocrab = Octocrab::builder()
         .personal_token(access_token.to_string())
         .base_uri(api_base)?
@@ -43,14 +44,32 @@
             let pages = response.number_of_pages().unwrap_or_default();
 
             for pull in response.items {
+                let url = pull.html_url.map(|url| url.to_string()).unwrap_or_else(|| pull.url);
+                let title = pull.title.clone().unwrap_or_default();
+                let body = pull.body.clone().unwrap_or_default();
+                let doc = StructuredDoc {
+                    source_id: source_id.to_string(),
+                    fields: StructuredDocFields::Pull(StructuredDocPullDocumentFields {
+                        link: url.clone(),
+                        title,
+                        body,
+                        merged: pull.merged_at.is_some(),
+                        diff: String::new(),
+                    }),
+                };
+
                 // skip closed but not merged pulls
                 if let Some(state) = pull.state {
                     if state == IssueState::Closed && pull.merged_at.is_none() {
-                        continue
+                        yield (FetchState{
+                            updated_at: pull.updated_at.unwrap(),
+                            should_clean: true,
+                        }, doc);
+                        continue;
                     }
                 }
 
-                let url = pull.html_url.map(|url| url.to_string()).unwrap_or_else(|| pull.url);
+
                 let diff = match octocrab.pulls(&owner, &repo).get_diff(pull.number).await {
                     Ok(x) if x.len() < 1024*1024*10 => x,
                     Ok(_) => {
@@ -71,10 +90,13 @@
                         body: pull.body.unwrap_or_default(),
                         diff,
                         merged: pull.merged_at.is_some(),
-                    })
-                };
+                })};
+
 
-                yield (pull.updated_at.unwrap(), doc);
+                yield (FetchState{
+                    updated_at: pull.updated_at.unwrap(),
+                    should_clean: false,
+                }, doc);
             }
 
             page += 1;