Progress on refactor

ee/tabby-schema/src/schema/integration.rs

@@ -47,6 +47,6 @@ pub trait IntegrationService: Send + Sync {
         first: Option<usize>,
         last: Option<usize>,
     ) -> Result<Vec<IntegrationAccessToken>>;
-
-    async fn sync_resources(&self, id: ID) -> Result<()>;
+    async fn get_integration(&self, id: ID) -> Result<IntegrationAccessToken>;
+    async fn update_integration_error(&self, id: ID, error: Option<String>) -> Result<()>;
 }

ee/tabby-schema/src/schema/repository/third_party.rs

@@ -28,5 +28,18 @@ pub trait ThirdPartyRepositoryService: Send + Sync {
     ) -> Result<Vec<ProvidedRepository>>;
 
     async fn update_repository_active(&self, id: ID, active: bool) -> Result<()>;
+    async fn upsert_repository(
+        &self,
+        integration_id: ID,
+        vendor_id: String,
+        display_name: String,
+        git_url: String,
+    ) -> Result<()>;
     async fn list_active_git_urls(&self) -> Result<Vec<String>>;
+    async fn sync_repositories(&self, kind: IntegrationKind) -> Result<()>;
+    async fn delete_outdated_repositories(
+        &self,
+        integration_id: ID,
+        before: DateTime<Utc>,
+    ) -> Result<usize>;
 }

ee/tabby-webserver/src/service/repository/third_party.rs

@@ -1,19 +1,26 @@
+use chrono::{DateTime, Utc};
 use juniper::ID;
-use std::marker::PhantomData;
-use strum::IntoEnumIterator;
+use std::sync::Arc;
 use tabby_schema::{
-    integration::IntegrationKind, repository::ProvidedRepository, AsRowid, DbEnum, Result,
+    integration::{IntegrationKind, IntegrationService},
+    repository::ProvidedRepository,
+    AsRowid, DbEnum, Result,
 };
+use tracing::{debug, error};
 use url::Url;
 
 use async_trait::async_trait;
 use tabby_db::DbConn;
-use tabby_schema::repository::{RepositoryProvider, ThirdPartyRepositoryService};
+use tabby_schema::repository::ThirdPartyRepositoryService;
 
 use crate::service::graphql_pagination_to_filter;
+use fetch::fetch_all_repos;
+
+mod fetch;
 
 struct ThirdPartyRepositoryServiceImpl {
     db: DbConn,
+    integration: Arc<dyn IntegrationService>,
 }
 
 #[async_trait]
@@ -56,7 +63,120 @@ impl ThirdPartyRepositoryService for ThirdPartyRepositoryServiceImpl {
 
     async fn list_active_git_urls(&self) -> Result<Vec<String>> {
         let mut urls = vec![];
+
+        let integrations = self
+            .integration
+            .list_integrations(None, None, None, None, None, None)
+            .await?;
+
+        for integration in integrations {
+            let repositories = self
+                .list_repositories(
+                    Some(vec![integration.id.clone()]),
+                    None,
+                    Some(true),
+                    None,
+                    None,
+                    None,
+                    None,
+                )
+                .await?;
+
+            for repository in repositories {
+                let url = format_authenticated_url(
+                    &repository.kind,
+                    &repository.git_url,
+                    &integration.access_token,
+                )?;
+                urls.push(url);
+            }
+        }
+
+        Ok(urls)
+    }
+
+    async fn sync_repositories(&self, kind: IntegrationKind) -> Result<()> {
+        todo!("Loop over integrations and call refresh_repositories_for_provider");
+        Ok(())
+    }
+
+    async fn upsert_repository(
+        &self,
+        integration_id: ID,
+        vendor_id: String,
+        display_name: String,
+        git_url: String,
+    ) -> Result<()> {
+        self.db
+            .upsert_provided_repository(
+                integration_id.as_rowid()?,
+                vendor_id,
+                display_name,
+                git_url,
+            )
+            .await?;
+        Ok(())
     }
+
+    async fn delete_outdated_repositories(
+        &self,
+        integration_id: ID,
+        before: DateTime<Utc>,
+    ) -> Result<usize> {
+        Ok(self
+            .db
+            .delete_outdated_provided_repositories(integration_id.as_rowid()?, before.into())
+            .await?)
+    }
+}
+
+async fn refresh_repositories_for_provider(
+    repository: Arc<dyn ThirdPartyRepositoryService>,
+    integration: Arc<dyn IntegrationService>,
+    provider_id: ID,
+) -> Result<()> {
+    let provider = integration.get_integration(provider_id.clone()).await?;
+    debug!(
+        "Refreshing repositories for provider: {}",
+        provider.display_name
+    );
+
+    let start = Utc::now();
+    let repos = match fetch_all_repos(provider.kind.clone(), &provider.access_token).await {
+        Ok(repos) => repos,
+        Err((e, true)) => {
+            integration
+                .update_integration_error(provider.id.clone(), Some("".into()))
+                .await?;
+            error!(
+                "Credentials for integration {} are expired or invalid",
+                provider.display_name
+            );
+            return Err(e.into());
+        }
+        Err((e, false)) => {
+            error!("Failed to fetch repositories from github: {e}");
+            return Err(e.into());
+        }
+    };
+    for repo in repos {
+        debug!("importing: {}", repo.name);
+
+        let id = repo.vendor_id;
+
+        repository
+            .upsert_repository(provider_id.clone(), id, repo.name, repo.git_url)
+            .await?;
+    }
+
+    integration
+        .update_integration_error(provider_id.clone(), None)
+        .await?;
+    let num_removed = repository
+        .delete_outdated_repositories(provider_id, start.into())
+        .await?;
+    debug!("Removed {} outdated repositories", num_removed);
+    Ok(())
 }
 
 fn format_authenticated_url(

ee/tabby-webserver/src/service/repository/third_party/fetch.rs

@@ -0,0 +1,135 @@
+use gitlab::{
+    api::{projects::Projects, AsyncQuery, Pagination},
+    GitlabBuilder,
+};
+use octocrab::{GitHubError, Octocrab};
+use tabby_schema::integration::IntegrationKind;
+
+pub struct RepositoryInfo {
+    pub name: String,
+    pub git_url: String,
+    pub vendor_id: String,
+}
+
+mod gitlab_types {
+    use gitlab::api::ApiError;
+    use serde::Deserialize;
+
+    #[derive(Deserialize)]
+    pub struct GitlabRepository {
+        pub id: u128,
+        pub path_with_namespace: String,
+        pub http_url_to_repo: String,
+    }
+
+    #[derive(thiserror::Error, Debug)]
+    pub enum GitlabError {
+        #[error(transparent)]
+        Rest(#[from] gitlab::api::ApiError<gitlab::RestError>),
+        #[error(transparent)]
+        Gitlab(#[from] gitlab::GitlabError),
+        #[error(transparent)]
+        Projects(#[from] gitlab::api::projects::ProjectsBuilderError),
+    }
+
+    impl GitlabError {
+        pub fn is_client_error(&self) -> bool {
+            match self {
+                GitlabError::Rest(source)
+                | GitlabError::Gitlab(gitlab::GitlabError::Api { source }) => {
+                    matches!(
+                        source,
+                        ApiError::Auth { .. }
+                            | ApiError::Client {
+                                source: gitlab::RestError::AuthError { .. }
+                            }
+                            | ApiError::Gitlab { .. }
+                    )
+                }
+                _ => false,
+            }
+        }
+    }
+}
+
+pub async fn fetch_all_repos(
+    kind: IntegrationKind,
+    access_token: &str,
+) -> Result<Vec<RepositoryInfo>, (anyhow::Error, bool)> {
+    match kind {
+        IntegrationKind::Github => match fetch_all_github_repos(access_token).await {
+            Ok(repos) => Ok(repos),
+            Err(octocrab::Error::GitHub {
+                source: source @ GitHubError { .. },
+                ..
+            }) if source.status_code.is_client_error() => Err((source.into(), true)),
+            Err(e) => Err((e.into(), false)),
+        },
+        IntegrationKind::Gitlab => match fetch_all_gitlab_repos(access_token).await {
+            Ok(repos) => Ok(repos),
+            Err(e) => {
+                let client_error = e.is_client_error();
+                Err((e.into(), client_error))
+            }
+        },
+    }
+}
+
+async fn fetch_all_gitlab_repos(
+    access_token: &str,
+) -> Result<Vec<RepositoryInfo>, gitlab_types::GitlabError> {
+    let gitlab = GitlabBuilder::new("gitlab.com", access_token)
+        .build_async()
+        .await?;
+    let repos: Vec<gitlab_types::GitlabRepository> = gitlab::api::paged(
+        Projects::builder().membership(true).build()?,
+        Pagination::All,
+    )
+    .query_async(&gitlab)
+    .await?;
+
+    Ok(repos
+        .into_iter()
+        .map(|repo| RepositoryInfo {
+            name: repo.path_with_namespace,
+            git_url: repo.http_url_to_repo,
+            vendor_id: repo.id.to_string(),
+        })
+        .collect())
+}
+
+async fn fetch_all_github_repos(
+    access_token: &str,
+) -> Result<Vec<RepositoryInfo>, octocrab::Error> {
+    let octocrab = Octocrab::builder()
+        .user_access_token(access_token.to_string())
+        .build()?;
+
+    let mut page = 1;
+    let mut repos = vec![];
+
+    loop {
+        let response = octocrab
+            .current()
+            .list_repos_for_authenticated_user()
+            .visibility("all")
+            .page(page)
+            .send()
+            .await?;
+
+        let pages = response.number_of_pages().unwrap_or_default() as u8;
+        repos.extend(response.items.into_iter().filter_map(|repo| {
+            Some(RepositoryInfo {
+                name: repo.full_name.unwrap_or(repo.name),
+                git_url: repo.html_url?.to_string(),
+                vendor_id: repo.id.to_string(),
+            })
+        }));
+
+        page += 1;
+        if page > pages {
+            break;
+        }
+    }
+    Ok(repos)
+}