Update Libra: support LFS Lock API & LFS Chunk API

ceres/Cargo.toml

@@ -29,3 +29,4 @@ bytes = { workspace = true }
 async-trait = { workspace = true }
 rand = { workspace = true }
 sha256 = { workspace = true }
+sea-orm = { workspace = true }

ceres/src/lfs/handler.rs

@@ -5,7 +5,8 @@ use anyhow::Result;
 use bytes::Bytes;
 use chrono::{prelude::*, Duration};
 use rand::prelude::*;
-
+use sea_orm::ActiveValue::Set;
+use sea_orm::IntoActiveModel;
 use callisto::{lfs_locks, lfs_objects, lfs_split_relations};
 use common::errors::{GitLFSError, MegaError};
 use jupiter::context::Context;
@@ -236,7 +237,7 @@ pub async fn lfs_process_batch(
 /// else return an error.
 pub async fn lfs_fetch_chunk_ids(
     context: &Context,
-    fetch_vars: &RequestVars,
+    oid: &String,
 ) -> Result<Vec<ChunkRepresentation>, GitLFSError> {
     let config = context.config.lfs.clone();
 
@@ -247,13 +248,13 @@ pub async fn lfs_fetch_chunk_ids(
     }
     let storage = context.services.lfs_db_storage.clone();
 
-    let meta = lfs_get_meta(storage.clone(), &fetch_vars.oid)
+    let meta = lfs_get_meta(storage.clone(), oid)
         .await
         .map_err(|_| GitLFSError::GeneralError("".to_string()))?;
     assert!(meta.splited, "database didn't match the split mode");
 
     let relations = storage
-        .get_lfs_relations(fetch_vars.oid.clone())
+        .get_lfs_relations(oid.to_owned())
         .await
         .map_err(|_| GitLFSError::GeneralError("".to_string()))?;
 
@@ -270,10 +271,7 @@ pub async fn lfs_fetch_chunk_ids(
         let tmp_request_vars = RequestVars {
             oid: relation.sub_oid.clone(),
             size: relation.size,
-            authorization: fetch_vars.authorization.clone(),
-            password: fetch_vars.password.clone(),
-            user: fetch_vars.user.clone(),
-            repo: fetch_vars.repo.clone(),
+            ..Default::default()
         };
         response_objects.push(ChunkRepresentation {
             sub_oid: relation.sub_oid,
@@ -557,7 +555,7 @@ async fn lfs_add_lock(
 
     match result {
         // Update
-        Some(mut val) => {
+        Some(val) => {
             let d = val.data.to_owned();
             let mut locks_from_data = if !d.is_empty() {
                 let locks_from_data: Vec<Lock> = serde_json::from_str(&d).unwrap();
@@ -575,7 +573,9 @@ async fn lfs_add_lock(
             });
             let d = serde_json::to_string(&locks_from_data).unwrap();
 
-            d.clone_into(&mut val.data);
+            // must turn into `ActiveModel` before modify, or update failed.
+            let mut val = val.into_active_model();
+            val.data = Set(d);
             let res = storage.update_lock(val).await;
             match res.is_ok() {
                 true => Ok(()),
@@ -678,7 +678,7 @@ async fn delete_lock(
     let result = storage.get_lock_by_id(repo).await.unwrap();
     match result {
         // Exist, then delete.
-        Some(mut val) => {
+        Some(val) => {
             let d = val.data.to_owned();
             let locks_from_data = if !d.is_empty() {
                 let locks_from_data: Vec<Lock> = serde_json::from_str(&d).unwrap();
@@ -728,7 +728,8 @@ async fn delete_lock(
 
             // Update remaining locks.
             let data = serde_json::to_string(&new_locks).unwrap();
-            data.clone_into(&mut val.data);
+            let mut val = val.into_active_model();
+            val.data = Set(data);
             let res = storage.update_lock(val).await;
             match res.is_ok() {
                 true => Ok(lock_to_delete),

ceres/src/lfs/lfs_structs.rs

@@ -88,7 +88,6 @@ pub struct BatchRequest {
     pub transfers: Vec<String>,
     pub objects: Vec<RequestVars>,
     pub hash_algo: String,
-    pub enable_split: Option<bool>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -105,7 +104,7 @@ pub struct FetchchunkResponse {
     pub chunks: Vec<ChunkRepresentation>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct Link {
     pub href: String,
     pub header: HashMap<String, String>,
@@ -192,9 +191,14 @@ pub struct VerifiableLockList {
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct LockListQuery {
+    #[serde(default)]
     pub path: String,
+    #[serde(default)]
     pub id: String,
+    #[serde(default)]
     pub cursor: String,
+    #[serde(default)]
     pub limit: String,
+    #[serde(default)]
     pub refspec: String,
 }

jupiter/src/storage/lfs_db_storage.rs

@@ -129,9 +129,9 @@ impl LfsDbStorage {
 
     pub async fn update_lock(
         &self,
-        lfs_lock: lfs_locks::Model,
+        lfs_lock: lfs_locks::ActiveModel,
     ) -> Result<lfs_locks::Model, MegaError> {
-        Ok(lfs_locks::Entity::update(lfs_lock.into_active_model())
+        Ok(lfs_locks::Entity::update(lfs_lock)
             .exec(self.get_connection())
             .await
             .unwrap())

libra/Cargo.toml

@@ -38,12 +38,13 @@ indicatif = "0.17.8"
 wax = "0.6.0"
 lazy_static = { workspace = true }
 regex = { workspace = true }
-sha2 = "0.10.8"
+ring = "0.17.8"
 hex = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 async_static = "0.1.3"
 once_cell = "1.19.0"
+byte-unit = "5.1.4"
 
 [target.'cfg(unix)'.dependencies] # only on Unix
 pager = "0.16.0"

libra/src/command/lfs.rs

@@ -3,9 +3,17 @@ use std::fs::{File, OpenOptions};
 use std::io;
 use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
 use std::path::Path;
+use byte_unit::UnitType;
+use reqwest::StatusCode;
+use ceres::lfs::lfs_structs::LockListQuery;
+use mercury::internal::index::Index;
+use crate::command::{ask_basic_auth, status};
+use crate::internal::head::Head;
+use crate::internal::protocol::lfs_client::LFS_CLIENT;
 use crate::utils::{lfs, path, util};
 use crate::utils::path_ext::PathExt;
 
+/// [Docs](https://github.com/git-lfs/git-lfs/tree/main/docs/man)
 #[derive(Subcommand, Debug)]
 pub enum LfsCmds {
     /// View or add LFS paths to Libra Attributes (root)
@@ -16,6 +24,40 @@ pub enum LfsCmds {
     Untrack {
         path: Vec<String>,
     },
+    /// Lists currently locked files from the Libra LFS server. (Current Branch)
+    Locks {
+        #[clap(long, short)]
+        id: Option<String>,
+        #[clap(long, short)]
+        path: Option<String>,
+        #[clap(long, short)]
+        limit: Option<u64>,
+    },
+    /// Set a file as "locked" on the Libra LFS server
+    Lock {
+        /// String path name of the locked file. This should be relative to the root of the repository working directory
+        path: String,
+    },
+    /// Remove "locked" setting for a file on the Libra LFS server
+    Unlock {
+        path: String,
+        #[clap(long, short)]
+        force: bool,
+        #[clap(long, short)]
+        id: Option<String>
+    },
+    /// Show information about Git LFS files in the index and working tree (current branch)
+    LsFiles {
+        /// Show the entire 64 character OID, instead of just first 10.
+        #[clap(long, short)]
+        long: bool,
+        /// Show the size of the LFS object between parenthesis at the end of a line.
+        #[clap(long, short)]
+        size: bool,
+        /// Show only the lfs tracked file names.
+        #[clap(long, short)]
+        name_only: bool,
+    }
 }
 
 pub async fn execute(cmd: LfsCmds) {
@@ -43,6 +85,131 @@ pub async fn execute(cmd: LfsCmds) {
             let path = convert_patterns_to_workdir(path); //
             untrack_lfs_patterns(&attr_path, path).unwrap();
         }
+        LfsCmds::Locks { id, path, limit } => {
+            let refspec = current_refspec().await.unwrap();
+            tracing::debug!("refspec: {}", refspec);
+            let query = LockListQuery {
+                id: id.unwrap_or_default(),
+                path: path.unwrap_or_default(),
+                limit: limit.map(|l| l.to_string()).unwrap_or_default(),
+                cursor: "".to_string(),
+                refspec,
+            };
+            let locks = LFS_CLIENT.await.get_locks(query).await.locks;
+            if !locks.is_empty() {
+                let max_path_len = locks.iter().map(|l| l.path.len()).max().unwrap();
+                for lock in locks {
+                    println!("{:<path_width$}\tID:{}", lock.path, lock.id, path_width = max_path_len);
+                }
+            }
+        }
+        LfsCmds::Lock { path } => {
+            // Only check existence
+            if !Path::new(&path).exists() {
+                eprintln!("fatal: pathspec '{}' did not match any files", path);
+                return;
+            }
+
+            let refspec = current_refspec().await.unwrap();
+            let mut auth = None;
+            loop {
+                let code = LFS_CLIENT.await.lock(path.clone(), refspec.clone(), auth.clone()).await;
+                if code.is_success() {
+                    println!("Locked {}", path);
+                } else if code == StatusCode::FORBIDDEN {
+                    eprintln!("Forbidden: You must have push access to create a lock");
+                    auth = Some(ask_basic_auth());
+                    continue;
+                } else if code == StatusCode::CONFLICT {
+                    eprintln!("Conflict: already created lock");
+                }
+                break;
+            }
+        }
+        LfsCmds::Unlock { path, force, id } => {
+            if !force {
+                if !Path::new(&path).exists() {
+                    eprintln!("fatal: pathspec '{}' did not match any files", path);
+                    return;
+                }
+                if !status::is_clean().await {
+                    eprintln!("fatal: working tree not clean");
+                    return;
+                }
+            }
+            let refspec = current_refspec().await.unwrap();
+            let id = match id {
+                None => {
+                    // get id by path
+                    let locks = LFS_CLIENT.await.get_locks(LockListQuery {
+                        refspec: refspec.clone(),
+                        path: path.clone(),
+                        id: "".to_string(),
+                        cursor: "".to_string(),
+                        limit: "".to_string(),
+                    }).await.locks;
+                    if locks.is_empty() {
+                        eprintln!("fatal: no lock found for path '{}'", path);
+                        return;
+                    }
+                    locks[0].id.clone()
+                }
+                Some(id) => id
+            };
+            let mut auth = None;
+            loop {
+                let code = LFS_CLIENT.await.unlock(id.clone(), refspec.clone(), force, auth.clone()).await;
+                if code.is_success() {
+                    println!("Unlocked {}", path);
+                } else if code == StatusCode::FORBIDDEN {
+                    eprintln!("Forbidden: You must have push access to unlock");
+                    auth = Some(ask_basic_auth());
+                    continue;
+                }
+                break;
+            }
+        }
+        LfsCmds::LsFiles { long, size, name_only} => {
+            let idx_file = path::index();
+            let index = Index::load(&idx_file).unwrap();
+            let entries = index.tracked_entries(0);
+            let storage = util::objects_storage();
+            for entry in entries {
+                let path_abs = util::workdir_to_absolute(&entry.name);
+                if lfs::is_lfs_tracked(&path_abs) {
+                    let data = storage.get(&entry.hash).unwrap();
+                    if let Some((oid, lfs_size)) = lfs::parse_pointer_data(&data) {
+                        let is_pointer = lfs::parse_pointer_file(&path_abs).is_ok();
+                        // An asterisk (*) after the OID indicates a full object, a minus (-) indicates an LFS pointer.
+                        // or not exists (-)
+                        let _type = if is_pointer || !path_abs.exists() { "-" } else { "*" };
+                        let oid = if long { oid } else { oid[..10].to_owned() };
+                        let tail = if size {
+                            let byte = byte_unit::Byte::from(lfs_size);
+                            let byte = byte.get_appropriate_unit(UnitType::Decimal);
+                            format!(" ({byte:.2})")
+                        } else {
+                            "".to_string()
+                        };
+                        if name_only {
+                            println!("{}{}", entry.name, tail);
+                        } else {
+                            println!("{} {} {}{}", oid, _type, entry.name, tail);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+pub(crate) async fn current_refspec() -> Option<String> {
+    match Head::current().await {
+        Head::Branch(name) => Some(format!("refs/heads/{}", name)),
+        Head::Detached(_) => {
+            println!("fatal: HEAD is detached");
+            None
+        }
     }
 }
 

libra/src/command/push.rs

@@ -120,7 +120,11 @@ pub async fn execute(args: PushArgs) {
 
     { // upload lfs files
         let client = LFSClient::from_url(&url);
-        client.push_objects(&objs, auth.clone()).await;
+        let res = client.push_objects(&objs, auth.clone()).await;
+        if res.is_err() {
+            eprintln!("fatal: LFS files upload failed, stop pushing");
+            return;
+        }
     }
 
     // let (tx, rx) = mpsc::channel::<Entry>();

libra/src/command/status.rs

@@ -108,6 +108,13 @@ pub async fn execute() {
     }
 }
 
+/// Check if the working tree is clean
+pub async fn is_clean() -> bool {
+    let staged = changes_to_be_committed().await;
+    let unstaged = changes_to_be_staged();
+    staged.is_empty() && unstaged.is_empty()
+}
+
 /**
  * Compare the difference between `index` and the last `Commit Tree`
  */

libra/src/internal/config.rs

@@ -56,6 +56,7 @@ impl Config {
     }
 
     /// Get remote repo name by branch name
+    /// - You may need to `[branch::set-upstream]` if return `None`
     pub async fn get_remote(branch: &str) -> Option<String> {
         // e.g. [branch "master"].remote = origin
         Config::get("branch", Some(branch), "remote").await