refactor: rename kvcache to cache and update related imports

Added lib/cache/src/conn.rs +137 −0 View lib/cache/src/conn.rs

@@ -0,0 +1,137 @@

1use std::sync::Arc;

2use std::sync::atomic::{AtomicUsize, Ordering};

3

4use config::identity::redis::RedisConfig;

5use redis::ConnectionInfo;

6use redis::aio::{ConnectionManager, ConnectionManagerConfig};

7use redis::cluster_async::ClusterConnection;

8

9use crate::error::{CacheError, CacheResult};

10

11#[derive(Clone)]

12pub struct CacheConn {

13 connections: Arc<Vec<CacheConnection>>,

14 cursor: Arc<AtomicUsize>,

15 db: i64,

16 cluster: bool,

17}

18

19#[derive(Clone)]

20pub enum CacheConnection {

21 Standard(ConnectionManager),

22 Cluster(ClusterConnection),

23}

24

25impl CacheConn {

26 pub async fn init(cfg: &RedisConfig) -> CacheResult<Self> {

27 cfg.validate().map_err(CacheError::Config)?;

28

29 let db = cfg.database();

30 let cluster = cfg.is_cluster();

31 let seeds = cfg.seeds().map_err(CacheError::Config)?;

32 let pool_size = cfg.pool.max_size.max(1);

33

34 let connections = if cluster {

35 let nodes = seeds

36 .iter()

37 .map(|(host, port)| connection_info(host, *port, cfg))

38 .collect::<CacheResult<Vec<_>>>()?;

39 let client = redis::cluster::ClusterClient::builder(nodes)

40 .connection_timeout(cfg.pool.create_timeout())

41 .build()

42 .map_err(|error| CacheError::config(error.to_string()))?;

43

44 let mut connections = Vec::with_capacity(pool_size);

45 for _ in 0..pool_size {

46 let connection = client

47 .get_async_connection()

48 .await

49 .map_err(CacheError::from)?;

50 connections.push(CacheConnection::Cluster(connection));

51 }

52 connections

53 } else {

54 let (host, port) = seeds

55 .first()

56 .cloned()

57 .ok_or_else(|| CacheError::config("redis requires at least one node"))?;

58 let info = connection_info(&host, port, cfg)?;

59 let client = redis::Client::open(info).map_err(CacheError::from)?;

60 let manager_config = ConnectionManagerConfig::new()

61 .set_connection_timeout(Some(cfg.pool.create_timeout()));

62

63 let mut connections = Vec::with_capacity(pool_size);

64 for _ in 0..pool_size {

65 let manager =

66 ConnectionManager::new_with_config(client.clone(), manager_config.clone())

67 .await

68 .map_err(CacheError::from)?;

69 connections.push(CacheConnection::Standard(manager));

70 }

71 connections

72 };

73

74 let conn = Self {

75 connections: Arc::new(connections),

76 cursor: Arc::new(AtomicUsize::new(0)),

77 db,

78 cluster,

79 };

80 Ok(conn)

81 }

82

83 pub async fn acquire(&self) -> CacheResult<CacheConnection> {

84 if self.connections.is_empty() {

85 return Err(CacheError::config("redis has no connections"));

86 }

87 let index = self.cursor.fetch_add(1, Ordering::Relaxed) % self.connections.len();

88 Ok(self.connections[index].clone())

89 }

90

91 pub async fn ping(&self) -> CacheResult<()> {

92 let mut conn = self.acquire().await?;

93 redis::cmd("PING")

94 .query_async::<String>(&mut conn)

95 .await

96 .map(|_| ())

97 .map_err(CacheError::from)

98 }

99

100 pub fn database(&self) -> i64 {

101 self.db

102 }

103

104 pub fn is_cluster(&self) -> bool {

105 self.cluster

106 }

107

108 pub fn pool_size(&self) -> usize {

109 self.connections.len()

110 }

111

112 pub fn available(&self) -> usize {

113 self.connections.len()

114 }

115}

116

117fn connection_info(host: &str, port: u16, cfg: &RedisConfig) -> CacheResult<ConnectionInfo> {

118 let host = if host.contains(':') && !host.starts_with('[') {

119 format!("[{host}]")

120 } else {

121 host.to_string()

122 };

123 let url = format!("redis://{host}:{port}");

124 let info = redis::Client::open(url)

125 .map_err(CacheError::from)?

126 .get_connection_info()

127 .clone();

128 let mut settings = info.redis_settings().clone();

129 settings = settings.set_db(cfg.database());

130 if let Some(username) = cfg.username.as_deref().filter(|value| !value.is_empty()) {

131 settings = settings.set_username(username);

132 }

133 if let Some(password) = cfg.password.as_deref().filter(|value| !value.is_empty()) {

134 settings = settings.set_password(password);

135 }

136 Ok(info.set_redis_settings(settings))

137}

Modified lib/cache/src/error.rs +106 −16 View lib/cache/src/error.rs

@@ -1,18 +1,114 @@

1use std::fmt;

1use redis::RedisError;

22

3use redis::RedisError;

3pub type CacheResult<T> = Result<T, CacheError>;

44

55#[derive(Debug)]

66pub enum CacheError {

7 Serialization(String),

7 Config(String),

88 Backend(RedisError),

9 Encode {

10 key: String,

11 message: String,

12 },

13 Decode {

14 key: String,

15 message: String,

16 },

17 Unsupported {

18 feature: &'static str,

19 reason: &'static str,

20 },

21 Lock {

22 key: String,

23 message: String,

24 },

25}

26

27impl CacheError {

28 pub fn config(message: impl Into<String>) -> Self {

29 Self::Config(message.into())

30 }

31

32 pub fn encode(key: impl Into<String>, error: serde_json::Error) -> Self {

33 Self::Encode {

34 key: key.into(),

35 message: error.to_string(),

36 }

37 }

38

39 pub fn decode(key: impl Into<String>, error: serde_json::Error) -> Self {

40 Self::Decode {

41 key: key.into(),

42 message: error.to_string(),

43 }

44 }

45

46 pub fn unsupported(feature: &'static str, reason: &'static str) -> Self {

47 Self::Unsupported { feature, reason }

48 }

49

50 pub fn lock(key: impl Into<String>, message: impl Into<String>) -> Self {

51 Self::Lock {

52 key: key.into(),

53 message: message.into(),

54 }

55 }

56

57 pub fn key(&self) -> Option<&str> {

58 match self {

59 Self::Encode { key, .. } | Self::Decode { key, .. } | Self::Lock { key, .. } => {

60 Some(key)

61 }

62 _ => None,

63 }

64 }

65

66 pub fn is_config(&self) -> bool {

67 matches!(self, Self::Config(_))

68 }

69

70 pub fn is_backend(&self) -> bool {

71 matches!(self, Self::Backend(_))

72 }

73

74 pub fn is_unsupported(&self) -> bool {

75 matches!(self, Self::Unsupported { .. })

76 }

77

78 pub fn is_closed(&self) -> bool {

79 match self {

80 Self::Backend(error) => error.is_connection_dropped(),

81 _ => false,

82 }

83 }

84

85 pub fn is_retryable(&self) -> bool {

86 match self {

87 Self::Backend(error) => retryable_backend(error),

88 _ => false,

89 }

90 }

991}

1092

11impl fmt::Display for CacheError {

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

93fn retryable_backend(error: &RedisError) -> bool {

94 error.is_io_error()

95 || error.is_timeout()

96 || error.is_connection_refusal()

97 || error.is_connection_dropped()

98 || error.is_cluster_error()

99}

100

101impl std::fmt::Display for CacheError {

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

13103 match self {

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

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

104 Self::Config(message) => write!(f, "cache config: {message}"),

105 Self::Backend(error) => write!(f, "cache backend: {error}"),

106 Self::Encode { key, message } => write!(f, "cache encode {key}: {message}"),

107 Self::Decode { key, message } => write!(f, "cache decode {key}: {message}"),

108 Self::Unsupported { feature, reason } => {

109 write!(f, "cache unsupported {feature}: {reason}")

110 }

111 Self::Lock { key, message } => write!(f, "cache lock {key}: {message}"),

16112 }

17113 }

18114}

@@ -20,20 +116,14 @@

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

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

22118 match self {

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

119 Self::Backend(error) => Some(error),

24120 _ => None,

25121 }

26122 }

27123}

28124

29125impl From<RedisError> for CacheError {

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

31 CacheError::Backend(err)

32 }

33}

34

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

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

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

126 fn from(error: RedisError) -> Self {

127 Self::Backend(error)

38128 }

39129}

Renamed lib/kvcache/src/exec.rs → lib/cache/src/exec.rs +3 −13 View lib/cache/src/exec.rs

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

11use std::io;

22

3use deadpool_redis::PoolError;

4use deadpool_redis::redis::aio::ConnectionLike;

5use deadpool_redis::redis::{Cmd, Pipeline, RedisError, RedisFuture, Value};

3use redis::aio::ConnectionLike;

4use redis::{Cmd, Pipeline, RedisError, RedisFuture, Value};

65

76use crate::conn::{CacheConn, CacheConnection};

87use crate::error::CacheError;

@@ -64,16 +63,7 @@

6463

6564fn connect_error(error: CacheError) -> RedisError {

6665 match error {

67 CacheError::Pool(error) => pool_error(error),

68 error => io::Error::other(error).into(),

69 }

70}

71

72pub(crate) fn pool_error(error: PoolError) -> RedisError {

73 match error {

74 PoolError::Backend(error) => error,

75 error @ PoolError::Timeout(_) => io::Error::new(io::ErrorKind::TimedOut, error).into(),

76 error @ PoolError::Closed => io::Error::new(io::ErrorKind::NotConnected, error).into(),

66 CacheError::Backend(error) => error,

7767 error => io::Error::other(error).into(),

7868 }

7969}

Renamed lib/kvcache/src/hash.rs → lib/cache/src/hash.rs +1 −1 View lib/cache/src/hash.rs

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

11use std::collections::HashMap;

22

3use deadpool_redis::redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

3use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

44use serde::Serialize;

55use serde::de::DeserializeOwned;

66

Renamed lib/kvcache/src/json.rs → lib/cache/src/json.rs +0 −0 View lib/cache/src/json.rs

No content changes.

Renamed lib/kvcache/src/keys.rs → lib/cache/src/keys.rs +1 −1 View lib/cache/src/keys.rs

@@ -1,4 +1,4 @@

1use deadpool_redis::redis::{AsyncCommands, AsyncIter};

1use redis::{AsyncCommands, AsyncIter};

22

33use crate::error::{CacheError, CacheResult};

44use crate::kv::Cache;

Renamed lib/kvcache/src/kv.rs → lib/cache/src/kv.rs +1 −14 View lib/cache/src/kv.rs

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

22

33use config::identity::loader::Section;

44use config::identity::redis::RedisConfig;

5use deadpool_redis::Status;

6use deadpool_redis::redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

5use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

76

87use crate::conn::{CacheConn, CacheConnection};

98use crate::error::{CacheError, CacheResult};

@@ -66,18 +65,6 @@

6665

6766 pub fn database(&self) -> i64 {

6867 self.conn.database()

69 }

70

71 pub fn status(&self) -> Status {

72 self.conn.status()

73 }

74

75 pub fn close(&self) {

76 self.conn.close();

77 }

78

79 pub fn is_closed(&self) -> bool {

80 self.conn.is_closed()

8168 }

8269

8370 pub async fn acquire(&self) -> CacheResult<CacheConnection> {

Modified lib/cache/src/lib.rs +22 −3 View lib/cache/src/lib.rs

@@ -1,5 +1,24 @@

1pub mod client;

1pub mod conn;

22pub mod error;

3pub mod exec;

4pub mod hash;

5pub mod json;

6pub mod keys;

7pub mod list;

8pub mod lock;

9pub mod pipeline;

10pub mod scripts;

11pub mod set;

12pub mod zset;

313

4pub use client::CacheClient;

5pub use error::CacheError;

14mod kv;

15

16pub use conn::{CacheConn, CacheConnection};

17pub use error::{CacheError, CacheResult};

18pub use keys::{CacheKey, SEPARATOR, join, join_tagged, same_slot, slot_tag};

19pub use kv::Cache;

20pub use lock::{CacheLock, CacheLocks, RETRY_INTERVAL};

21pub use pipeline::CachePipeline;

22pub use scripts::{LuaScripts, RateLimit, unix_millis};

23

24pub use redis;

Renamed lib/kvcache/src/list.rs → lib/cache/src/list.rs +1 −1 View lib/cache/src/list.rs

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

11use std::num::NonZeroUsize;

22

3use deadpool_redis::redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

3use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

44

55use crate::error::{CacheError, CacheResult};

66use crate::kv::Cache;

Renamed lib/kvcache/src/lock.rs → lib/cache/src/lock.rs +0 −0 View lib/cache/src/lock.rs

No content changes.

Renamed lib/kvcache/src/pipeline.rs → lib/cache/src/pipeline.rs +1 −1 View lib/cache/src/pipeline.rs

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

11use std::time::Duration;

22

3use deadpool_redis::redis::{Cmd, FromRedisValue, Pipeline, ToRedisArgs, Value, cmd};

3use redis::{Cmd, FromRedisValue, Pipeline, ToRedisArgs, Value, cmd};

44use serde::Serialize;

55

66use crate::error::{CacheError, CacheResult};

Renamed lib/kvcache/src/scripts.rs → lib/cache/src/scripts.rs +1 −1 View lib/cache/src/scripts.rs

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

11use std::sync::LazyLock;

22use std::time::{Duration, SystemTime, UNIX_EPOCH};

33

4use deadpool_redis::redis::Script;

4use redis::Script;

55

66use crate::error::{CacheError, CacheResult};

77use crate::keys;

Renamed lib/kvcache/src/set.rs → lib/cache/src/set.rs +1 −1 View lib/cache/src/set.rs

@@ -1,4 +1,4 @@

1use deadpool_redis::redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

1use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

22

33use crate::error::{CacheError, CacheResult};

44use crate::kv::Cache;

Renamed lib/kvcache/src/zset.rs → lib/cache/src/zset.rs +1 −1 View lib/cache/src/zset.rs

@@ -1,4 +1,4 @@

1use deadpool_redis::redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

1use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};

22

33use crate::error::{CacheError, CacheResult};

44use crate::kv::Cache;

Modified lib/cache/Cargo.toml +6 −2 View lib/cache/Cargo.toml

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

44edition.workspace = true

55

66[dependencies]

7export = { workspace = true }

7config = { workspace = true }

88redis = { workspace = true }

9serde = { workspace = true }

9futures-util = "0.3.34"

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

1011serde_json = { workspace = true }

12tokio = { workspace = true, features = ["time", "sync"] }

13tracing = { workspace = true }

14uuid = { workspace = true }

No newline at end of file

Renamed lib/cache/src/client.rs → lib/export/src/cache.rs +43 −6 View lib/export/src/cache.rs

@@ -1,11 +1,48 @@

1use std::fmt;

12use std::time::Duration;

23

3use export::{AppRedis, RedisConnection};

4use ::redis::RedisError;

45use serde::Serialize;

56use serde::de::DeserializeOwned;

67

7use crate::error::CacheError;

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

89

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

946#[derive(Clone)]

1047pub struct CacheClient {

1148 redis: AppRedis,

@@ -43,16 +80,16 @@

4380 }

4481

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

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

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

4784 }

4885

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

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

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

5188 }

5289

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

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

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

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

5693 cmd.arg(key);

5794 let value = match &mut conn {

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

@@ -71,7 +108,7 @@

71108 return self.delete(key).await;

72109 }

73110

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

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

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

76113 if let Some(ttl) = ttl {

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

Modified lib/export/src/lib.rs +1 −0 View lib/export/src/lib.rs

@@ -1,4 +1,5 @@

11mod api_error;

2pub mod cache;

23pub mod database;

34pub mod error;

45pub mod events;

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

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

11use std::path::PathBuf;

22use std::sync::Arc;

33

4use cache::CacheClient;

54use export::AppDatabase;

5use export::cache::CacheClient;

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

77use models::repo::repos;

88use uuid::Uuid;

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

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

11use std::time::Duration;

22

3use cache::CacheClient;

3use export::cache::CacheClient;

44use serde::Serialize;

55use serde::de::DeserializeOwned;

66

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

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

11use std::path::PathBuf;

22use std::sync::Arc;

33

4use cache::CacheClient;

4use export::cache::CacheClient;

55use models::repo::repos;

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

77

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(cache::CacheClient::new(self.state.redis.clone()));

417 let cache = std::sync::Arc::new(export::cache::CacheClient::new(self.state.redis.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] cache::CacheError),

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

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

88 InvalidArgument(String),

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

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

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

1111config = { workspace = true }

1212models = { workspace = true }

1313space = { workspace = true }

14cache = { workspace = true }

1514serde = { workspace = true }

1615serde_json = { workspace = true }

1716thiserror = { workspace = true }

Modified lib/identity/src/auth/grpc/agreement.rs +2 −2 View lib/identity/src/auth/grpc/agreement.rs

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

2828}

2929

3030pub async fn acceptance_of(

31 cache: &crate::kvcache::Cache,

31 cache: &crate::cache::Cache,

3232 user_id: i64,

3333 key: &str,

3434) -> AuthResult<Option<Acceptance>> {

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

3636}

3737

3838pub async fn record_acceptance(

39 cache: &crate::kvcache::Cache,

39 cache: &crate::cache::Cache,

4040 user_id: i64,

4141 key: &str,

4242 version: &str,

Modified lib/identity/src/auth/grpc/store.rs +5 −8 View lib/identity/src/auth/grpc/store.rs

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

248248}

249249

250250pub async fn load_session(

251 cache: &crate::kvcache::Cache,

251 cache: &crate::cache::Cache,

252252 session_key: &str,

253253) -> AuthResult<Option<crate::session::SessionState>> {

254254 use crate::session::SessionStore;

@@ -270,7 +270,7 @@

270270}

271271

272272pub async fn index_session_key(

273 cache: &crate::kvcache::Cache,

273 cache: &crate::cache::Cache,

274274 user_id: i64,

275275 session_key: &str,

276276 ttl: std::time::Duration,

@@ -281,14 +281,11 @@

281281 Ok(())

282282}

283283

284pub async fn session_index(cache: &crate::kvcache::Cache, user_id: i64) -> AuthResult<Vec<String>> {

284pub async fn session_index(cache: &crate::cache::Cache, user_id: i64) -> AuthResult<Vec<String>> {

285285 Ok(cache.smembers(&session_index_key(user_id)).await?)

286286}

287287

288pub async fn drop_session_state(

289 cache: &crate::kvcache::Cache,

290 session_key: &str,

291) -> AuthResult<()> {

288pub async fn drop_session_state(cache: &crate::cache::Cache, session_key: &str) -> AuthResult<()> {

292289 use crate::session::{SessionKey, SessionStore};

293290

294291 let key =

@@ -300,7 +297,7 @@

300297}

301298

302299pub async fn drop_user_session_states(

303 cache: &crate::kvcache::Cache,

300 cache: &crate::cache::Cache,

304301 user_id: i64,

305302 keep: Option<Uuid>,

306303) -> AuthResult<u32> {

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

@@ -36,12 +36,12 @@

3636 )

3737}

3838

39pub async fn invalidate_me_cache(cache: &crate::kvcache::Cache, user_id: i64) {

39pub async fn invalidate_me_cache(cache: &crate::cache::Cache, user_id: i64) {

4040 let (cache_key, count_key) = me_cache_keys(user_id);

4141 let dropped = async {

4242 cache.delete(&cache_key).await?;

4343 cache.delete(&count_key).await?;

44 Ok::<(), crate::kvcache::CacheError>(())

44 Ok::<(), crate::cache::CacheError>(())

4545 }

4646 .await;

4747 if let Err(error) = dropped {

Modified lib/identity/src/auth/pow/store.rs +1 −1 View lib/identity/src/auth/pow/store.rs

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

11use std::time::Duration;

22

3use crate::cache::Cache;

34use crate::config::pow::PowConfig;

4use crate::kvcache::Cache;

55use crate::session::Session;

66

77use crate::auth::challenge::types::{ChallengeResponse, PowProofRequest, SESSION_CHALLENGE};

Modified lib/identity/src/auth/share/codes.rs +1 −1 View lib/identity/src/auth/share/codes.rs

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

11use std::time::Duration;

22

3use crate::kvcache::Cache;

3use crate::cache::Cache;

44

55use crate::auth::errors::{AuthError, AuthResult};

66use crate::auth::share::crypto;

Modified lib/identity/src/auth/share/rate.rs +1 −1 View lib/identity/src/auth/share/rate.rs

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

11use std::time::Duration;

22

3use crate::kvcache::Cache;

3use crate::cache::Cache;

44

55use crate::auth::errors::{AuthError, AuthResult};

66

Modified lib/identity/src/auth/auth_svc.rs +1 −1 View lib/identity/src/auth/auth_svc.rs

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

11use std::sync::Arc;

22

3use crate::cache::Cache;

34use crate::config::app::AppRegistry;

45use crate::config::auth::AuthConfig;

56use crate::config::jwt::JwtConfig;

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

78use crate::db::DbConn;

89use crate::email::EmailService;

910use crate::events::{EventHandler, Labels, RuntimeEvent};

10use crate::kvcache::Cache;

1111use serde::Serialize;

1212

1313pub use super::errors::{AuthError, AuthResult};

Modified lib/identity/src/auth/errors.rs +1 −1 View lib/identity/src/auth/errors.rs

@@ -154,7 +154,7 @@

154154 #[error("database error: {0}")]

155155 Database(#[from] crate::db::DbError),

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

157 Cache(#[from] crate::kvcache::CacheError),

157 Cache(#[from] crate::cache::CacheError),

158158 #[error("session error: {0}")]

159159 Session(#[from] crate::session::SessionError),

160160 #[error("email error: {0}")]

Modified lib/identity/src/oauth/mod.rs +1 −1 View lib/identity/src/oauth/mod.rs

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

11mod assets;

22

33use crate::auth::AuthSvc;

4use crate::cache::Cache;

45use crate::config::Config;

56use crate::config::app::AppRegistry;

67use crate::config::auth::AuthConfig;

@@ -14,7 +15,6 @@

1415use crate::db::DbConn;

1516use crate::email::EmailServiceBuilder;

1617use crate::events::EventHandler;

17use crate::kvcache::Cache;

1818use crate::queue::RocketMqConn;

1919use crate::session::{RedisStore, SessionMiddleware};

2020use actix_web::{App, HttpServer, web};

Modified lib/identity/src/lib.rs +2 −2 View lib/identity/src/lib.rs

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

11//! KenOAuth 认证中心。

22//!

33//! 业务逻辑保留在 `auth` / `oauth`；账号、会话、令牌之外的基础设施

4//! （config、db、email、envelope、events、jwt、kvcache、models、oltp、

4//! （config、db、email、envelope、events、jwt、cache、models、oltp、

55//! protos、queue、session、shutdown、crypt）已拆成工作区公共 crate，

66//! 这里按原模块名重新导出，保持既有调用路径不变。

77

@@ -11,13 +11,13 @@

1111pub mod client;

1212pub mod oauth;

1313

14pub use cache;

1415pub use config::identity as config;

1516pub use crypt;

1617pub use db;

1718pub use email;

1819pub use envelope;

1920pub use events;

20pub use kvcache;

2121pub use models;

2222pub use oltp::{self, shutdown};

2323pub use proto as protos;

Modified lib/identity/Cargo.toml +1 −7 View lib/identity/Cargo.toml

@@ -20,7 +20,7 @@

2020email = { workspace = true }

2121envelope = { workspace = true }

2222events = { workspace = true }

23kvcache = { workspace = true }

23cache = { workspace = true }

2424oltp = { workspace = true }

2525proto = { workspace = true }

2626queue = { workspace = true }

@@ -35,12 +35,6 @@

3535base64 = "0.23.1"

3636bytes = "1.12.1"

3737chrono = { version = "0.4.45", features = ["serde"] }

38deadpool-redis = { version = "0.23.1", features = [

39 "cluster",

40 "cluster-async",

41 "script",

42 "tokio-comp",

43] }

4438ed25519-dalek = "2.2.0"

4539figment = { version = "0.10", default-features = false, features = [

4640 "toml",

Deleted lib/kvcache/src/conn.rs +0 −158

@@ -1,158 +0,0 @@

1use crate::error::{CacheError, CacheResult};

2use config::identity::redis::RedisConfig;

3use deadpool_redis::{

4 ConnectionAddr, ConnectionInfo, PoolConfig, RedisConnectionInfo, Runtime, Status, Timeouts,

5};

6

7#[derive(Clone)]

8pub struct CacheConn {

9 pool: CachePool,

10 db: i64,

11}

12

13#[derive(Clone)]

14enum CachePool {

15 Standard(deadpool_redis::Pool),

16 Cluster(deadpool_redis::cluster::Pool),

17}

18

19pub enum CacheConnection {

20 Standard(deadpool_redis::Connection),

21 Cluster(deadpool_redis::cluster::Connection),

22}

23

24impl CacheConn {

25 pub async fn init(cfg: &RedisConfig) -> CacheResult<Self> {

26 let pool = &cfg.pool;

27 let pool_config = PoolConfig {

28 max_size: pool.max_size,

29 timeouts: Timeouts {

30 wait: Some(pool.wait_timeout()),

31 create: Some(pool.create_timeout()),

32 recycle: Some(pool.recycle_timeout()),

33 },

34 ..PoolConfig::default()

35 };

36 Self::init_with_pool_config(cfg, pool_config).await

37 }

38

39 pub async fn init_with_pool_config(

40 cfg: &RedisConfig,

41 pool_config: PoolConfig,

42 ) -> CacheResult<Self> {

43 cfg.validate().map_err(CacheError::Config)?;

44

45 let db = cfg.database();

46 let cluster = cfg.is_cluster();

47 let seeds = cfg.seeds().map_err(CacheError::Config)?;

48 let connections: Vec<ConnectionInfo> = seeds

49 .into_iter()

50 .map(|(host, port)| ConnectionInfo {

51 addr: ConnectionAddr::Tcp(host, port),

52 redis: RedisConnectionInfo {

53 db,

54 username: cfg.username.clone(),

55 password: cfg.password.clone(),

56 ..RedisConnectionInfo::default()

57 },

58 })

59 .collect();

60

61 let pool = if cluster {

62 let manager = deadpool_redis::cluster::Manager::new(connections, false)?;

63 CachePool::Cluster(

64 deadpool_redis::cluster::Pool::builder(manager)

65 .config(pool_config)

66 .runtime(Runtime::Tokio1)

67 .build()

68 .map_err(|error| CacheError::config(error.to_string()))?,

69 )

70 } else {

71 let connection = single_connection(connections)?;

72 let manager = deadpool_redis::Manager::new(connection)?;

73 CachePool::Standard(

74 deadpool_redis::Pool::builder(manager)

75 .config(pool_config)

76 .runtime(Runtime::Tokio1)

77 .build()

78 .map_err(|error| CacheError::config(error.to_string()))?,

79 )

80 };

81

82 let conn = Self { pool, db };

83 if let Err(error) = conn.acquire().await {

84 conn.close();

85 return Err(error);

86 }

87 Ok(conn)

88 }

89

90 pub async fn acquire(&self) -> CacheResult<CacheConnection> {

91 match &self.pool {

92 CachePool::Standard(pool) => pool.get().await.map(CacheConnection::Standard),

93 CachePool::Cluster(pool) => pool.get().await.map(CacheConnection::Cluster),

94 }

95 .map_err(CacheError::from)

96 }

97

98 pub fn database(&self) -> i64 {

99 self.db

100 }

101

102 pub fn is_cluster(&self) -> bool {

103 matches!(self.pool, CachePool::Cluster(_))

104 }

105

106 pub fn status(&self) -> Status {

107 match &self.pool {

108 CachePool::Standard(pool) => pool.status(),

109 CachePool::Cluster(pool) => pool.status(),

110 }

111 }

112

113 pub fn close(&self) {

114 match &self.pool {

115 CachePool::Standard(pool) => pool.close(),

116 CachePool::Cluster(pool) => pool.close(),

117 }

118 }

119

120 pub fn is_closed(&self) -> bool {

121 match &self.pool {

122 CachePool::Standard(pool) => pool.is_closed(),

123 CachePool::Cluster(pool) => pool.is_closed(),

124 }

125 }

126

127 pub fn pool_size(&self) -> usize {

128 match &self.pool {

129 CachePool::Standard(pool) => pool.status().size,

130 CachePool::Cluster(pool) => pool.status().size,

131 }

132 }

133

134 pub fn available(&self) -> usize {

135 match &self.pool {

136 CachePool::Standard(pool) => pool.status().available,

137 CachePool::Cluster(pool) => pool.status().available,

138 }

139 }

140}

141

142impl CacheConnection {

143 pub fn invalidate(self) {

144 match self {

145 Self::Standard(connection) => drop(deadpool_redis::Connection::take(connection)),

146 Self::Cluster(connection) => {

147 drop(deadpool_redis::cluster::Connection::take(connection));

148 }

149 }

150 }

151}

152

153fn single_connection(connections: Vec<ConnectionInfo>) -> CacheResult<ConnectionInfo> {

154 connections

155 .into_iter()

156 .next()

157 .ok_or_else(|| CacheError::config("redis requires at least one node"))

158}

Deleted lib/kvcache/src/error.rs +0 −145

@@ -1,145 +0,0 @@

1use deadpool_redis::PoolError;

2use deadpool_redis::redis::RedisError;

3

4pub type CacheResult<T> = Result<T, CacheError>;

5

6#[derive(Debug)]

7pub enum CacheError {

8 Config(String),

9 Pool(PoolError),

10 Backend(RedisError),

11 Encode {

12 key: String,

13 message: String,

14 },

15 Decode {

16 key: String,

17 message: String,

18 },

19 Unsupported {

20 feature: &'static str,

21 reason: &'static str,

22 },

23 Lock {

24 key: String,

25 message: String,

26 },

27}

28

29impl CacheError {

30 pub fn config(message: impl Into<String>) -> Self {

31 Self::Config(message.into())

32 }

33

34 pub fn encode(key: impl Into<String>, error: serde_json::Error) -> Self {

35 Self::Encode {

36 key: key.into(),

37 message: error.to_string(),

38 }

39 }

40

41 pub fn decode(key: impl Into<String>, error: serde_json::Error) -> Self {

42 Self::Decode {

43 key: key.into(),

44 message: error.to_string(),

45 }

46 }

47

48 pub fn unsupported(feature: &'static str, reason: &'static str) -> Self {

49 Self::Unsupported { feature, reason }

50 }

51

52 pub fn lock(key: impl Into<String>, message: impl Into<String>) -> Self {

53 Self::Lock {

54 key: key.into(),

55 message: message.into(),

56 }

57 }

58

59 pub fn key(&self) -> Option<&str> {

60 match self {

61 Self::Encode { key, .. } | Self::Decode { key, .. } | Self::Lock { key, .. } => {

62 Some(key)

63 }

64 _ => None,

65 }

66 }

67

68 pub fn is_config(&self) -> bool {

69 matches!(self, Self::Config(_))

70 }

71

72 pub fn is_backend(&self) -> bool {

73 matches!(self, Self::Backend(_))

74 }

75

76 pub fn is_unsupported(&self) -> bool {

77 matches!(self, Self::Unsupported { .. })

78 }

79

80 pub fn is_closed(&self) -> bool {

81 match self {

82 Self::Pool(PoolError::Closed) => true,

83 Self::Backend(error) => error.is_connection_dropped(),

84 _ => false,

85 }

86 }

87

88 pub fn is_retryable(&self) -> bool {

89 match self {

90 Self::Pool(PoolError::Timeout(_) | PoolError::Closed) => true,

91 Self::Pool(PoolError::Backend(error)) => retryable_backend(error),

92 Self::Backend(error) => retryable_backend(error),

93 _ => false,

94 }

95 }

96}

97

98fn retryable_backend(error: &RedisError) -> bool {

99 error.is_io_error()

100 || error.is_timeout()

101 || error.is_connection_refusal()

102 || error.is_connection_dropped()

103 || error.is_cluster_error()

104}

105

106impl std::fmt::Display for CacheError {

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

108 match self {

109 Self::Config(message) => write!(f, "cache config: {message}"),

110 Self::Pool(error) => write!(f, "cache pool: {error}"),

111 Self::Backend(error) => write!(f, "cache backend: {error}"),

112 Self::Encode { key, message } => write!(f, "cache encode {key}: {message}"),

113 Self::Decode { key, message } => write!(f, "cache decode {key}: {message}"),

114 Self::Unsupported { feature, reason } => {

115 write!(f, "cache unsupported {feature}: {reason}")

116 }

117 Self::Lock { key, message } => write!(f, "cache lock {key}: {message}"),

118 }

119 }

120}

121

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

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

124 match self {

125 Self::Pool(error) => Some(error),

126 Self::Backend(error) => Some(error),

127 _ => None,

128 }

129 }

130}

131

132impl From<RedisError> for CacheError {

133 fn from(error: RedisError) -> Self {

134 Self::Backend(error)

135 }

136}

137

138impl From<PoolError> for CacheError {

139 fn from(error: PoolError) -> Self {

140 match error {

141 PoolError::Backend(error) => Self::Backend(error),

142 other => Self::Pool(other),

143 }

144 }

145}

Deleted lib/kvcache/src/lib.rs +0 −24

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

1pub mod conn;

2pub mod error;

3pub mod exec;

4pub mod hash;

5pub mod json;

6pub mod keys;

7pub mod list;

8pub mod lock;

9pub mod pipeline;

10pub mod scripts;

11pub mod set;

12pub mod zset;

13

14mod kv;

15

16pub use conn::{CacheConn, CacheConnection};

17pub use error::{CacheError, CacheResult};

18pub use keys::{CacheKey, SEPARATOR, join, join_tagged, same_slot, slot_tag};

19pub use kv::Cache;

20pub use lock::{CacheLock, CacheLocks, RETRY_INTERVAL};

21pub use pipeline::CachePipeline;

22pub use scripts::{LuaScripts, RateLimit, unix_millis};

23

24pub use deadpool_redis::{self, PoolConfig, Status, Timeouts, redis};

Deleted lib/kvcache/Cargo.toml +0 −16

@@ -1,16 +0,0 @@

1[package]

2name = "kvcache"

3version.workspace = true

4edition.workspace = true

5

6[dependencies]

7config = { workspace = true }

8bytes = { workspace = true }

9deadpool-redis = { version = "0.23.1", features = ["cluster", "cluster-async", "script", "tokio-comp"] }

10futures-util = "0.3.34"

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

12serde_json = { workspace = true }

13thiserror = { workspace = true }

14tokio = { workspace = true, features = ["time", "sync"] }

15tracing = { workspace = true }

16uuid = { workspace = true }

Modified lib/session/src/error.rs +2 −2 View lib/session/src/error.rs

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

11use std::error::Error;

22use std::fmt::{self, Display, Formatter};

33

4use kvcache::CacheError;

5use kvcache::redis::RedisError;

4use cache::CacheError;

5use cache::redis::RedisError;

66

77#[derive(Debug)]

88pub enum SessionError {

Modified lib/session/src/redis.rs +2 −2 View lib/session/src/redis.rs

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

11use std::sync::LazyLock;

22

33use actix_web::cookie::time::Duration;

4use kvcache::redis::{AsyncCommands, Script, ScriptInvocation};

5use kvcache::{Cache, CacheConnection};

4use cache::redis::{AsyncCommands, Script, ScriptInvocation};

5use cache::{Cache, CacheConnection};

66

77use config::identity::redis::RedisConfig;

88

Modified lib/session/Cargo.toml +1 −1 View lib/session/Cargo.toml

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

55

66[dependencies]

77config = { workspace = true }

8kvcache = { workspace = true }

8cache = { workspace = true }

99actix-web = "4.16.0"

1010bytes = { workspace = true }

1111rand = "0.10.3"

Modified Cargo.lock +7 −50 View Cargo.lock

@@ -1166,10 +1166,14 @@

11661166name = "cache"

11671167version = "1.0.0"

11681168dependencies = [

1169 "export",

1169 "config 1.0.0",

1170 "futures-util",

11701171 "redis",

11711172 "serde",

11721173 "serde_json",

1174 "tokio",

1175 "tracing",

1176 "uuid",

11731177]

11741178

11751179[[package]]

@@ -1847,35 +1851,6 @@

18471851]

18481852

18491853[[package]]

1850name = "deadpool"

1851version = "0.13.1"

1852source = "registry+https://github.com/rust-lang/crates.io-index"

1853checksum = "3e98a7e119cd347f4201e1159b19831029e203e2d8b790547708e8157b4acf1e"

1854dependencies = [

1855 "deadpool-runtime",

1856 "tokio",

1857]

1858

1859[[package]]

1860name = "deadpool-redis"

1861version = "0.23.1"

1862source = "registry+https://github.com/rust-lang/crates.io-index"

1863checksum = "243852fff546d8c5aac6ceaac7587f7137915b102c3fbec6256a64ce52452bd2"

1864dependencies = [

1865 "deadpool",

1866 "redis",

1867]

1868

1869[[package]]

1870name = "deadpool-runtime"

1871version = "0.3.1"

1872source = "registry+https://github.com/rust-lang/crates.io-index"

1873checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae"

1874dependencies = [

1875 "tokio",

1876]

1877

1878[[package]]

18791854name = "deletion"

18801855version = "1.0.0"

18811856dependencies = [

@@ -2771,7 +2746,6 @@

27712746 "async-trait",

27722747 "base64 0.23.1",

27732748 "bytes",

2774 "cache",

27752749 "config 1.0.0",

27762750 "export",

27772751 "flate2",

@@ -3492,11 +3466,11 @@

34923466 "async-trait",

34933467 "base64 0.23.1",

34943468 "bytes",

3469 "cache",

34953470 "chrono",

34963471 "config 1.0.0",

34973472 "crypt",

34983473 "db",

3499 "deadpool-redis",

35003474 "ed25519-dalek",

35013475 "email",

35023476 "envelope",

@@ -3506,7 +3480,6 @@

35063480 "hmac 0.12.1",

35073481 "http 1.5.0",

35083482 "http-body 1.1.0",

3509 "kvcache",

35103483 "lettre",

35113484 "maud",

35123485 "models",

@@ -4014,22 +3987,6 @@

40143987 "euclid",

40153988 "polycool",

40163989 "smallvec",

4017]

4018

4019[[package]]

4020name = "kvcache"

4021version = "1.0.0"

4022dependencies = [

4023 "bytes",

4024 "config 1.0.0",

4025 "deadpool-redis",

4026 "futures-util",

4027 "serde",

4028 "serde_json",

4029 "thiserror 2.0.21",

4030 "tokio",

4031 "tracing",

4032 "uuid",

40333990]

40343991

40353992[[package]]

@@ -7153,8 +7110,8 @@

71537110dependencies = [

71547111 "actix-web",

71557112 "bytes",

7113 "cache",

71567114 "config 1.0.0",

7157 "kvcache",

71587115 "rand 0.10.3",

71597116 "serde",

71607117 "serde_json",

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

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

3535email = { path = "lib/email" }

3636envelope = { path = "lib/envelope" }

3737events = { path = "lib/events" }

38kvcache = { path = "lib/kvcache" }

3938oltp = { path = "lib/oltp" }

4039proto = { path = "lib/proto" }

4140session = { path = "lib/session" }