refactor: update database handling and introduce BlobStore trait for improved abstraction and flexibility

Modified lib/auth/src/service.rs +2 −2 View lib/auth/src/service.rs

@@ -2,7 +2,7 @@

22use std::time::Duration;

33

44use config::AppConfig;

5use export::{AppDatabase, AppRedis, AppState};

5use export::{AppRedis, AppState, Database};

66use tonic::Request;

77

88use crate::AuthSvc;

@@ -55,7 +55,7 @@

5555 &self.state.config

5656 }

5757

58 pub fn db(&self) -> &Arc<AppDatabase> {

58 pub fn db(&self) -> &Arc<dyn Database> {

5959 &self.state.database

6060 }

6161

Modified lib/chat/src/services/service.rs +5 −5 View lib/chat/src/services/service.rs

@@ -3,9 +3,9 @@

33

44use config::AppConfig;

55use export::AppState;

6use export::database::AppDatabase;

6use export::database::Database;

77use space::services::spaces::UserDto;

8use storage::S3Client;

8use storage::BlobStore;

99

1010use crate::errors::ChatError;

1111use crate::share::realtime::Realtime;

@@ -26,12 +26,12 @@

2626 &self.state.config

2727 }

2828

29 pub fn db(&self) -> &Arc<AppDatabase> {

29 pub fn db(&self) -> &Arc<dyn Database> {

3030 &self.state.database

3131 }

3232

33 pub fn storage(&self) -> &S3Client {

34 &self.state.storage

33 pub fn storage(&self) -> &dyn BlobStore {

34 self.state.storage.as_ref()

3535 }

3636

3737 pub async fn load_user_briefs(&self, uids: &[i64]) -> Result<HashMap<i64, UserDto>, ChatError> {

Modified lib/export/src/database.rs +36 −3 View lib/export/src/database.rs

@@ -3,7 +3,10 @@

33use crate::error::ExportError;

44use config::database::{DatabaseConfig, DatabaseRole as ConfigRole};

55use sea_orm::prelude::async_trait::async_trait;

6use sea_orm::{Database, DatabaseConnection, DbBackend, DbErr, ExecResult, QueryResult, Statement};

6use sea_orm::{

7 Database as SeaDatabase, DatabaseConnection, DbBackend, DbErr, ExecResult, QueryResult,

8 Statement,

9};

710

811#[derive(Debug, Clone, Copy, PartialEq, Eq)]

912pub enum DatabaseRole {

@@ -29,6 +32,36 @@

2932 }

3033}

3134

35pub trait Database: Send + Sync {

36 fn config(&self) -> &DatabaseConfig;

37 fn primary(&self) -> &DatabaseConnection;

38 fn replicas(&self) -> &[DatabaseConnection];

39 fn read(&self) -> &DatabaseConnection;

40 fn write(&self) -> &DatabaseConnection;

41}

42

43impl Database for AppDatabase {

44 fn config(&self) -> &DatabaseConfig {

45 AppDatabase::config(self)

46 }

47

48 fn primary(&self) -> &DatabaseConnection {

49 AppDatabase::primary(self)

50 }

51

52 fn replicas(&self) -> &[DatabaseConnection] {

53 AppDatabase::replicas(self)

54 }

55

56 fn read(&self) -> &DatabaseConnection {

57 AppDatabase::read(self)

58 }

59

60 fn write(&self) -> &DatabaseConnection {

61 AppDatabase::write(self)

62 }

63}

64

3265pub struct AppDatabase {

3366 config: DatabaseConfig,

3467 primary: DatabaseConnection,

@@ -43,14 +76,14 @@

4376 let primary_opts = config

4477 .primary_connect_options()

4578 .map_err(ExportError::InvalidConfig)?;

46 let primary = Database::connect(primary_opts).await?;

79 let primary = SeaDatabase::connect(primary_opts).await?;

4780

4881 let mut replicas = Vec::with_capacity(config.replica_count());

4982 for i in 0..config.replica_count() {

5083 let opts = config

5184 .replica_connect_options(i)

5285 .map_err(ExportError::InvalidConfig)?;

53 let conn = Database::connect(opts).await?;

86 let conn = SeaDatabase::connect(opts).await?;

5487 replicas.push(conn);

5588 }

5689

Modified lib/export/src/health.rs +3 −3 View lib/export/src/health.rs

@@ -4,11 +4,11 @@

44use sea_orm::{ConnectionTrait, DbBackend, Statement};

55

66use crate::http::{Router, State, get};

7use crate::{AppDatabase, AppRedis, AppState, AuthApi};

7use crate::{AppRedis, AppState, AuthApi, Database};

88

99#[derive(Clone)]

1010struct HealthState {

11 database: Option<Arc<AppDatabase>>,

11 database: Option<Arc<dyn Database>>,

1212 redis: AppRedis,

1313 kenauth: Option<Arc<dyn AuthApi>>,

1414}

@@ -51,7 +51,7 @@

5151}

5252

5353pub fn router_with(

54 database: Option<Arc<AppDatabase>>,

54 database: Option<Arc<dyn Database>>,

5555 redis: AppRedis,

5656 kenauth: Option<Arc<dyn AuthApi>>,

5757) -> Router {

Modified lib/export/src/imaging.rs +6 −4 View lib/export/src/imaging.rs

@@ -14,14 +14,16 @@

1414use imaging::preset::PreviewPreset;

1515use imaging::{PreviewOutcome, render_preview};

1616use serde_json::Value;

17use storage::S3Client;

1817

1918use crate::redis::{self, AppRedis};

2019

2120/// Look up an already-stored derivative. `key_base` comes from

2221/// [`preview_key_base`]; because the encoder choice (JPEG vs WebP) is not

2322/// derivable from the original's extension, both candidates are probed.

24pub async fn find_stored_preview(client: &S3Client, key_base: &str) -> Option<PreviewState> {

23pub async fn find_stored_preview(

24 client: &dyn storage::BlobStore,

25 key_base: &str,

26) -> Option<PreviewState> {

2527 for candidate in preview_candidate_keys(key_base) {

2628 match client.head(&candidate).await {

2729 Ok(Some((bytes, content_type))) => {

@@ -48,7 +50,7 @@

4850/// `None` means "nothing recorded", which is exactly the legacy-row case this is

4951/// meant to heal without a batch job.

5052pub async fn ensure_preview(

51 client: &S3Client,

53 client: &dyn storage::BlobStore,

5254 redis: &AppRedis,

5355 key_base: &str,

5456 original_key: &str,

@@ -104,7 +106,7 @@

104106/// Read the original, render a derivative, store it, and drop the other

105107/// encoding so a JPEG->WebP switch cannot leave two "current" previews behind.

106108async fn generate_preview(

107 client: &S3Client,

109 client: &dyn storage::BlobStore,

108110 key_base: &str,

109111 original_key: &str,

110112 content_type: &str,

Modified lib/export/src/lib.rs +27 −9 View lib/export/src/lib.rs

@@ -24,18 +24,18 @@

2424pub use api_error::is_unique_violation;

2525pub use auth_api::{AuthApi, KenAuthError};

2626use config::AppConfig;

27pub use database::{AppDatabase, DatabaseRole};

27pub use database::{AppDatabase, Database, DatabaseRole};

2828pub use error::ExportError;

2929pub use redis::{AppRedis, RedisConnection};

30pub use storage::S3Client;

30pub use storage::{BlobStore, S3Client};

3131pub use urls::{resolve_space_image_url, resolve_user_avatar_url};

3232

3333#[derive(Clone)]

3434pub struct AppState {

35 pub database: Arc<AppDatabase>,

35 pub database: Arc<dyn Database>,

3636 pub redis: AppRedis,

3737 pub cache: ::cache::Cache,

38 pub storage: Arc<S3Client>,

38 pub storage: Arc<dyn BlobStore>,

3939 pub config: AppConfig,

4040 pub kenauth: Option<Arc<dyn AuthApi>>,

4141 pub mail: Option<::email::MailPublisher>,

@@ -50,13 +50,14 @@

5050 }

5151

5252 pub async fn with_config(config: AppConfig) -> Result<Self, ExportError> {

53 let database = Arc::new(AppDatabase::connect(config.database.clone()).await?);

53 let database: Arc<dyn Database> =

54 Arc::new(AppDatabase::connect(config.database.clone()).await?);

5455 Self::with_identity(config, database, None, None, None).await

5556 }

5657

5758 pub async fn with_identity(

5859 config: AppConfig,

59 database: Arc<AppDatabase>,

60 database: Arc<dyn Database>,

6061 identity: Option<Arc<dyn AuthApi>>,

6162 mail: Option<::email::MailPublisher>,

6263 events: Option<::events::EventHandler>,

@@ -68,14 +69,31 @@

6869 // 入队点遍布各业务 crate，但只有这里有配置：把开关写进句柄，让

6970 // `export::push::enqueue` 能自己判断该不该产生任务。

7071 redis.set_push_enabled(config.push.enabled);

71 let storage = Arc::new(S3Client::new(&config.s3).map_err(ExportError::from)?);

72 let storage: Arc<dyn BlobStore> =

73 Arc::new(S3Client::new(&config.s3).map_err(ExportError::from)?);

7274

7375 let kenauth = identity;

7476 if let Some(client) = &kenauth {

7577 identity_client::install(client.clone());

7678 }

7779

78 Ok(Self {

80 Ok(Self::from_parts(

81 database, redis, cache, storage, config, kenauth, mail, events,

82 ))

83 }

84

85 #[allow(clippy::too_many_arguments)]

86 pub fn from_parts(

87 database: Arc<dyn Database>,

88 redis: AppRedis,

89 cache: ::cache::Cache,

90 storage: Arc<dyn BlobStore>,

91 config: AppConfig,

92 kenauth: Option<Arc<dyn AuthApi>>,

93 mail: Option<::email::MailPublisher>,

94 events: Option<::events::EventHandler>,

95 ) -> Self {

96 Self {

7997 database,

8098 redis,

8199 cache,

@@ -84,6 +102,6 @@

84102 kenauth,

85103 mail,

86104 events,

87 })

105 }

88106 }

89107}

Modified lib/git/src/context/facade.rs +2 −2 View lib/git/src/context/facade.rs

@@ -2,7 +2,7 @@

22use std::sync::Arc;

33

44use ::cache::Cache;

5use export::AppDatabase;

5use export::Database;

66use models::repo::repos;

77use proto::kenshell::common::v1::{GitActor, RepositoryRef, WriteOptions};

88use uuid::Uuid;

@@ -11,7 +11,7 @@

1111pub struct GitContext {

1212 pub model: repos::Model,

1313 pub cache: Cache,

14 pub database: Arc<AppDatabase>,

14 pub database: Arc<dyn Database>,

1515 pub space_name: String,

1616 pub repo_root: PathBuf,

1717 pub ks: proto::KenShellClient,

Modified lib/git/src/service/branches.rs +2 −2 View lib/git/src/service/branches.rs

@@ -67,7 +67,7 @@

6767 .await

6868 .map_err(map_baby)?;

6969 let mut rows = self.branch_rows(&ctx).await?;

70 let policy = load_branch_policy(&self.state.database, ctx.model.id).await?;

70 let policy = load_branch_policy(self.state.database.as_ref(), ctx.model.id).await?;

7171

7272 for b in &branches {

7373 if !rows.contains_key(&b.name) && !b.name.is_empty() {

@@ -218,7 +218,7 @@

218218 force: bool,

219219 ) -> Result<(), GitError> {

220220 let name = validate::validate_branch_name(&name)?;

221 let policy = load_branch_policy(&self.state.database, ctx.model.id).await?;

221 let policy = load_branch_policy(self.state.database.as_ref(), ctx.model.id).await?;

222222 if policy.is_match(&name) {

223223 return Err(GitError::Forbidden);

224224 }

Modified lib/git/src/service/refs.rs +4 −3 View lib/git/src/service/refs.rs

@@ -100,7 +100,7 @@

100100 .list_branches(crate::rpcgit::branch::ListBranchesOptions::default())

101101 .await

102102 .map_err(map_baby)?;

103 let policy = load_branch_policy(&self.state.database, ctx.model.id).await?;

103 let policy = load_branch_policy(self.state.database.as_ref(), ctx.model.id).await?;

104104 let mut out: Vec<BranchInfoDto> = branches

105105 .into_iter()

106106 .map(|b| BranchInfoDto {

@@ -189,12 +189,13 @@

189189 name: &str,

190190 ) -> Result<bool, GitError> {

191191 if let Some(branch) = name.strip_prefix("refs/heads/") {

192 let policy = load_branch_policy(&self.state.database, ctx.model.id).await?;

192 let policy = load_branch_policy(self.state.database.as_ref(), ctx.model.id).await?;

193193 return Ok(policy.is_match(branch));

194194 }

195195 if let Some(tag) = name.strip_prefix("refs/tags/") {

196196 let policy =

197 crate::share::policy::load_tag_policy(&self.state.database, ctx.model.id).await?;

197 crate::share::policy::load_tag_policy(self.state.database.as_ref(), ctx.model.id)

198 .await?;

198199 return Ok(policy.is_match(tag));

199200 }

200201 Ok(false)

Modified lib/git/src/service/tags.rs +4 −2 View lib/git/src/service/tags.rs

@@ -76,7 +76,8 @@

7676 "invalid tag name or target".to_string(),

7777 ));

7878 }

79 let tag_policy = policy::load_tag_policy(&self.state.database, ctx.model.id).await?;

79 let tag_policy =

80 policy::load_tag_policy(self.state.database.as_ref(), ctx.model.id).await?;

8081 if tag_policy.is_match(&cmd.name) {

8182 return Err(GitError::Forbidden);

8283 }

@@ -204,7 +205,8 @@

204205 }

205206

206207 pub async fn delete_tag(&self, ctx: &GitContext, name: String) -> Result<(), GitError> {

207 let tag_policy = policy::load_tag_policy(&self.state.database, ctx.model.id).await?;

208 let tag_policy =

209 policy::load_tag_policy(self.state.database.as_ref(), ctx.model.id).await?;

208210 if tag_policy.is_match(&name) {

209211 return Err(GitError::Forbidden);

210212 }

Modified lib/git/src/share/policy.rs +2 −2 View lib/git/src/share/policy.rs

@@ -35,7 +35,7 @@

3535}

3636

3737pub async fn load_branch_policy(

38 db: &export::AppDatabase,

38 db: &dyn export::Database,

3939 repo_id: Uuid,

4040) -> Result<BranchPolicy, GitError> {

4141 let rows = repo_branch_protection_policy::Entity::find()

@@ -46,7 +46,7 @@

4646}

4747

4848pub async fn load_tag_policy(

49 db: &export::AppDatabase,

49 db: &dyn export::Database,

5050 repo_id: Uuid,

5151) -> Result<BranchPolicy, GitError> {

5252 let rows = repo_tag_protection_policy::Entity::find()

Modified lib/issues/src/services/service.rs +2 −2 View lib/issues/src/services/service.rs

@@ -3,7 +3,7 @@

33

44use config::AppConfig;

55use export::AppState;

6use export::database::AppDatabase;

6use export::database::Database;

77use i18n;

88use models::issue::issue_event_line::{self, IssueEventType};

99use models::issue::issue_item;

@@ -50,7 +50,7 @@

5050 )

5151 }

5252

53 pub fn db(&self) -> &Arc<AppDatabase> {

53 pub fn db(&self) -> &Arc<dyn Database> {

5454 &self.state.database

5555 }

5656

Modified lib/space/src/services/service.rs +5 −5 View lib/space/src/services/service.rs

@@ -1,8 +1,8 @@

11use std::sync::Arc;

22

33use config::AppConfig;

4use export::{AppDatabase, AppState};

5use storage::S3Client;

4use export::{AppState, Database};

5use storage::BlobStore;

66

77use crate::errors::SpaceError;

88

@@ -20,12 +20,12 @@

2020 &self.state.config

2121 }

2222

23 pub fn db(&self) -> &Arc<AppDatabase> {

23 pub fn db(&self) -> &Arc<dyn Database> {

2424 &self.state.database

2525 }

2626

27 pub fn storage(&self) -> &Arc<S3Client> {

28 &self.state.storage

27 pub fn storage(&self) -> &dyn BlobStore {

28 self.state.storage.as_ref()

2929 }

3030

3131 /// Language for one recipient's space mail.

Modified lib/storage/src/lib.rs +2 −0 View lib/storage/src/lib.rs

@@ -1,5 +1,7 @@

11pub mod error;

22pub mod s3;

3pub mod store;

34

45pub use error::StorageError;

56pub use s3::{MAX_PRESIGN_EXPIRY_SECS, ObjectInfo, ObjectPage, S3Client};

7pub use store::BlobStore;

Added lib/storage/src/store.rs +88 −0 View lib/storage/src/store.rs

@@ -0,0 +1,88 @@

1use std::pin::Pin;

2

3use bytes::Bytes;

4use futures::Stream;

5use http::HeaderMap;

6

7use crate::error::StorageError;

8use crate::s3::S3Client;

9

10#[async_trait::async_trait]

11pub trait BlobStore: Send + Sync {

12 fn bucket_name(&self) -> &str;

13

14 async fn put_with_content_type(

15 &self,

16 key: &str,

17 body: &[u8],

18 content_type: &str,

19 ) -> Result<(), StorageError>;

20

21 async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError>;

22

23 async fn get_stream(

24 &self,

25 key: &str,

26 ) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send>>, StorageError>;

27

28 async fn delete(&self, key: &str) -> Result<(), StorageError>;

29

30 async fn head(&self, key: &str) -> Result<Option<(u64, String)>, StorageError>;

31

32 fn presign_get(&self, key: &str, expiry_secs: u32) -> Result<String, StorageError>;

33

34 fn presign_put_with_headers(

35 &self,

36 key: &str,

37 expiry_secs: u32,

38 custom_headers: HeaderMap,

39 ) -> Result<String, StorageError>;

40}

41

42#[async_trait::async_trait]

43impl BlobStore for S3Client {

44 fn bucket_name(&self) -> &str {

45 S3Client::bucket_name(self)

46 }

47

48 async fn put_with_content_type(

49 &self,

50 key: &str,

51 body: &[u8],

52 content_type: &str,

53 ) -> Result<(), StorageError> {

54 S3Client::put_with_content_type(self, key, body, content_type).await

55 }

56

57 async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError> {

58 S3Client::get(self, key).await

59 }

60

61 async fn get_stream(

62 &self,

63 key: &str,

64 ) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send>>, StorageError> {

65 S3Client::get_stream(self, key).await

66 }

67

68 async fn delete(&self, key: &str) -> Result<(), StorageError> {

69 S3Client::delete(self, key).await

70 }

71

72 async fn head(&self, key: &str) -> Result<Option<(u64, String)>, StorageError> {

73 S3Client::head(self, key).await

74 }

75

76 fn presign_get(&self, key: &str, expiry_secs: u32) -> Result<String, StorageError> {

77 S3Client::presign_get(self, key, expiry_secs)

78 }

79

80 fn presign_put_with_headers(

81 &self,

82 key: &str,

83 expiry_secs: u32,

84 custom_headers: HeaderMap,

85 ) -> Result<String, StorageError> {

86 S3Client::presign_put_with_headers(self, key, expiry_secs, custom_headers)

87 }

88}

Modified lib/storage/Cargo.toml +1 −0 View lib/storage/Cargo.toml

@@ -8,6 +8,7 @@

88rust-s3-async = { workspace = true }

99tokio = { workspace = true, features = ["macros", "rt", "io-util"] }

1010http = { workspace = true }

11async-trait = { workspace = true }

1112futures = { workspace = true }

1213bytes = { workspace = true }

1314tracing = { workspace = true }

No newline at end of file

Modified lib/user/src/services/service.rs +5 −5 View lib/user/src/services/service.rs

@@ -1,8 +1,8 @@

11use std::sync::Arc;

22

33use config::AppConfig;

4use export::{AppDatabase, AppState};

5use storage::S3Client;

4use export::{AppState, Database};

5use storage::BlobStore;

66

77#[derive(Clone)]

88pub struct UserService {

@@ -18,11 +18,11 @@

1818 &self.state.config

1919 }

2020

21 pub fn db(&self) -> &Arc<AppDatabase> {

21 pub fn db(&self) -> &Arc<dyn Database> {

2222 &self.state.database

2323 }

2424

25 pub fn storage(&self) -> &Arc<S3Client> {

26 &self.state.storage

25 pub fn storage(&self) -> &dyn BlobStore {

26 self.state.storage.as_ref()

2727 }

2828}

Modified Cargo.lock +1 −0 View Cargo.lock

@@ -7451,6 +7451,7 @@

74517451name = "storage"

74527452version = "1.0.0"

74537453dependencies = [

7454 "async-trait",

74547455 "bytes",

74557456 "config 1.0.0",

74567457 "futures",