1use std::sync::Arc;
6use std::collections::HashMap;
7use chrono::{DateTime, Duration, Utc};
8use tokio::sync::RwLock;
9use sa_token_adapter::storage::SaStorage;
10use crate::config::SaTokenConfig;
11use crate::error::{SaTokenError, SaTokenResult};
12use crate::token::{TokenInfo, TokenValue, TokenGenerator};
13use crate::session::SaSession;
14use crate::event::{SaTokenEventBus, SaTokenEvent};
15use crate::online::OnlineManager;
16use crate::distributed::DistributedSessionManager;
17
18#[derive(Clone)]
20pub struct SaTokenManager {
21 pub(crate) storage: Arc<dyn SaStorage>,
22 pub config: SaTokenConfig,
23 pub(crate) user_permissions: Arc<RwLock<HashMap<String, Vec<String>>>>,
25 pub(crate) user_roles: Arc<RwLock<HashMap<String, Vec<String>>>>,
27 pub(crate) event_bus: SaTokenEventBus,
29 online_manager: Option<Arc<OnlineManager>>,
31 distributed_manager: Option<Arc<DistributedSessionManager>>,
33}
34
35impl SaTokenManager {
36 pub fn new(storage: Arc<dyn SaStorage>, config: SaTokenConfig) -> Self {
38 Self {
39 storage,
40 config,
41 user_permissions: Arc::new(RwLock::new(HashMap::new())),
42 user_roles: Arc::new(RwLock::new(HashMap::new())),
43 event_bus: SaTokenEventBus::new(),
44 online_manager: None,
45 distributed_manager: None,
46 }
47 }
48
49 pub fn with_online_manager(mut self, manager: Arc<OnlineManager>) -> Self {
50 self.online_manager = Some(manager);
51 self
52 }
53
54 pub fn with_distributed_manager(mut self, manager: Arc<DistributedSessionManager>) -> Self {
55 self.distributed_manager = Some(manager);
56 self
57 }
58
59 pub fn online_manager(&self) -> Option<&Arc<OnlineManager>> {
60 self.online_manager.as_ref()
61 }
62
63 pub fn distributed_manager(&self) -> Option<&Arc<DistributedSessionManager>> {
64 self.distributed_manager.as_ref()
65 }
66
67 pub fn event_bus(&self) -> &SaTokenEventBus {
69 &self.event_bus
70 }
71
72 pub async fn login(&self, login_id: impl Into<String>) -> SaTokenResult<TokenValue> {
74 self.login_with_options(login_id, None, None, None, None, None).await
75 }
76
77 pub async fn login_with_options(
99 &self,
100 login_id: impl Into<String>,
101 login_type: Option<String>,
102 device: Option<String>,
103 extra_data: Option<serde_json::Value>,
104 nonce: Option<String>,
105 expire_time: Option<DateTime<Utc>>,
106 ) -> SaTokenResult<TokenValue> {
107 let login_id = login_id.into();
108
109 let token = TokenGenerator::generate_with_login_id(&self.config, &login_id);
111
112 let mut token_info = TokenInfo::new(token.clone(), login_id.clone());
114
115 token_info.login_type = login_type.unwrap_or_else(|| "default".to_string());
117
118 if let Some(device_str) = device {
120 token_info.device = Some(device_str);
121 }
122
123 if let Some(extra) = extra_data {
125 token_info.extra_data = Some(extra);
126 }
127
128 if let Some(nonce_str) = nonce {
130 token_info.nonce = Some(nonce_str);
131 }
132
133 if let Some(custom_expire_time) = expire_time {
135 token_info.expire_time = Some(custom_expire_time);
136 }
137 self.login_with_token_info(token_info).await
141 }
142
143 pub async fn login_with_token_info(&self, mut token_info: TokenInfo) -> SaTokenResult<TokenValue> {
172 let login_id = token_info.login_id.clone();
173
174 let token = if token_info.token.as_str().is_empty() {
176 TokenGenerator::generate_with_login_id(&self.config, &login_id)
177 } else {
178 token_info.token.clone()
179 };
180
181 token_info.token = token.clone();
183
184 token_info.update_active_time();
186
187 let now = Utc::now();
189 if token_info.expire_time.is_none() {
190 if let Some(timeout) = self.config.timeout_duration() {
191 token_info.expire_time = Some(now + Duration::from_std(timeout).unwrap());
192 }
193 }
194
195 if token_info.login_type.is_empty() {
197 token_info.login_type = "default".to_string();
198 }
199
200 let key = format!("sa:token:{}", token.as_str());
202 let value = serde_json::to_string(&token_info)
203 .map_err(|e| SaTokenError::SerializationError(e))?;
204
205 self.storage.set(&key, &value, self.config.timeout_duration()).await
206 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
207
208 let login_token_key = if !token_info.login_type.is_empty() && token_info.login_type != "default" {
212 format!("sa:login:token:{}:{}", login_id, token_info.login_type)
213 } else {
214 format!("sa:login:token:{}", login_id)
215 };
216 self.storage.set(&login_token_key, token.as_str(), self.config.timeout_duration()).await
217 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
218
219 if !self.config.is_concurrent {
221 self.logout_by_login_id(&login_id).await?;
222 }
223
224 let event = SaTokenEvent::login(login_id.clone(), token.as_str())
226 .with_login_type(&token_info.login_type);
227 self.event_bus.publish(event).await;
228
229 Ok(token)
230 }
231
232 pub async fn logout(&self, token: &TokenValue) -> SaTokenResult<()> {
234 tracing::debug!("Manager: 开始 logout,token: {}", token);
235
236 let key = format!("sa:token:{}", token.as_str());
238 tracing::debug!("Manager: 查询 token 信息,key: {}", key);
239
240 let token_info_str = self.storage.get(&key).await
241 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
242
243 let token_info = if let Some(value) = token_info_str {
244 tracing::debug!("Manager: 找到 token 信息: {}", value);
245 serde_json::from_str::<TokenInfo>(&value).ok()
246 } else {
247 tracing::debug!("Manager: 未找到 token 信息");
248 None
249 };
250
251 tracing::debug!("Manager: 删除 token,key: {}", key);
253 self.storage.delete(&key).await
254 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
255 tracing::debug!("Manager: token 已从存储中删除");
256
257 if let Some(info) = token_info.clone() {
259 tracing::debug!("Manager: 触发登出事件,login_id: {}, login_type: {}", info.login_id, info.login_type);
260 let event = SaTokenEvent::logout(&info.login_id, token.as_str())
261 .with_login_type(&info.login_type);
262 self.event_bus.publish(event).await;
263
264 if let Some(online_mgr) = &self.online_manager {
266 tracing::debug!("Manager: 标记用户下线,login_id: {}", info.login_id);
267 online_mgr.mark_offline(&info.login_id, token.as_str()).await;
268 }
269 }
270
271 tracing::debug!("Manager: logout 完成,token: {}", token);
272 Ok(())
273 }
274
275 pub async fn logout_by_login_id(&self, login_id: &str) -> SaTokenResult<()> {
277 let token_prefix = "sa:token:";
279
280 if let Ok(keys) = self.storage.keys(&format!("{}*", token_prefix)).await {
282 for key in keys {
284 if let Ok(Some(token_info_str)) = self.storage.get(&key).await {
286 if let Ok(token_info) = serde_json::from_str::<TokenInfo>(&token_info_str) {
288 if token_info.login_id == login_id {
290 let token_str = key[token_prefix.len()..].to_string();
292 let token = TokenValue::new(token_str);
293
294 let _ = self.logout(&token).await;
296 }
297 }
298 }
299 }
300 }
301
302 Ok(())
303 }
304
305 pub async fn get_token_info(&self, token: &TokenValue) -> SaTokenResult<TokenInfo> {
307 let key = format!("sa:token:{}", token.as_str());
308 let value = self.storage.get(&key).await
309 .map_err(|e| SaTokenError::StorageError(e.to_string()))?
310 .ok_or(SaTokenError::TokenNotFound)?;
311
312 let token_info: TokenInfo = serde_json::from_str(&value)
313 .map_err(|e| SaTokenError::SerializationError(e))?;
314
315 if token_info.is_expired() {
317 self.logout(token).await?;
319 return Err(SaTokenError::TokenExpired);
320 }
321
322 if self.config.auto_renew {
325 let renew_timeout = if self.config.active_timeout > 0 {
326 self.config.active_timeout
327 } else {
328 self.config.timeout
329 };
330
331 let _ = self.renew_timeout_internal(token, renew_timeout, &token_info).await;
333 }
334
335 Ok(token_info)
336 }
337
338 pub async fn is_valid(&self, token: &TokenValue) -> bool {
340 self.get_token_info(token).await.is_ok()
341 }
342
343 pub async fn get_session(&self, login_id: &str) -> SaTokenResult<SaSession> {
345 let key = format!("sa:session:{}", login_id);
346 let value = self.storage.get(&key).await
347 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
348
349 if let Some(value) = value {
350 let session: SaSession = serde_json::from_str(&value)
351 .map_err(|e| SaTokenError::SerializationError(e))?;
352 Ok(session)
353 } else {
354 Ok(SaSession::new(login_id))
355 }
356 }
357
358 pub async fn save_session(&self, session: &SaSession) -> SaTokenResult<()> {
360 let key = format!("sa:session:{}", session.id);
361 let value = serde_json::to_string(session)
362 .map_err(|e| SaTokenError::SerializationError(e))?;
363
364 self.storage.set(&key, &value, None).await
365 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
366
367 Ok(())
368 }
369
370 pub async fn delete_session(&self, login_id: &str) -> SaTokenResult<()> {
372 let key = format!("sa:session:{}", login_id);
373 self.storage.delete(&key).await
374 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
375 Ok(())
376 }
377
378 pub async fn renew_timeout(
380 &self,
381 token: &TokenValue,
382 timeout_seconds: i64,
383 ) -> SaTokenResult<()> {
384 let token_info = self.get_token_info(token).await?;
385 self.renew_timeout_internal(token, timeout_seconds, &token_info).await
386 }
387
388 async fn renew_timeout_internal(
390 &self,
391 token: &TokenValue,
392 timeout_seconds: i64,
393 token_info: &TokenInfo,
394 ) -> SaTokenResult<()> {
395 let mut new_token_info = token_info.clone();
396
397 use chrono::{Utc, Duration};
399 let new_expire_time = Utc::now() + Duration::seconds(timeout_seconds);
400 new_token_info.expire_time = Some(new_expire_time);
401
402 let key = format!("sa:token:{}", token.as_str());
404 let value = serde_json::to_string(&new_token_info)
405 .map_err(|e| SaTokenError::SerializationError(e))?;
406
407 let timeout = std::time::Duration::from_secs(timeout_seconds as u64);
408 self.storage.set(&key, &value, Some(timeout)).await
409 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
410
411 Ok(())
412 }
413
414 pub async fn kick_out(&self, login_id: &str) -> SaTokenResult<()> {
416 let token_result = self.storage.get(&format!("sa:login:token:{}", login_id)).await;
417
418 if let Some(online_mgr) = &self.online_manager {
419 let _ = online_mgr.kick_out_notify(login_id, "Account kicked out".to_string()).await;
420 }
421
422 self.logout_by_login_id(login_id).await?;
423 self.delete_session(login_id).await?;
424
425 if let Ok(Some(token_str)) = token_result {
426 let event = SaTokenEvent::kick_out(login_id, token_str);
427 self.event_bus.publish(event).await;
428 }
429
430 Ok(())
431 }
432}