feat(webserver): implement new storage layer for repository meta

crates/tabby-common/src/config.rs

@@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize};
 use crate::{
     path::repositories_dir,
     terminal::{HeaderFormat, InfoMessage},
+    SourceFile,
 };
 
 #[derive(Serialize, Deserialize, Default)]
@@ -145,6 +146,10 @@ impl Default for ServerConfig {
 #[async_trait]
 pub trait RepositoryAccess: Send + Sync {
     async fn list_repositories(&self) -> Result<Vec<RepositoryConfig>>;
+
+    fn start_snapshot(&self, _version: u64) {}
+    fn process_file(&self, _version: u64, _file: SourceFile) {}
+    fn finish_snapshot(&self, _version: u64) {}
 }
 
 pub struct ConfigRepositoryAccess;

crates/tabby-common/src/lib.rs

@@ -21,7 +21,7 @@ use path::dataset_dir;
 use serde::{Deserialize, Serialize};
 use serde_jsonlines::JsonLinesReader;
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct SourceFile {
     pub git_url: String,
     pub filepath: String,

crates/tabby-scheduler/src/dataset.rs

@@ -15,7 +15,7 @@
 use lazy_static::lazy_static;
 use serde_jsonlines::WriteExt;
 use tabby_common::{
-    config::RepositoryConfig,
+    config::{RepositoryAccess, RepositoryConfig},
     path::{dataset_dir, dependency_file},
     DependencyFile, SourceFile,
 };
@@ -25,11 +25,21 @@
 use crate::utils::tqdm;
 
 trait RepositoryExt {
-    fn create_dataset(&self, writer: &mut impl Write) -> Result<()>;
+    fn create_dataset(
+        &self,
+        writer: &mut impl Write,
+        access: &impl RepositoryAccess,
+        snapshot_version: u64,
+    ) -> Result<()>;
 }
 
 impl RepositoryExt for RepositoryConfig {
-    fn create_dataset(&self, writer: &mut impl Write) -> Result<()> {
+    fn create_dataset(
+        &self,
+        writer: &mut impl Write,
+        access: &impl RepositoryAccess,
+        snapshot_version: u64,
+    ) -> Result<()> {
         let dir = self.dir();
 
         let walk_dir_iter = || {
@@ -70,7 +80,8 @@
                         language,
                         content: file_content,
                     };
-                    writer.write_json_lines([source_file])?;
+                    writer.write_json_lines([source_file.clone()])?;
+                    access.process_file(snapshot_version.clone(), source_file);
                 }
                 Err(e) => {
                     error!("Cannot read {relative_path:?}: {e:?}");
@@ -95,9 +106,10 @@
     }
 }
 
-pub fn create_dataset(config: &[RepositoryConfig]) -> Result<()> {
+pub fn create_dataset(config: &[RepositoryConfig], access: &impl RepositoryAccess) -> Result<()> {
     fs::remove_dir_all(dataset_dir()).ok();
     fs::create_dir_all(dataset_dir())?;
+
     let mut writer = FileRotate::new(
         SourceFile::files_jsonl(),
         AppendCount::new(usize::max_value()),
@@ -107,15 +119,21 @@
         None,
     );
 
+    let snapshot_version = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("Failed to read system clock")
+        .as_millis() as u64;
+    access.start_snapshot(snapshot_version);
     let mut deps = DependencyFile::default();
     for repository in config {
         deps::collect(repository.dir().as_path(), &mut deps);
-        repository.create_dataset(&mut writer)?;
+        repository.create_dataset(&mut writer, access, snapshot_version.clone())?;
     }
 
     serdeconv::to_json_file(&deps, dependency_file())?;
 
     writer.flush()?;
+    access.finish_snapshot(snapshot_version);
     Ok(())
 }
 

crates/tabby-scheduler/src/lib.rs

@@ -15,7 +15,7 @@
 pub async fn scheduler<T: RepositoryAccess + 'static>(now: bool, access: T) -> Result<()> {
     if now {
         let repositories = access.list_repositories().await?;
-        job_sync(&repositories)?;
+        job_sync(&repositories, &access)?;
         job_index(&repositories)?;
     } else {
         let access = Arc::new(access);
@@ -37,7 +37,7 @@
                         .list_repositories()
                         .await
                         .expect("Must be able to retrieve repositories for sync");
-                    if let Err(e) = job_sync(&repositories) {
+                    if let Err(e) = job_sync(&repositories, &*access) {
                         error!("{e}");
                     }
                     if let Err(e) = job_index(&repositories) {
@@ -66,15 +66,15 @@
     Ok(())
 }
 
-fn job_sync(repositories: &[RepositoryConfig]) -> Result<()> {
+fn job_sync(repositories: &[RepositoryConfig], access: &impl RepositoryAccess) -> Result<()> {
     println!("Syncing {} repositories...", repositories.len());
     let ret = repository::sync_repositories(repositories);
     if let Err(err) = ret {
         return Err(err.context("Failed to sync repositories"));
     }
 
     println!("Building dataset...");
-    let ret = dataset::create_dataset(repositories);
+    let ret = dataset::create_dataset(repositories, access);
     if let Err(err) = ret {
         return Err(err.context("Failed to build dataset"));
     }

ee/tabby-webserver/Cargo.toml

@@ -22,6 +22,7 @@ hyper = { workspace = true, features = ["client"] }
 jsonwebtoken = "9.1.0"
 juniper.workspace = true
 juniper-axum = { path = "../../crates/juniper-axum" }
+kv = { version = "0.24.0", features = ["json-value"] }
 lazy_static.workspace = true
 lettre = { version = "0.11.3", features = ["tokio1", "tokio1-native-tls"] }
 mime_guess = "2.0.4"