Repository: TabbyML/tabby
File: ee/tabby-webserver/src/service/background_job/third_party_integration.rs (Rust)

Code before:
                integration_service
                    .update_integration_sync_status(integration.id, Some(e.to_string()))
                    .await?;
                logkit::error!("Failed to fetch issues: {}", e);
                return Err(e);
            }
        };

        let pull_stream = match fetch_all_pulls(&integration, &repository).await {
            Ok(s) => s,
            Err(e) => {
                integration_service
                    .update_integration_sync_status(integration.id, Some(e.to_string()))
                    .await?;
                logkit::error!("Failed to fetch pulls: {}", e);
                return Err(e);
            }
        };

        stream! {
            let mut count = 0;
            let mut num_updated = 0;
            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, {} docs deleted", count, num_updated, num_deleted);
                };
            }

            logkit::info!("{} docs seen, {} docs updated, {} docs deleted", count, num_updated, num_deleted);
            index.commit();
        }
        .count()
        .await;

        Ok(())
    }

    pub async fn cron(
        _now: DateTime<Utc>,
        repository: Arc<dyn ThirdPartyRepositoryService>,
        job: Arc<dyn JobService>,

Review comment (suggestion):
```suggestion
                if !index.add(state, doc).await {
```

let indexer to decide whether to delete document based on `state`

Code after:
                integration_service
                    .update_integration_sync_status(integration.id, Some(e.to_string()))
                    .await?;
                logkit::error!("Failed to fetch issues: {}", e);
                return Err(e);
            }
        };

        let pull_stream = match fetch_all_pulls(&integration, &repository).await {
            Ok(s) => s,
            Err(e) => {
                integration_service
                    .update_integration_sync_status(integration.id, Some(e.to_string()))
                    .await?;
                logkit::error!("Failed to fetch pulls: {}", e);
                return Err(e);
            }
        };

        stream! {
            let mut count = 0;
            let mut num_updated = 0;
            for await (state, doc) in issue_stream.chain(pull_stream) {
                if index.sync(state, doc).await {
                    num_updated += 1
                }

                count += 1;
                if count % 100 == 0 {
                    logkit::info!("{} docs seen, {} docs updated", count, num_updated);
                };
            }

            logkit::info!("{} docs seen, {} docs updated", count, num_updated);
            index.commit();
        }
        .count()
        .await;

        Ok(())
    }

    pub async fn cron(
        _now: DateTime<Utc>,
        repository: Arc<dyn ThirdPartyRepositoryService>,
        job: Arc<dyn JobService>,
    ) -> tabby_schema::Result<()> {
        for repository in repository
            .list_repositories_with_filter(None, None, Some(true), None, None, None, None)
            .await?
        {