diff --git a/adapters/src/manifest/addressables.rs b/adapters/src/manifest/addressables.rs index 9afd645..3fa4967 100644 --- a/adapters/src/manifest/addressables.rs +++ b/adapters/src/manifest/addressables.rs @@ -23,19 +23,23 @@ impl AddressablesCatalogDriver { } fn locator_id(json: &Value) -> Option { - json.get("m_LocatorId") + let locator = json + .get("m_LocatorId") .and_then(|value| value.as_str()) - .map(ToOwned::to_owned) + .map(ToOwned::to_owned); + locator } fn cdn_prefixes(json: &Value) -> Vec { - json.get("m_InternalIdPrefixes") + let prefixes = json + .get("m_InternalIdPrefixes") .and_then(|value| value.as_array()) .into_iter() .flatten() .filter_map(|value| value.as_str()) .map(ToOwned::to_owned) - .collect() + .collect(); + prefixes } fn resource_type_for_path(path: &str) -> ResourceType { @@ -49,7 +53,8 @@ impl AddressablesCatalogDriver { } fn resources(json: &Value) -> Vec { - json.get("m_InternalIds") + let resources = json + .get("m_InternalIds") .and_then(|value| value.as_array()) .into_iter() .flatten() @@ -67,7 +72,8 @@ impl AddressablesCatalogDriver { resource_type: Self::resource_type_for_path(path), }) }) - .collect() + .collect(); + resources } } diff --git a/adapters/src/manifest/registry.rs b/adapters/src/manifest/registry.rs index 940fbaa..6a2cd3c 100644 --- a/adapters/src/manifest/registry.rs +++ b/adapters/src/manifest/registry.rs @@ -179,7 +179,8 @@ impl ManifestDriverRegistry { /// ``` pub async fn parse(&self, raw_data: &[u8]) -> Result { let driver = self.select_driver(raw_data)?; - driver.parse(raw_data).await + let manifest = driver.parse(raw_data).await?; + Ok(manifest) } /// 获取所有已注册的 Driver diff --git a/adapters/src/unity/unity_2021_3.rs b/adapters/src/unity/unity_2021_3.rs index ce77f03..64788b2 100644 --- a/adapters/src/unity/unity_2021_3.rs +++ b/adapters/src/unity/unity_2021_3.rs @@ -5,6 +5,9 @@ use super::adapter::{ParsedAssetBundle, RawAssetBundle, UnityAdapter, VersionRange}; use async_trait::async_trait; +const PARSE_NOT_IMPLEMENTED: &str = "parse() 将在 Phase 2 实现"; +const SERIALIZE_NOT_IMPLEMENTED: &str = "serialize() 将在 Phase 2 实现"; + /// Unity 2021.3 Adapter pub struct Unity2021_3Adapter; @@ -14,35 +17,30 @@ impl Unity2021_3Adapter { Self } - /// 检测 Unity 版本（从文件头） - fn detect_unity_version(data: &[u8]) -> Option { - // UnityFS 文件头结构： - // 0x00-0x06: "UnityFS\0" - // 0x08-0x0B: format version - // 0x0C-...: Unity version string - + fn unity_version_bytes(data: &[u8]) -> Option<&[u8]> { if data.len() < 20 { return None; } - // 检查签名 if &data[0..7] != b"UnityFS" { return None; } - // 读取 Unity 版本字符串 - // 版本字符串从偏移 0x14 开始（可能因格式版本不同而变化） - if let Some(version_start) = data.windows(7).position(|w| w.starts_with(b"2021.3.")) { - // 读取版本字符串直到 null 或非 ASCII - let version_bytes = &data[version_start..]; - if let Some(end) = version_bytes.iter().position(|&b| b == 0 || !b.is_ascii()) { - if let Ok(version) = std::str::from_utf8(&version_bytes[..end]) { - return Some(version.to_string()); - } - } - } + let version_start = data.windows(7).position(|window| window == b"2021.3.")?; + let version_bytes = &data[version_start..]; + let version_end = version_bytes + .iter() + .position(|&byte| byte == 0 || !byte.is_ascii())?; - None + Some(&version_bytes[..version_end]) + } + + /// 检测 Unity 版本（从文件头） + fn detect_unity_version(data: &[u8]) -> Option { + let version_bytes = Self::unity_version_bytes(data)?; + std::str::from_utf8(version_bytes) + .map(ToOwned::to_owned) + .ok() } } @@ -78,7 +76,7 @@ impl UnityAdapter for Unity2021_3Adapter { // 2. 解压缩数据块 // 3. 解析 TypeTree // 4. 提取 Asset 对象 - Err("parse() 将在 Phase 2 实现".to_string()) + Err(PARSE_NOT_IMPLEMENTED.to_string()) } async fn serialize(&self, _parsed: &ParsedAssetBundle) -> Result, String> { @@ -88,7 +86,7 @@ impl UnityAdapter for Unity2021_3Adapter { // 2. 重新构建 TypeTree // 3. 压缩数据块 // 4. 写入 UnityFS 文件头 - Err("serialize() 将在 Phase 2 实现".to_string()) + Err(SERIALIZE_NOT_IMPLEMENTED.to_string()) } } diff --git a/core/src/repositories/cas_repository.rs b/core/src/repositories/cas_repository.rs index 7b725a7..1b696c5 100644 --- a/core/src/repositories/cas_repository.rs +++ b/core/src/repositories/cas_repository.rs @@ -328,9 +328,8 @@ pub trait CasRepository: Send + Sync { /// - 确保有写入权限 async fn export_to_file(&self, id: &ObjectId, path: &Path) -> crate::Result<()> { let data = self.get(id).await?; - tokio::fs::write(path, data) - .await - .map_err(crate::Error::Io)?; + let write_result = tokio::fs::write(path, data).await.map_err(crate::Error::Io); + write_result?; Ok(()) } } @@ -356,6 +355,7 @@ mod tests { let id: ObjectId = "hash_abc123".to_string(); map.insert(id.clone(), "value"); - assert_eq!(map.get(&id), Some(&"value")); + let value = map.get(&id); + assert_eq!(value, Some(&"value")); } } diff --git a/core/src/repositories/resource_repository.rs b/core/src/repositories/resource_repository.rs index b9f5215..7676e33 100644 --- a/core/src/repositories/resource_repository.rs +++ b/core/src/repositories/resource_repository.rs @@ -366,7 +366,7 @@ pub trait ResourceRepository: Send + Sync { /// # 示例 /// /// ```rust,ignore - /// repo.delete("res_001").await?; + /// let () = repo.delete("res_001").await?; /// ``` async fn delete(&self, id: &str) -> crate::Result<()>; diff --git a/core/src/repositories/translation_repository.rs b/core/src/repositories/translation_repository.rs index 07d6254..83dae73 100644 --- a/core/src/repositories/translation_repository.rs +++ b/core/src/repositories/translation_repository.rs @@ -407,7 +407,7 @@ pub trait TranslationRepository: Send + Sync { /// /// ```rust,ignore /// // 删除错误的翻译 - /// repo.delete("bad_translation_001").await?; + /// let () = repo.delete("bad_translation_001").await?; /// ``` /// /// # 实现建议 diff --git a/crates/bat-cas-engine/src/hash.rs b/crates/bat-cas-engine/src/hash.rs index 15703eb..ad0c494 100644 --- a/crates/bat-cas-engine/src/hash.rs +++ b/crates/bat-cas-engine/src/hash.rs @@ -29,8 +29,9 @@ impl Hash { /// 从十六进制字符串解析 pub fn from_hex(s: &str) -> Result { - let bytes = - hex::decode(s).map_err(|_| crate::error::CasError::InvalidHash(s.to_string()))?; + let decoded = + hex::decode(s).map_err(|_| crate::error::CasError::InvalidHash(s.to_string())); + let bytes = decoded?; if bytes.len() != 32 { return Err(crate::error::CasError::InvalidHash(s.to_string())); @@ -113,6 +114,12 @@ pub async fn compute_file_hash(path: &std::path::Path) -> crate::error::Result (Hash, String) { + let hash = compute_hash(b"Test data"); + let hex = hash.to_hex(); + (hash, hex) + } + #[test] fn test_compute_hash() { let data = b"Hello, World!"; @@ -125,9 +132,7 @@ mod tests { #[test] fn test_hash_to_string() { - let data = b"Test data"; - let hash = compute_hash(data); - let hex = hash.to_hex(); + let (_hash, hex) = test_hash_and_hex(); // 应该是 64 个字符的十六进制字符串 assert_eq!(hex.len(), 64); @@ -135,9 +140,7 @@ mod tests { #[test] fn test_hash_from_string() { - let data = b"Test data"; - let hash = compute_hash(data); - let hex = hash.to_hex(); + let (hash, hex) = test_hash_and_hex(); let parsed = Hash::from_hex(&hex).unwrap(); assert_eq!(hash, parsed); diff --git a/crates/bat-cas-engine/src/refcount.rs b/crates/bat-cas-engine/src/refcount.rs index fcbf41f..2665dc9 100644 --- a/crates/bat-cas-engine/src/refcount.rs +++ b/crates/bat-cas-engine/src/refcount.rs @@ -2,7 +2,7 @@ use crate::error::{CasError, Result}; use crate::hash::Hash; -use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteQueryResult}; use sqlx::SqlitePool; use std::path::Path; use std::str::FromStr; @@ -54,43 +54,50 @@ impl SqliteRefCounter { } async fn init_schema(&self) -> Result<()> { - sqlx::query( - r#" - CREATE TABLE IF NOT EXISTS cas_objects ( - hash TEXT PRIMARY KEY NOT NULL, - size INTEGER NOT NULL CHECK(size >= 0), - ref_count INTEGER NOT NULL CHECK(ref_count >= 0), - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL, - zero_ref_at INTEGER - ) - "#, + let _schema_result = Self::execute_query( + &self.pool, + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS cas_objects ( + hash TEXT PRIMARY KEY NOT NULL, + size INTEGER NOT NULL CHECK(size >= 0), + ref_count INTEGER NOT NULL CHECK(ref_count >= 0), + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + zero_ref_at INTEGER + ) + "#, + ), ) - .execute(&self.pool) .await?; - sqlx::query( - r#" - CREATE INDEX IF NOT EXISTS idx_cas_objects_ref_count - ON cas_objects(ref_count) - "#, + let _index_result = Self::execute_query( + &self.pool, + sqlx::query( + r#" + CREATE INDEX IF NOT EXISTS idx_cas_objects_ref_count + ON cas_objects(ref_count) + "#, + ), ) - .execute(&self.pool) .await?; Ok(()) } - fn now() -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|duration| duration.as_secs() as i64) - .unwrap_or_default() + async fn execute_query<'q>( + pool: &SqlitePool, + query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>, + ) -> Result { + let result = query.execute(pool).await?; + Ok(result) } - /// 注册对象元数据。如果对象已存在，仅更新大小和更新时间。 - pub async fn ensure_object(&self, hash: &Hash, size: u64) -> Result<()> { - let now = Self::now(); + fn insert_or_update_object_query( + hash: &Hash, + size: u64, + now: i64, + ) -> sqlx::query::Query<'_, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'_>> { sqlx::query( r#" INSERT INTO cas_objects(hash, size, ref_count, created_at, updated_at, zero_ref_at) @@ -103,8 +110,20 @@ impl SqliteRefCounter { .bind(hash.to_string()) .bind(size as i64) .bind(now) - .execute(&self.pool) - .await?; + } + + fn now() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs() as i64) + .unwrap_or_default() + } + + /// 注册对象元数据。如果对象已存在，仅更新大小和更新时间。 + pub async fn ensure_object(&self, hash: &Hash, size: u64) -> Result<()> { + let now = Self::now(); + let query = Self::insert_or_update_object_query(hash, size, now); + let _upsert_result = Self::execute_query(&self.pool, query).await?; Ok(()) } diff --git a/crates/bat-cas-engine/src/repository.rs b/crates/bat-cas-engine/src/repository.rs index 0975dc2..fbcf1c5 100644 --- a/crates/bat-cas-engine/src/repository.rs +++ b/crates/bat-cas-engine/src/repository.rs @@ -76,7 +76,8 @@ impl FileSystemCasRepository { /// 读取对象并验证 Hash。 pub async fn get(&self, hash: &Hash) -> Result> { - self.storage.get(hash).await + let data = self.storage.get(hash).await?; + Ok(data) } /// 检查对象是否存在。 @@ -92,7 +93,8 @@ impl FileSystemCasRepository { let size = self.storage.size(hash).await?; self.ref_counter.ensure_object(hash, size).await?; - self.ref_counter.add_reference(hash).await + let count = self.ref_counter.add_reference(hash).await?; + Ok(count) } /// 减少引用计数。 @@ -146,10 +148,15 @@ mod tests { use super::*; use std::sync::Arc; - #[tokio::test] - async fn store_increments_reference_count() { + async fn temp_repo() -> (tempfile::TempDir, FileSystemCasRepository) { let temp_dir = tempfile::tempdir().unwrap(); let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + (temp_dir, repo) + } + + #[tokio::test] + async fn store_increments_reference_count() { + let (_temp_dir, repo) = temp_repo().await; let hash = repo.store(b"same data").await.unwrap(); let second = repo.store(b"same data").await.unwrap(); @@ -161,8 +168,7 @@ mod tests { #[tokio::test] async fn gc_deletes_only_zero_reference_objects() { - let temp_dir = tempfile::tempdir().unwrap(); - let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, repo) = temp_repo().await; let keep = repo.store(b"keep").await.unwrap(); let delete = repo.store(b"delete").await.unwrap(); @@ -180,8 +186,8 @@ mod tests { #[tokio::test] async fn concurrent_store_is_safe_and_counts_references() { - let temp_dir = tempfile::tempdir().unwrap(); - let repo = Arc::new(FileSystemCasRepository::new(temp_dir.path()).await.unwrap()); + let (_temp_dir, repo) = temp_repo().await; + let repo = Arc::new(repo); let mut tasks = Vec::new(); for _ in 0..32 { @@ -203,13 +209,11 @@ mod tests { #[tokio::test] async fn corrupted_object_is_detected_through_repository() { - let temp_dir = tempfile::tempdir().unwrap(); - let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, repo) = temp_repo().await; let hash = repo.store(b"valid").await.unwrap(); - tokio::fs::write(repo.storage().object_path(&hash), b"invalid") - .await - .unwrap(); + let write_result = tokio::fs::write(repo.storage().object_path(&hash), b"invalid").await; + write_result.unwrap(); assert!(matches!( repo.get(&hash).await, @@ -219,8 +223,7 @@ mod tests { #[tokio::test] async fn add_reference_requires_existing_object() { - let temp_dir = tempfile::tempdir().unwrap(); - let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, repo) = temp_repo().await; let missing = compute_hash(b"missing"); assert!(matches!( @@ -231,8 +234,7 @@ mod tests { #[tokio::test] async fn remove_reference_does_not_go_below_zero() { - let temp_dir = tempfile::tempdir().unwrap(); - let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, repo) = temp_repo().await; let hash = repo.store(b"underflow").await.unwrap(); assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0); diff --git a/crates/bat-cas-engine/src/storage.rs b/crates/bat-cas-engine/src/storage.rs index 99341b3..80de5d0 100644 --- a/crates/bat-cas-engine/src/storage.rs +++ b/crates/bat-cas-engine/src/storage.rs @@ -108,16 +108,33 @@ impl FileSystemStorage { } async fn write_temp_file(&self, path: &Path, data: &[u8]) -> Result<()> { - let mut file = tokio::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(path) - .await?; + let mut file = tokio::fs::File::create_new(path).await?; file.write_all(data).await?; file.sync_all().await?; Ok(()) } + async fn validate_existing_object(&self, hash: &Hash) -> Result { + if !self.exists(hash).await { + return Ok(false); + } + + let existing = self.get(hash).await; + match existing { + Ok(_) => Ok(true), + Err(CasError::HashMismatch { .. }) => Ok(false), + Err(error) => Err(error), + } + } + + async fn remove_temp_file(path: &Path) -> Result<()> { + match tokio::fs::remove_file(path).await { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(CasError::Io(error)), + } + } + async fn child_directories(path: &Path) -> Result> { let mut directories = Vec::new(); let mut entries = tokio::fs::read_dir(path).await?; @@ -164,6 +181,30 @@ impl FileSystemStorage { Ok(shard_directories) } + + async fn commit_temp_file(&self, tmp_path: &Path, path: &Path, hash: &Hash) -> Result { + let parent = path + .parent() + .ok_or_else(|| CasError::Other(anyhow::anyhow!("Object path has no parent")))?; + + let rename_result = tokio::fs::rename(tmp_path, path).await; + match rename_result { + Ok(()) => { + Self::sync_directory(parent)?; + Ok(*hash) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + Self::remove_temp_file(tmp_path).await?; + let stored_data = self.get(hash).await?; + drop(stored_data); + Ok(*hash) + } + Err(error) => { + Self::remove_temp_file(tmp_path).await?; + Err(CasError::Io(error)) + } + } + } } #[async_trait] @@ -172,12 +213,8 @@ impl Storage for FileSystemStorage { let hash = compute_hash(data); let path = self.object_path(&hash); - if self.exists(&hash).await { - match self.get(&hash).await { - Ok(_) => return Ok(hash), - Err(CasError::HashMismatch { .. }) => {} - Err(error) => return Err(error), - } + if self.validate_existing_object(&hash).await? { + return Ok(hash); } let parent = path @@ -188,22 +225,7 @@ impl Storage for FileSystemStorage { let tmp_path = self.temp_path(&hash); self.write_temp_file(&tmp_path, data).await?; - - match tokio::fs::rename(&tmp_path, &path).await { - Ok(()) => { - Self::sync_directory(parent)?; - Ok(hash) - } - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { - let _ = tokio::fs::remove_file(&tmp_path).await; - self.get(&hash).await?; - Ok(hash) - } - Err(error) => { - let _ = tokio::fs::remove_file(&tmp_path).await; - Err(CasError::Io(error)) - } - } + self.commit_temp_file(&tmp_path, &path, &hash).await } async fn get(&self, hash: &Hash) -> Result> { @@ -288,10 +310,15 @@ impl Storage for FileSystemStorage { mod tests { use super::*; - #[tokio::test] - async fn put_and_get_roundtrip() { + async fn temp_storage() -> (tempfile::TempDir, FileSystemStorage) { let temp_dir = tempfile::tempdir().unwrap(); let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); + (temp_dir, storage) + } + + #[tokio::test] + async fn put_and_get_roundtrip() { + let (_temp_dir, storage) = temp_storage().await; let data = b"Hello, CAS!"; let hash = storage.put(data).await.unwrap(); @@ -303,8 +330,7 @@ mod tests { #[tokio::test] async fn put_is_deduplicated() { - let temp_dir = tempfile::tempdir().unwrap(); - let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, storage) = temp_storage().await; let data = b"Same content"; let hash1 = storage.put(data).await.unwrap(); @@ -317,25 +343,23 @@ mod tests { #[tokio::test] async fn exists_and_delete() { - let temp_dir = tempfile::tempdir().unwrap(); - let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, storage) = temp_storage().await; let hash = storage.put(b"Test data").await.unwrap(); assert!(storage.exists(&hash).await); - storage.delete(&hash).await.unwrap(); + let deleted = storage.delete(&hash).await; + deleted.unwrap(); assert!(!storage.exists(&hash).await); } #[tokio::test] async fn corrupted_object_is_rejected() { - let temp_dir = tempfile::tempdir().unwrap(); - let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); + let (_temp_dir, storage) = temp_storage().await; let hash = storage.put(b"valid data").await.unwrap(); - tokio::fs::write(storage.object_path(&hash), b"corrupted") - .await - .unwrap(); + let write_result = tokio::fs::write(storage.object_path(&hash), b"corrupted").await; + write_result.unwrap(); let error = storage.get(&hash).await.unwrap_err(); assert!(matches!(error, CasError::HashMismatch { .. })); @@ -345,7 +369,8 @@ mod tests { async fn init_reports_path_errors() { let temp_dir = tempfile::tempdir().unwrap(); let file_path = temp_dir.path().join("not-a-directory"); - tokio::fs::write(&file_path, b"file").await.unwrap(); + let write_result = tokio::fs::write(&file_path, b"file").await; + write_result.unwrap(); let result = FileSystemStorage::new(file_path.join("cas")).await; assert!(matches!(result, Err(CasError::Io(_)))); diff --git a/infrastructure/src/cas/filesystem.rs b/infrastructure/src/cas/filesystem.rs index b9b0ace..c994ac0 100644 --- a/infrastructure/src/cas/filesystem.rs +++ b/infrastructure/src/cas/filesystem.rs @@ -83,11 +83,13 @@ impl CasRepository for FileSystemCasRepository { async fn get(&self, id: &ObjectId) -> bat_core::Result> { let hash = Self::parse_object_id(id)?; - self.engine() + let data = self + .engine() .await? .get(&hash) .await - .map_err(Self::map_error) + .map_err(Self::map_error)?; + Ok(data) } async fn exists(&self, id: &ObjectId) -> bool { @@ -175,10 +177,8 @@ mod tests { assert_eq!(repo.gc().await.unwrap(), 1); assert!(!repo.exists(&id).await); - assert!(matches!( - repo.get(&id).await, - Err(bat_core::Error::NotFound(_)) - )); + let missing = repo.get(&id).await; + assert!(matches!(missing, Err(bat_core::Error::NotFound(_)))); } #[tokio::test]