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 let key = format!("sa:token:{}", token.as_str());
236 let token_info_str = self.storage.get(&key).await
237 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
238
239 let token_info = if let Some(value) = token_info_str {
240 serde_json::from_str::<TokenInfo>(&value).ok()
241 } else {
242 None
243 };
244
245 self.storage.delete(&key).await
247 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
248
249 if let Some(info) = token_info.clone() {
251 let event = SaTokenEvent::logout(&info.login_id, token.as_str())
252 .with_login_type(&info.login_type);
253 self.event_bus.publish(event).await;
254
255 if let Some(online_mgr) = &self.online_manager {
257 online_mgr.mark_offline(&info.login_id, token.as_str()).await;
258 }
259 }
260
261 Ok(())
262 }
263
264 pub async fn logout_by_login_id(&self, login_id: &str) -> SaTokenResult<()> {
266 let token_prefix = "sa:token:";
268
269 if let Ok(keys) = self.storage.keys(&format!("{}*", token_prefix)).await {
271 for key in keys {
273 if let Ok(Some(token_info_str)) = self.storage.get(&key).await {
275 if let Ok(token_info) = serde_json::from_str::<TokenInfo>(&token_info_str) {
277 if token_info.login_id == login_id {
279 let token_str = key[token_prefix.len()..].to_string();
281 let token = TokenValue::new(token_str);
282
283 let _ = self.logout(&token).await;
285 }
286 }
287 }
288 }
289 }
290
291 Ok(())
292 }
293
294 pub async fn get_token_info(&self, token: &TokenValue) -> SaTokenResult<TokenInfo> {
296 let key = format!("sa:token:{}", token.as_str());
297 let value = self.storage.get(&key).await
298 .map_err(|e| SaTokenError::StorageError(e.to_string()))?
299 .ok_or(SaTokenError::TokenNotFound)?;
300
301 let token_info: TokenInfo = serde_json::from_str(&value)
302 .map_err(|e| SaTokenError::SerializationError(e))?;
303
304 if token_info.is_expired() {
306 self.logout(token).await?;
308 return Err(SaTokenError::TokenExpired);
309 }
310
311 if self.config.auto_renew {
314 let renew_timeout = if self.config.active_timeout > 0 {
315 self.config.active_timeout
316 } else {
317 self.config.timeout
318 };
319
320 let _ = self.renew_timeout_internal(token, renew_timeout, &token_info).await;
322 }
323
324 Ok(token_info)
325 }
326
327 pub async fn is_valid(&self, token: &TokenValue) -> bool {
329 self.get_token_info(token).await.is_ok()
330 }
331
332 pub async fn get_session(&self, login_id: &str) -> SaTokenResult<SaSession> {
334 let key = format!("sa:session:{}", login_id);
335 let value = self.storage.get(&key).await
336 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
337
338 if let Some(value) = value {
339 let session: SaSession = serde_json::from_str(&value)
340 .map_err(|e| SaTokenError::SerializationError(e))?;
341 Ok(session)
342 } else {
343 Ok(SaSession::new(login_id))
344 }
345 }
346
347 pub async fn save_session(&self, session: &SaSession) -> SaTokenResult<()> {
349 let key = format!("sa:session:{}", session.id);
350 let value = serde_json::to_string(session)
351 .map_err(|e| SaTokenError::SerializationError(e))?;
352
353 self.storage.set(&key, &value, None).await
354 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
355
356 Ok(())
357 }
358
359 pub async fn delete_session(&self, login_id: &str) -> SaTokenResult<()> {
361 let key = format!("sa:session:{}", login_id);
362 self.storage.delete(&key).await
363 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
364 Ok(())
365 }
366
367 pub async fn renew_timeout(
369 &self,
370 token: &TokenValue,
371 timeout_seconds: i64,
372 ) -> SaTokenResult<()> {
373 let token_info = self.get_token_info(token).await?;
374 self.renew_timeout_internal(token, timeout_seconds, &token_info).await
375 }
376
377 async fn renew_timeout_internal(
379 &self,
380 token: &TokenValue,
381 timeout_seconds: i64,
382 token_info: &TokenInfo,
383 ) -> SaTokenResult<()> {
384 let mut new_token_info = token_info.clone();
385
386 use chrono::{Utc, Duration};
388 let new_expire_time = Utc::now() + Duration::seconds(timeout_seconds);
389 new_token_info.expire_time = Some(new_expire_time);
390
391 let key = format!("sa:token:{}", token.as_str());
393 let value = serde_json::to_string(&new_token_info)
394 .map_err(|e| SaTokenError::SerializationError(e))?;
395
396 let timeout = std::time::Duration::from_secs(timeout_seconds as u64);
397 self.storage.set(&key, &value, Some(timeout)).await
398 .map_err(|e| SaTokenError::StorageError(e.to_string()))?;
399
400 Ok(())
401 }
402
403 pub async fn kick_out(&self, login_id: &str) -> SaTokenResult<()> {
405 let token_result = self.storage.get(&format!("sa:login:token:{}", login_id)).await;
406
407 if let Some(online_mgr) = &self.online_manager {
408 let _ = online_mgr.kick_out_notify(login_id, "Account kicked out".to_string()).await;
409 }
410
411 self.logout_by_login_id(login_id).await?;
412 self.delete_session(login_id).await?;
413
414 if let Ok(Some(token_str)) = token_result {
415 let event = SaTokenEvent::kick_out(login_id, token_str);
416 self.event_bus.publish(event).await;
417 }
418
419 Ok(())
420 }
421}