refactor: update storage client references for improved consistency and clarity

Modified lib/chat/src/services/attachments.rs +7 −8 View lib/chat/src/services/attachments.rs

@@ -197,7 +197,7 @@

197197 let id = Uuid::now_v7();

198198 let filename = sanitize_filename(&cmd.filename);

199199 let key = format!("{ATTACHMENT_PREFIX}/{caller}/{id}/{filename}");

200 let bucket = self.storage().client().bucket_name().to_string();

200 let bucket = self.storage().bucket_name().to_string();

201201

202202 let row = attachments::ActiveModel {

203203 id: Set(id),

@@ -228,7 +228,6 @@

228228 }

229229 let upload_url = self

230230 .storage()

231 .client()

232231 .presign_put_with_headers(&key, ttl as u32, put_headers)

233232 .map_err(ChatError::from)?;

234233

@@ -256,7 +255,7 @@

256255 "attachment already finalized".to_string(),

257256 ));

258257 }

259 let head = self.storage().client().head(&row.s3_key).await?;

258 let head = self.storage().head(&row.s3_key).await?;

260259 let Some((actual_size, actual_content_type)) = head else {

261260 return Err(ChatError::BadRequest("object not uploaded yet".to_string()));

262261 };

@@ -305,7 +304,7 @@

305304 async fn generate_attachment_preview(&self, row: &attachments::Model) -> Result<(), ChatError> {

306305 let key_base = export::imaging::preview_key_base_of(&row.s3_key);

307306 let preview = export::imaging::ensure_preview(

308 self.storage().client(),

307 self.storage(),

309308 &self.state.redis,

310309 &key_base,

311310 &row.s3_key,

@@ -343,7 +342,7 @@

343342 async fn delete_attachment_previews(&self, original_key: &str) {

344343 let key_base = export::imaging::preview_key_base_of(original_key);

345344 for candidate in imaging::layout::preview_candidate_keys(&key_base) {

346 if let Err(error) = self.storage().client().delete(&candidate).await {

345 if let Err(error) = self.storage().delete(&candidate).await {

347346 tracing::warn!(%error, key = %candidate, "failed to delete attachment preview");

348347 }

349348 }

@@ -363,7 +362,7 @@

363362 let key_base = export::imaging::preview_key_base_of(&row.s3_key);

364363 let recorded = imaging::layout::preview_from_variants(&row.variants);

365364 let state = export::imaging::ensure_preview(

366 self.storage().client(),

365 self.storage(),

367366 &self.state.redis,

368367 &key_base,

369368 &row.s3_key,

@@ -423,7 +422,7 @@

423422 .exec(self.db().write())

424423 .await;

425424 if shared == 0 {

426 let _ = self.storage().client().delete(&row.s3_key).await;

425 let _ = self.storage().delete(&row.s3_key).await;

427426 self.delete_attachment_previews(&row.s3_key).await;

428427 }

429428 }

@@ -469,7 +468,7 @@

469468 return Err(ChatError::NotFound);

470469 }

471470 let (key, content_type) = self.attachment_variant_locator(&row, variant).await;

472 let stream = self.storage().client().get_stream(&key).await?;

471 let stream = self.storage().get_stream(&key).await?;

473472 let body = Box::pin(futures::StreamExt::map(stream, |item| {

474473 item.map(Bytes::from)

475474 .map_err(|e| std::io::Error::other(e.to_string()))

Modified lib/chat/src/services/emojis.rs +3 −4 View lib/chat/src/services/emojis.rs

@@ -141,7 +141,6 @@

141141 };

142142 let key = format!("{EMOJI_PREFIX}/{}/{id}.{ext}", space.id);

143143 self.storage()

144 .client()

145144 .put_with_content_type(&key, &bytes, &cmd.content_type)

146145 .await

147146 .map_err(ChatError::from)?;

@@ -152,7 +151,7 @@

152151 name: Set(name),

153152 lower_name: Set(lower),

154153 user_id: Set(caller),

155 bucket: Set(self.storage().client().bucket_name().to_string()),

154 bucket: Set(self.storage().bucket_name().to_string()),

156155 s3_key: Set(key),

157156 animated: Set(cmd.animated),

158157 disabled: Set(false),

@@ -188,7 +187,7 @@

188187 emojis::Entity::delete_by_id(row.id)

189188 .exec(self.db().write())

190189 .await?;

191 if let Err(error) = self.storage().client().delete(&row.s3_key).await {

190 if let Err(error) = self.storage().delete(&row.s3_key).await {

192191 tracing::warn!(%error, "failed to delete emoji object");

193192 }

194193 Ok(())

@@ -204,6 +203,6 @@

204203 .one(self.db().read())

205204 .await?

206205 .ok_or(ChatError::NotFound)?;

207 Ok(self.storage().client().get(&row.s3_key).await?)

206 Ok(self.storage().get(&row.s3_key).await?)

208207 }

209208}

Modified lib/chat/src/services/messages.rs +2 −2 View lib/chat/src/services/messages.rs

@@ -24,7 +24,7 @@

2424async fn delete_attachment_previews(service: &ChatService, original_key: &str) {

2525 let key_base = export::imaging::preview_key_base_of(original_key);

2626 for candidate in imaging::layout::preview_candidate_keys(&key_base) {

27 if let Err(error) = service.storage().client().delete(&candidate).await {

27 if let Err(error) = service.storage().delete(&candidate).await {

2828 tracing::warn!(%error, key = %candidate, "failed to delete attachment preview");

2929 }

3030 }

@@ -1227,7 +1227,7 @@

12271227 .exec(self.db().write())

12281228 .await;

12291229 if shared == 0 {

1230 let _ = self.storage().client().delete(&attachment.s3_key).await;

1230 let _ = self.storage().delete(&attachment.s3_key).await;

12311231 // The derivative is an orphan once the last reference to the

12321232 // original is gone; leaving it behind would leak an object that

12331233 // nothing can ever address again.

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

@@ -4,8 +4,8 @@

44use config::AppConfig;

55use export::AppState;

66use export::database::AppDatabase;

7use export::storage::AppStorage;

87use space::services::spaces::UserDto;

8use storage::S3Client;

99

1010use crate::errors::ChatError;

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

@@ -30,7 +30,7 @@

3030 &self.state.database

3131 }

3232

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

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

3434 &self.state.storage

3535 }

3636

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

@@ -4,5 +4,7 @@

44pub mod migrate;

55mod routing;

66

7pub use routing::requires_primary;

8

79pub use conn::DbConn;

810pub use error::{DbError, DbResult};

Modified lib/db/src/routing.rs +1 −1 View lib/db/src/routing.rs

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

22use sqlparser::keywords::Keyword;

33use sqlparser::tokenizer::{Token, Tokenizer};

44

5pub(super) fn requires_primary(sql: &str) -> bool {

5pub fn requires_primary(sql: &str) -> bool {

66 let Ok(tokens) = Tokenizer::new(&PostgreSqlDialect {}, sql).tokenize() else {

77 return true;

88 };

Deleted lib/export/src/cache.rs +0 −125

@@ -1,125 +0,0 @@

1use std::fmt;

2use std::time::Duration;

3

4use ::redis::RedisError;

5use serde::Serialize;

6use serde::de::DeserializeOwned;

7

8use crate::redis::{self, AppRedis, RedisConnection};

9

10#[derive(Debug)]

11pub enum CacheError {

12 Serialization(String),

13 Backend(RedisError),

14}

15

16impl fmt::Display for CacheError {

17 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

18 match self {

19 CacheError::Serialization(msg) => write!(f, "cache serialization error: {msg}"),

20 CacheError::Backend(err) => write!(f, "cache backend error: {err}"),

21 }

22 }

23}

24

25impl std::error::Error for CacheError {

26 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {

27 match self {

28 CacheError::Backend(err) => Some(err),

29 _ => None,

30 }

31 }

32}

33

34impl From<RedisError> for CacheError {

35 fn from(err: RedisError) -> Self {

36 CacheError::Backend(err)

37 }

38}

39

40impl From<serde_json::Error> for CacheError {

41 fn from(err: serde_json::Error) -> Self {

42 CacheError::Serialization(err.to_string())

43 }

44}

45

46#[derive(Clone)]

47pub struct CacheClient {

48 redis: AppRedis,

49}

50

51impl CacheClient {

52 pub fn new(redis: AppRedis) -> Self {

53 Self { redis }

54 }

55

56 pub fn raw(&self) -> &AppRedis {

57 &self.redis

58 }

59

60 pub async fn get<V>(&self, key: &str) -> Result<Option<V>, CacheError>

61 where

62 V: DeserializeOwned,

63 {

64 self.get_bytes(key)

65 .await?

66 .map(|bytes| serde_json::from_slice(&bytes).map_err(CacheError::from))

67 .transpose()

68 }

69

70 pub async fn set<V>(

71 &self,

72 key: &str,

73 value: &V,

74 ttl: Option<Duration>,

75 ) -> Result<(), CacheError>

76 where

77 V: Serialize,

78 {

79 self.set_bytes(key, serde_json::to_vec(value)?, ttl).await

80 }

81

82 pub async fn delete(&self, key: &str) -> Result<(), CacheError> {

83 Ok(redis::delete(&self.redis, key).await?)

84 }

85

86 pub async fn exists(&self, key: &str) -> Result<bool, CacheError> {

87 Ok(redis::exists(&self.redis, key).await?)

88 }

89

90 pub async fn get_bytes(&self, key: &str) -> Result<Option<Vec<u8>>, CacheError> {

91 let mut conn = self.redis.connection_clone();

92 let mut cmd = ::redis::cmd("GET");

93 cmd.arg(key);

94 let value = match &mut conn {

95 RedisConnection::Standalone(conn) => cmd.query_async(conn).await?,

96 RedisConnection::Cluster(conn) => cmd.query_async(conn).await?,

97 };

98 Ok(value)

99 }

100

101 pub async fn set_bytes(

102 &self,

103 key: &str,

104 value: Vec<u8>,

105 ttl: Option<Duration>,

106 ) -> Result<(), CacheError> {

107 if ttl.is_some_and(|ttl| ttl.is_zero()) {

108 return self.delete(key).await;

109 }

110

111 let mut cmd = ::redis::cmd("SET");

112 cmd.arg(key).arg(value);

113 if let Some(ttl) = ttl {

114 let millis = ttl.as_millis().max(1).min(u64::MAX as u128) as u64;

115 cmd.arg("PX").arg(millis);

116 }

117

118 let mut conn = self.redis.connection_clone();

119 match &mut conn {

120 RedisConnection::Standalone(conn) => cmd.exec_async(conn).await?,

121 RedisConnection::Cluster(conn) => cmd.exec_async(conn).await?,

122 }

123 Ok(())

124 }

125}

Modified lib/export/src/database.rs +1 −115 View lib/export/src/database.rs

@@ -102,7 +102,7 @@

102102 }

103103

104104 pub fn route_for(&self, sql: &str) -> &DatabaseConnection {

105 if is_write_sql(sql) {

105 if db::requires_primary(sql) {

106106 self.write()

107107 } else {

108108 self.read()

@@ -116,120 +116,6 @@

116116 let idx = self.round_robin.fetch_add(1, Ordering::Relaxed) % self.replicas.len();

117117 Some(&self.replicas[idx])

118118 }

119}

120

121fn strip_sql_lead(sql: &str) -> &str {

122 let mut s = sql.trim_start();

123 loop {

124 if let Some(rest) = s.strip_prefix("--") {

125 let end = rest.find('\n').unwrap_or(rest.len());

126 s = rest[end..].trim_start();

127 continue;

128 }

129 if let Some(rest) = s.strip_prefix("/*") {

130 let end = rest.find("*/").map(|i| i + 2).unwrap_or(rest.len());

131 s = rest[end..].trim_start();

132 continue;

133 }

134 break;

135 }

136 s

137}

138

139fn first_keyword(sql: &str) -> Option<String> {

140 let lead = strip_sql_lead(sql);

141 let head: String = lead

142 .chars()

143 .take_while(|c| c.is_ascii_alphanumeric() || *c == '_')

144 .collect();

145 if head.is_empty() {

146 None

147 } else {

148 Some(head.to_uppercase())

149 }

150}

151

152fn is_write_sql(sql: &str) -> bool {

153 let Some(kw) = first_keyword(sql) else {

154 return true;

155 };

156 if kw == "WITH" {

157 return with_clause_is_write(sql);

158 }

159 matches!(

160 kw.as_str(),

161 "INSERT"

162 | "UPDATE"

163 | "DELETE"

164 | "MERGE"

165 | "UPSERT"

166 | "REPLACE"

167 | "TRUNCATE"

168 | "CREATE"

169 | "DROP"

170 | "ALTER"

171 | "RENAME"

172 | "GRANT"

173 | "REVOKE"

174 | "COMMIT"

175 | "ROLLBACK"

176 | "SAVEPOINT"

177 | "LOCK"

178 | "UNLOCK"

179 | "COPY"

180 | "CALL"

181 | "SET"

182 | "RESET"

183 | "DO"

184 )

185}

186

187fn with_clause_is_write(sql: &str) -> bool {

188 let upper = sql.to_uppercase();

189 contains_top_level(&upper, "INSERT")

190 || contains_top_level(&upper, "UPDATE")

191 || contains_top_level(&upper, "DELETE")

192 || contains_top_level(&upper, "MERGE")

193}

194

195fn contains_top_level(haystack: &str, needle: &str) -> bool {

196 let bytes = haystack.as_bytes();

197 let n = needle.as_bytes();

198 let mut depth: i32 = 0;

199 let mut in_string = false;

200 let mut i = 0;

201 while i < bytes.len() {

202 let c = bytes[i] as char;

203 if in_string {

204 if c == '\'' {

205 if i + 1 < bytes.len() && bytes[i + 1] == b'\'' {

206 i += 2;

207 continue;

208 }

209 in_string = false;

210 }

211 i += 1;

212 continue;

213 }

214 match c {

215 '\'' => in_string = true,

216 '(' => depth += 1,

217 ')' => depth -= 1,

218 _ if depth == 0 && i + n.len() <= bytes.len() && &bytes[i..i + n.len()] == n => {

219 let before_ok = i == 0

220 || !(bytes[i - 1] as char).is_ascii_alphanumeric() && bytes[i - 1] != b'_';

221 let after_ok = i + n.len() == bytes.len()

222 || !(bytes[i + n.len()] as char).is_ascii_alphanumeric()

223 && bytes[i + n.len()] != b'_';

224 if before_ok && after_ok {

225 return true;

226 }

227 }

228 _ => {}

229 }

230 i += 1;

231 }

232 false

233119}

234120

235121#[async_trait]

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

@@ -1,6 +1,5 @@

11mod api_error;

22pub mod auth_api;

3pub mod cache;

43pub mod database;

54pub mod error;

65pub mod events;

@@ -17,7 +16,6 @@

1716pub mod rest;

1817pub mod serde_datetime;

1918pub mod serde_patch;

20pub mod storage;

2119pub mod telemetry;

2220pub mod urls;

2321pub mod users;

@@ -30,7 +28,7 @@

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

3129pub use error::ExportError;

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

33pub use storage::AppStorage;

31pub use storage::S3Client;

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

3533

3634#[derive(Clone)]

@@ -38,7 +36,7 @@

3836 pub database: Arc<AppDatabase>,

3937 pub redis: AppRedis,

4038 pub cache: ::cache::Cache,

41 pub storage: Arc<AppStorage>,

39 pub storage: Arc<S3Client>,

4240 pub config: AppConfig,

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

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

@@ -71,7 +69,7 @@

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

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

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

74 let storage = Arc::new(AppStorage::connect(config.s3.clone())?);

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

7573

7674 let kenauth = identity;

7775 if let Some(client) = &kenauth {

Deleted lib/export/src/storage.rs +0 −24

@@ -1,24 +0,0 @@

1use config::s3::S3Config;

2use storage::S3Client;

3

4use crate::error::ExportError;

5

6pub struct AppStorage {

7 config: S3Config,

8 client: S3Client,

9}

10

11impl AppStorage {

12 pub fn connect(config: S3Config) -> Result<Self, ExportError> {

13 let client = S3Client::new(&config).map_err(ExportError::from)?;

14 Ok(Self { config, client })

15 }

16

17 pub fn config(&self) -> &S3Config {

18 &self.config

19 }

20

21 pub fn client(&self) -> &S3Client {

22 &self.client

23 }

24}

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

@@ -18,6 +18,7 @@

1818session = { workspace = true }

1919envelope = { workspace = true }

2020imaging = { workspace = true }

21db = { workspace = true }

2122actix-web = { workspace = true }

2223http = "1.5.0"

2324serde = { workspace = true, features = ["derive"] }

Modified lib/git/src/context/cache.rs +8 −3 View lib/git/src/context/cache.rs

@@ -10,16 +10,21 @@

1010 where

1111 V: serde::Serialize,

1212 {

13 Ok(self.cache.set(key, value, ttl).await?)

13 match ttl {

14 Some(ttl) => self.cache.set_json_ex(key, value, ttl).await?,

15 None => self.cache.set_json(key, value).await?,

16 }

17 Ok(())

1418 }

1519 pub async fn cache_get<V>(&self, key: &str) -> Result<Option<V>, crate::share::errors::GitError>

1620 where

1721 V: serde::de::DeserializeOwned,

1822 {

19 Ok(self.cache.get(key).await?)

23 Ok(self.cache.get_json(key).await?)

2024 }

2125 pub async fn cache_remove(&self, key: &str) -> Result<(), crate::share::errors::GitError> {

22 Ok(self.cache.delete(key).await?)

26 let _ = self.cache.delete(key).await?;

27 Ok(())

2328 }

2429 pub async fn cache_exists(&self, key: &str) -> Result<bool, crate::share::errors::GitError> {

2530 Ok(self.cache.exists(key).await?)

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

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

11use std::path::PathBuf;

22use std::sync::Arc;

33

4use ::cache::Cache;

45use export::AppDatabase;

5use export::cache::CacheClient;

66use models::repo::repos;

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

88use uuid::Uuid;

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

1010#[derive(Clone)]

1111pub struct GitContext {

1212 pub model: repos::Model,

13 pub cache: Arc<CacheClient>,

13 pub cache: Cache,

1414 pub database: Arc<AppDatabase>,

1515 pub space_name: String,

1616 pub repo_root: PathBuf,

Modified lib/git/src/service/cache.rs +14 −21 View lib/git/src/service/cache.rs

@@ -1,6 +1,6 @@

11use std::time::Duration;

22

3use export::cache::CacheClient;

3use ::cache::Cache;

44use serde::Serialize;

55use serde::de::DeserializeOwned;

66

@@ -23,7 +23,7 @@

2323}

2424

2525pub async fn cached_fetch<V, F, Fut>(

26 cache: &CacheClient,

26 cache: &Cache,

2727 key: &str,

2828 ttl: Duration,

2929 fetch: F,

@@ -34,11 +34,11 @@

3434 Fut: std::future::Future<Output = Result<V, GitError>>,

3535{

3636 let sanitized = sanitize_key(key);

37 if let Ok(Some(cached)) = cache.get::<V>(&sanitized).await {

37 if let Ok(Some(cached)) = cache.get_json::<V>(&sanitized).await {

3838 return Ok(cached);

3939 }

4040 let val = fetch().await?;

41 let _ = cache.set(&sanitized, &val, Some(ttl)).await;

41 let _ = cache.set_json_ex(&sanitized, &val, ttl).await;

4242 Ok(val)

4343}

4444

@@ -50,32 +50,23 @@

5050 sanitize_key(&format!("{module}.{func}.{key_part}:{tip}"))

5151}

5252

53pub async fn cache_oid_get<V>(

54 cache: &CacheClient,

55 kind: &str,

56 oid: &str,

57) -> Result<Option<V>, GitError>

53pub async fn cache_oid_get<V>(cache: &Cache, kind: &str, oid: &str) -> Result<Option<V>, GitError>

5854where

5955 V: DeserializeOwned,

6056{

61 Ok(cache.get(&oid_key(kind, oid)).await?)

57 Ok(cache.get_json(&oid_key(kind, oid)).await?)

6258}

6359

64pub async fn cache_oid_put<V>(

65 cache: &CacheClient,

66 kind: &str,

67 oid: &str,

68 v: &V,

69) -> Result<(), GitError>

60pub async fn cache_oid_put<V>(cache: &Cache, kind: &str, oid: &str, v: &V) -> Result<(), GitError>

7061where

7162 V: Serialize,

7263{

73 cache.set(&oid_key(kind, oid), v, None).await?;

64 cache.set_json(&oid_key(kind, oid), v).await?;

7465 Ok(())

7566}

7667

7768pub async fn cache_rev_get<V>(

78 cache: &CacheClient,

69 cache: &Cache,

7970 module: &str,

8071 func: &str,

8172 key_part: &str,

@@ -84,11 +75,13 @@

8475where

8576 V: DeserializeOwned,

8677{

87 Ok(cache.get(&rev_key(module, func, key_part, tip)).await?)

78 Ok(cache

79 .get_json(&rev_key(module, func, key_part, tip))

80 .await?)

8881}

8982

9083pub async fn cache_rev_put<V>(

91 cache: &CacheClient,

84 cache: &Cache,

9285 module: &str,

9386 func: &str,

9487 key_part: &str,

@@ -99,7 +92,7 @@

9992 V: Serialize,

10093{

10194 cache

102 .set(&rev_key(module, func, key_part, tip), v, Some(REV_TTL))

95 .set_json_ex(&rev_key(module, func, key_part, tip), v, REV_TTL)

10396 .await?;

10497 Ok(())

10598}

Modified lib/git/src/service/context.rs +1 −3 View lib/git/src/service/context.rs

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

11use std::path::PathBuf;

2use std::sync::Arc;

32

4use export::cache::CacheClient;

53use models::repo::repos;

64use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};

75

@@ -47,7 +45,7 @@

4745 found.ok_or(GitError::NotFound)?

4846 };

4947

50 let cache = Arc::new(CacheClient::new(self.state.redis.clone()));

48 let cache = self.state.cache.clone();

5149 let git = &self.state.config.git;

5250

5351 Ok(GitContext {

Modified lib/git/src/service/repos.rs +1 −1 View lib/git/src/service/repos.rs

@@ -414,7 +414,7 @@

414414 (dir, repo)

415415 };

416416

417 let cache = std::sync::Arc::new(export::cache::CacheClient::new(self.state.redis.clone()));

417 let cache = self.state.cache.clone();

418418 let git_cfg = &self.state.config.git;

419419 let ctx = GitContext {

420420 model: repo.clone(),

Modified lib/git/src/share/errors.rs +1 −1 View lib/git/src/share/errors.rs

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

33#[derive(Debug, thiserror::Error)]

44pub enum GitError {

55 #[error("cache error: {0}")]

6 Cache(#[from] export::cache::CacheError),

6 Cache(#[from] ::cache::CacheError),

77 #[error("invalid argument: {0}")]

88 InvalidArgument(String),

99 #[error("conflict: {0}")]

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

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

88tonic = { workspace = true }

99prost-types = { workspace = true }

1010export = { workspace = true }

11cache = { workspace = true }

1112config = { workspace = true }

1213models = { workspace = true }

1314space = { workspace = true }

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

@@ -158,7 +158,7 @@

158158 let id = Uuid::now_v7();

159159 let filename = sanitize_filename(&cmd.filename);

160160 let key = format!("{ATTACHMENT_PREFIX}/{}/{caller}/{id}/{filename}", space.id);

161 let bucket = self.state.storage.client().bucket_name().to_string();

161 let bucket = self.state.storage.bucket_name().to_string();

162162

163163 issue_attachment::ActiveModel {

164164 id: Set(id),

@@ -191,7 +191,6 @@

191191 let upload_url = self

192192 .state

193193 .storage

194 .client()

195194 .presign_put_with_headers(&key, ttl as u32, put_headers)

196195 .map_err(storage_failure)?;

197196

@@ -222,7 +221,6 @@

222221 let head = self

223222 .state

224223 .storage

225 .client()

226224 .head(&row.s3_key)

227225 .await

228226 .map_err(storage_failure)?

@@ -271,7 +269,7 @@

271269 }

272270

273271 if row.status == IssueAttachmentStatus::Uploaded

274 && let Err(err) = self.state.storage.client().delete(&row.s3_key).await

272 && let Err(err) = self.state.storage.delete(&row.s3_key).await

275273 {

276274 tracing::warn!(error = %err, key = %row.s3_key, "failed to delete issue attachment object");

277275 }

@@ -329,7 +327,6 @@

329327 let stream = self

330328 .state

331329 .storage

332 .client()

333330 .get_stream(&row.s3_key)

334331 .await

335332 .map_err(storage_failure)?;

Modified lib/issues/src/services/issues.rs +1 −1 View lib/issues/src/services/issues.rs

@@ -1226,7 +1226,7 @@

12261226 .await?;

12271227

12281228 for key in attachment_keys {

1229 if let Err(err) = self.state.storage.client().delete(&key).await {

1229 if let Err(err) = self.state.storage.delete(&key).await {

12301230 tracing::warn!(error = %err, key = %key, "failed to delete issue attachment object");

12311231 }

12321232 }

Modified lib/push/src/worker.rs +1 −6 View lib/push/src/worker.rs

@@ -275,12 +275,7 @@

275275 if key.is_empty() {

276276 return None;

277277 }

278 match self

279 .state

280 .storage

281 .client()

282 .presign_get(key, 7 * 24 * 60 * 60)

283 {

278 match self.state.storage.presign_get(key, 7 * 24 * 60 * 60) {

284279 Ok(url) => Some(url),

285280 Err(error) => {

286281 tracing::warn!(%error, %uid, "avatar presign failed");

Modified lib/space/src/services/images.rs +7 −12 View lib/space/src/services/images.rs

@@ -257,7 +257,7 @@

257257 .recorded_space_image_preview(space_id, kind, &key_base)

258258 .await;

259259 let state = export::imaging::ensure_preview(

260 self.storage().client(),

260 self.storage(),

261261 &self.state.redis,

262262 &key_base,

263263 &original_key,

@@ -282,15 +282,14 @@

282282 key_base: &str,

283283 ) -> Option<PreviewState> {

284284 let meta_key = space_image_meta_key(space_id, kind);

285 let fetched: Result<Vec<u8>, storage::StorageError> =

286 self.storage().client().get(&meta_key).await;

285 let fetched: Result<Vec<u8>, storage::StorageError> = self.storage().get(&meta_key).await;

287286 if let Ok(bytes) = fetched

288287 && let Ok(record) = serde_json::from_slice::<serde_json::Value>(&bytes)

289288 && let Some(state) = export::imaging::recorded_preview(&record)

290289 {

291290 return Some(state);

292291 }

293 export::imaging::find_stored_preview(self.storage().client(), key_base).await

292 export::imaging::find_stored_preview(self.storage(), key_base).await

294293 }

295294

296295 /// Persist the sidecar with the derivative that is currently authoritative.

@@ -322,7 +321,6 @@

322321 }

323322 if let Err(error) = self

324323 .storage()

325 .client()

326324 .put_with_content_type(

327325 &space_image_meta_key(space_id, kind),

328326 meta.to_string().as_bytes(),

@@ -342,7 +340,7 @@

342340 };

343341 let key_base = export::imaging::preview_key_base_of(&original_key);

344342 for candidate in imaging::layout::preview_candidate_keys(&key_base) {

345 if let Err(error) = self.storage().client().delete(&candidate).await {

343 if let Err(error) = self.storage().delete(&candidate).await {

346344 tracing::warn!(%error, key = %candidate, "failed to delete old space image preview");

347345 }

348346 }

@@ -365,7 +363,6 @@

365363 let key = space_image_object_key(stored_url).ok_or(SpaceError::NotFound)?;

366364 let url = self

367365 .storage()

368 .client()

369366 .presign_get(&key, SPACE_IMAGE_URL_TTL_SECS as u32)?;

370367 if let Err(error) = export::redis::set_string(

371368 &self.state.redis,

@@ -448,7 +445,6 @@

448445 let now = OffsetDateTime::now_utc();

449446

450447 self.storage()

451 .client()

452448 .put_with_content_type(&object_key, bytes, fmt.content_type())

453449 .await?;

454450

@@ -477,7 +473,7 @@

477473 self.delete_space_image_previews(&stored_url).await;

478474 let key_base = export::imaging::preview_key_base_of(&object_key);

479475 let preview = export::imaging::ensure_preview(

480 self.storage().client(),

476 self.storage(),

481477 &self.state.redis,

482478 &key_base,

483479 &object_key,

@@ -549,13 +545,12 @@

549545 }

550546

551547 if let Some(key) = old_url.as_deref().and_then(space_image_object_key) {

552 if let Err(error) = self.storage().client().delete(&key).await {

548 if let Err(error) = self.storage().delete(&key).await {

553549 tracing::warn!(%error, "failed to delete old space image object");

554550 }

555551 self.delete_space_image_previews(&key).await;

556552 if let Err(error) = self

557553 .storage()

558 .client()

559554 .delete(&space_image_meta_key(space_id, kind))

560555 .await

561556 {

@@ -587,7 +582,7 @@

587582 let format = AvatarFormat::from_extension(ext).unwrap_or(AvatarFormat::Jpg);

588583 let mut bytes = 0u64;

589584 if let Some(key) = space_image_object_key(&url) {

590 match self.storage().client().head(&key).await {

585 match self.storage().head(&key).await {

591586 Ok(Some((size, _))) => bytes = size,

592587 Ok(None) => {}

593588 Err(error) => {

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

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

11use std::sync::Arc;

22

33use config::AppConfig;

4use export::{AppDatabase, AppState, AppStorage};

4use export::{AppDatabase, AppState};

5use storage::S3Client;

56

67use crate::errors::SpaceError;

78

@@ -23,7 +24,7 @@

2324 &self.state.database

2425 }

2526

26 pub fn storage(&self) -> &Arc<AppStorage> {

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

2728 &self.state.storage

2829 }

2930

Modified lib/user/src/services/avatar.rs +7 −10 View lib/user/src/services/avatar.rs

@@ -140,7 +140,6 @@

140140 };

141141 let url = self

142142 .storage()

143 .client()

144143 .presign_get(&key, AVATAR_URL_TTL_SECS as u32)

145144 .map_err(UserError::from)?;

146145 if let Err(error) =

@@ -178,7 +177,7 @@

178177 let key_base = export::imaging::preview_key_base_of(&original_key);

179178 let recorded = self.recorded_avatar_preview(uid, &key_base).await;

180179 let state = export::imaging::ensure_preview(

181 self.storage().client(),

180 self.storage(),

182181 &self.state.redis,

183182 &key_base,

184183 &original_key,

@@ -198,13 +197,13 @@

198197 /// recognised once its derivative has been generated.

199198 async fn recorded_avatar_preview(&self, uid: i64, key_base: &str) -> Option<PreviewState> {

200199 let meta_key = avatar_meta_key(uid);

201 if let Ok(bytes) = self.storage().client().get(&meta_key).await

200 if let Ok(bytes) = self.storage().get(&meta_key).await

202201 && let Ok(record) = serde_json::from_slice::<serde_json::Value>(&bytes)

203202 && let Some(state) = export::imaging::recorded_preview(&record)

204203 {

205204 return Some(state);

206205 }

207 export::imaging::find_stored_preview(self.storage().client(), key_base).await

206 export::imaging::find_stored_preview(self.storage(), key_base).await

208207 }

209208

210209 /// Persist the sidecar next to the original, recording which derivative is

@@ -230,7 +229,6 @@

230229 Ok(meta_bytes) => {

231230 if let Err(error) = self

232231 .storage()

233 .client()

234232 .put_with_content_type(&avatar_meta_key(uid), &meta_bytes, "application/json")

235233 .await

236234 {

@@ -248,7 +246,7 @@

248246 async fn delete_avatar_previews(&self, original_key: &str) {

249247 let key_base = export::imaging::preview_key_base_of(original_key);

250248 for candidate in imaging::layout::preview_candidate_keys(&key_base) {

251 if let Err(error) = self.storage().client().delete(&candidate).await {

249 if let Err(error) = self.storage().delete(&candidate).await {

252250 tracing::warn!(%error, key = %candidate, "failed to delete old avatar preview");

253251 }

254252 }

@@ -299,7 +297,6 @@

299297

300298 let key = format!("{AVATAR_PREFIX}/{uid}.{}", fmt.extension());

301299 self.storage()

302 .client()

303300 .put_with_content_type(&key, bytes, fmt.content_type())

304301 .await

305302 .map_err(UserError::from)?;

@@ -322,7 +319,7 @@

322319 // clearing it first keeps exactly one candidate object alive.

323320 self.delete_avatar_previews(&key).await;

324321 let preview = export::imaging::ensure_preview(

325 self.storage().client(),

322 self.storage(),

326323 &self.state.redis,

327324 &key_base,

328325 &key,

@@ -374,11 +371,11 @@

374371 if let Some(old_key) = avatar_object_key(&old_url)

375372 && old_key.starts_with(&format!("{AVATAR_PREFIX}/"))

376373 {

377 if let Err(error) = self.storage().client().delete(&old_key).await {

374 if let Err(error) = self.storage().delete(&old_key).await {

378375 tracing::warn!(%error, "failed to delete old user avatar object");

379376 }

380377 self.delete_avatar_previews(&old_key).await;

381 if let Err(error) = self.storage().client().delete(&avatar_meta_key(uid)).await {

378 if let Err(error) = self.storage().delete(&avatar_meta_key(uid)).await {

382379 tracing::warn!(%error, "failed to delete old user avatar meta json");

383380 }

384381 }

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

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

11use std::sync::Arc;

22

33use config::AppConfig;

4use export::{AppDatabase, AppState, AppStorage};

4use export::{AppDatabase, AppState};

5use storage::S3Client;

56

67#[derive(Clone)]

78pub struct UserService {

@@ -21,7 +22,7 @@

2122 &self.state.database

2223 }

2324

24 pub fn storage(&self) -> &Arc<AppStorage> {

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

2526 &self.state.storage

2627 }

2728}

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

@@ -2334,6 +2334,7 @@

23342334 "bytes",

23352335 "cache",

23362336 "config 1.0.0",

2337 "db",

23372338 "email",

23382339 "envelope",

23392340 "events",

@@ -2719,6 +2720,7 @@

27192720 "async-trait",

27202721 "base64 0.23.1",

27212722 "bytes",

2723 "cache",

27222724 "config 1.0.0",

27232725 "export",

27242726 "flate2",