feat(sync): add git metadata schema and sync pipeline

Author: marcialwushu

Cargo.lock

@@ -2,6 +2,8 @@
 members = [
     "crates/gitbase-cli",
     "crates/gitbase-db",
+    "crates/gitbase-git",
+    "crates/gitbase-loader",
     "crates/gitbase-pgwire",
 ]
 resolver = "2"
@@ -15,9 +17,12 @@ rust-version = "1.80"
 anyhow = "1"
 async-trait = "0.1"
 clap = { version = "4", features = ["derive", "env"] }
+gix = "0.66"
 pgwire = "0.38"
+sha1 = "0.10"
 sqlx = { version = "0.8", features = ["postgres", "runtime-tokio", "migrate", "macros"] }
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "net"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
 futures = "0.3"
+walkdir = "2"

Cargo.toml

@@ -11,6 +11,7 @@ path = "src/main.rs"
 anyhow.workspace = true
 clap.workspace = true
 gitbase-db = { path = "../gitbase-db" }
+gitbase-loader = { path = "../gitbase-loader" }
 gitbase-pgwire = { path = "../gitbase-pgwire" }
 tokio.workspace = true
 tracing.workspace = true

crates/gitbase-cli/Cargo.toml

@@ -27,6 +27,27 @@ enum Commands {
         #[arg(long, env = "GITBASE_DB_MAX_CONNECTIONS", default_value_t = 10)]
         max_connections: u32,
     },
+
+    /// Sync Git metadata into PostgreSQL
+    Sync {
+        /// Root directories containing Git repositories
+        #[arg(
+            long,
+            env = "GITBASE_REPO_ROOTS",
+            value_delimiter = ',',
+            num_args = 1..,
+            default_value = "./"
+        )]
+        repo_roots: Vec<String>,
+
+        /// PostgreSQL connection string
+        #[arg(long, env = "DATABASE_URL")]
+        database_url: String,
+
+        /// Maximum database connections
+        #[arg(long, env = "GITBASE_DB_MAX_CONNECTIONS", default_value_t = 10)]
+        max_connections: u32,
+    },
 }
 
 #[tokio::main]
@@ -50,6 +71,30 @@ async fn main() -> Result<()> {
             let factory = Arc::new(gitbase_pgwire::GitbaseServerFactory::new(pool));
             gitbase_pgwire::serve(&bind, factory).await?;
         }
+        Commands::Sync {
+            repo_roots,
+            database_url,
+            max_connections,
+        } => {
+            let pool = gitbase_db::connect(&database_url, max_connections).await?;
+            gitbase_db::health_check(&pool).await?;
+            tracing::info!("health check passed");
+
+            let roots = repo_roots
+                .iter()
+                .map(|root| root.into())
+                .collect::<Vec<_>>();
+            let report = gitbase_loader::sync_repositories(&pool, &roots).await?;
+            tracing::info!(
+                repositories = report.repositories,
+                refs = report.refs,
+                commits = report.commits,
+                commit_parents = report.commit_parents,
+                tree_entries = report.tree_entries,
+                files = report.files,
+                "sync completed"
+            );
+        }
     }
 
     Ok(())

crates/gitbase-cli/src/main.rs

@@ -2,6 +2,8 @@ use anyhow::Result;
 use sqlx::postgres::PgPoolOptions;
 use sqlx::PgPool;
 
+pub mod metadata;
+
 /// Create a connection pool to PostgreSQL and run migrations.
 pub async fn connect(database_url: &str, max_connections: u32) -> Result<PgPool> {
     let pool = PgPoolOptions::new()

crates/gitbase-db/src/lib.rs

@@ -0,0 +1,195 @@
+use anyhow::Result;
+use sqlx::PgPool;
+
+#[derive(Debug, Clone)]
+pub struct RepositoryRecord {
+    pub id: String,
+    pub name: String,
+    pub path: String,
+    pub default_ref: Option<String>,
+    pub is_bare: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct RefRecord {
+    pub repository_id: String,
+    pub name: String,
+    pub target_hash: String,
+    pub kind: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct CommitRecord {
+    pub repository_id: String,
+    pub hash: String,
+    pub tree_hash: String,
+    pub author_name: Option<String>,
+    pub author_email: Option<String>,
+    pub committer_name: Option<String>,
+    pub committer_email: Option<String>,
+    pub message: Option<String>,
+    pub committed_at_seconds: Option<i64>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CommitParentRecord {
+    pub repository_id: String,
+    pub commit_hash: String,
+    pub parent_hash: String,
+    pub parent_index: i32,
+}
+
+#[derive(Debug, Clone)]
+pub struct TreeEntryRecord {
+    pub repository_id: String,
+    pub commit_hash: String,
+    pub path: String,
+    pub object_hash: String,
+    pub object_type: String,
+    pub file_mode: String,
+    pub size: Option<i64>,
+}
+
+#[derive(Debug, Clone)]
+pub struct FileRecord {
+    pub repository_id: String,
+    pub commit_hash: String,
+    pub path: String,
+    pub blob_hash: String,
+    pub language: Option<String>,
+    pub size: Option<i64>,
+    pub is_binary: bool,
+}
+
+pub async fn upsert_repository(pool: &PgPool, record: &RepositoryRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.repositories (id, name, path, default_ref, is_bare)\
+         VALUES ($1, $2, $3, $4, $5)\
+         ON CONFLICT (id) DO UPDATE\
+         SET name = EXCLUDED.name,\
+             path = EXCLUDED.path,\
+             default_ref = EXCLUDED.default_ref,\
+             is_bare = EXCLUDED.is_bare",
+    )
+    .bind(&record.id)
+    .bind(&record.name)
+    .bind(&record.path)
+    .bind(&record.default_ref)
+    .bind(record.is_bare)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
+pub async fn upsert_ref(pool: &PgPool, record: &RefRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.refs (repository_id, name, target_hash, kind)\
+         VALUES ($1, $2, $3, $4)\
+         ON CONFLICT (repository_id, name) DO UPDATE\
+         SET target_hash = EXCLUDED.target_hash,\
+             kind = EXCLUDED.kind",
+    )
+    .bind(&record.repository_id)
+    .bind(&record.name)
+    .bind(&record.target_hash)
+    .bind(&record.kind)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
+pub async fn upsert_commit(pool: &PgPool, record: &CommitRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.commits (repository_id, hash, tree_hash, author_name, author_email,\
+         committer_name, committer_email, message, committed_at)\
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8,\
+         CASE WHEN $9 IS NULL THEN NULL ELSE to_timestamp($9) END)\
+         ON CONFLICT (repository_id, hash) DO UPDATE\
+         SET tree_hash = EXCLUDED.tree_hash,\
+             author_name = EXCLUDED.author_name,\
+             author_email = EXCLUDED.author_email,\
+             committer_name = EXCLUDED.committer_name,\
+             committer_email = EXCLUDED.committer_email,\
+             message = EXCLUDED.message,\
+             committed_at = EXCLUDED.committed_at",
+    )
+    .bind(&record.repository_id)
+    .bind(&record.hash)
+    .bind(&record.tree_hash)
+    .bind(&record.author_name)
+    .bind(&record.author_email)
+    .bind(&record.committer_name)
+    .bind(&record.committer_email)
+    .bind(&record.message)
+    .bind(record.committed_at_seconds)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
+pub async fn upsert_commit_parent(pool: &PgPool, record: &CommitParentRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.commit_parents (repository_id, commit_hash, parent_hash, parent_index)\
+         VALUES ($1, $2, $3, $4)\
+         ON CONFLICT (repository_id, commit_hash, parent_index) DO UPDATE\
+         SET parent_hash = EXCLUDED.parent_hash",
+    )
+    .bind(&record.repository_id)
+    .bind(&record.commit_hash)
+    .bind(&record.parent_hash)
+    .bind(record.parent_index)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
+pub async fn upsert_tree_entry(pool: &PgPool, record: &TreeEntryRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.tree_entries (repository_id, commit_hash, path, object_hash,\
+         object_type, file_mode, size)\
+         VALUES ($1, $2, $3, $4, $5, $6, $7)\
+         ON CONFLICT (repository_id, commit_hash, path) DO UPDATE\
+         SET object_hash = EXCLUDED.object_hash,\
+             object_type = EXCLUDED.object_type,\
+             file_mode = EXCLUDED.file_mode,\
+             size = EXCLUDED.size",
+    )
+    .bind(&record.repository_id)
+    .bind(&record.commit_hash)
+    .bind(&record.path)
+    .bind(&record.object_hash)
+    .bind(&record.object_type)
+    .bind(&record.file_mode)
+    .bind(record.size)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
+pub async fn upsert_file(pool: &PgPool, record: &FileRecord) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO gitbase.files (repository_id, commit_hash, path, blob_hash, language, size, is_binary)\
+         VALUES ($1, $2, $3, $4, $5, $6, $7)\
+         ON CONFLICT (repository_id, commit_hash, path) DO UPDATE\
+         SET blob_hash = EXCLUDED.blob_hash,\
+             language = EXCLUDED.language,\
+             size = EXCLUDED.size,\
+             is_binary = EXCLUDED.is_binary",
+    )
+    .bind(&record.repository_id)
+    .bind(&record.commit_hash)
+    .bind(&record.path)
+    .bind(&record.blob_hash)
+    .bind(&record.language)
+    .bind(record.size)
+    .bind(record.is_binary)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}

crates/gitbase-db/src/metadata.rs

@@ -0,0 +1,11 @@
+[package]
+name = "gitbase-git"
+version = "0.1.0"
+edition.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+gix.workspace = true
+sha1.workspace = true
+tracing.workspace = true
+walkdir.workspace = true