systemprompt_oauth/services/session/
mod.rs

1//! Anonymous session creation and reuse.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6mod creation;
7mod lookup;
8
9use crate::error::OauthResult;
10use http::{HeaderMap, Uri};
11use std::collections::HashMap;
12use std::sync::Arc;
13use tokio::sync::RwLock;
14use uuid::Uuid;
15
16use systemprompt_identifiers::{ClientId, SessionId, SessionSource, UserId};
17use systemprompt_traits::{
18    AnalyticsProvider, CreateSessionInput, FingerprintProvider, SessionAnalytics, UserEvent,
19    UserEventPublisher, UserProvider,
20};
21
22const MAX_SESSION_AGE_SECONDS: i64 = 7 * 24 * 60 * 60;
23
24#[derive(Debug, thiserror::Error)]
25pub enum SessionCreationError {
26    #[error("User not found: {user_id}")]
27    UserNotFound { user_id: UserId },
28
29    #[error("Session creation failed: {0}")]
30    Internal(String),
31}
32
33struct SessionCreationParams<'a> {
34    analytics: SessionAnalytics,
35    is_bot: bool,
36    is_ai_crawler: bool,
37    fingerprint: String,
38    client_id: &'a ClientId,
39    session_source: SessionSource,
40}
41
42#[derive(Debug, Clone)]
43pub struct AnonymousSessionInfo {
44    pub session_id: SessionId,
45    pub user_id: UserId,
46    pub is_new: bool,
47    pub jwt_token: String,
48    pub fingerprint_hash: String,
49}
50
51#[derive(Debug, Clone)]
52pub struct AuthenticatedSessionInfo {
53    pub session_id: SessionId,
54}
55
56#[derive(Debug)]
57pub struct CreateAnonymousSessionInput<'a> {
58    pub headers: &'a HeaderMap,
59    pub uri: Option<&'a Uri>,
60    pub client_id: &'a ClientId,
61    pub session_source: SessionSource,
62}
63
64#[derive(Clone)]
65pub struct SessionCreationService {
66    analytics_provider: Arc<dyn AnalyticsProvider>,
67    user_provider: Arc<dyn UserProvider>,
68    fingerprint_locks: Arc<RwLock<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
69    event_publisher: Option<Arc<dyn UserEventPublisher>>,
70    fingerprint_provider: Option<Arc<dyn FingerprintProvider>>,
71}
72
73impl std::fmt::Debug for SessionCreationService {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("SessionCreationService")
76            .field("analytics_provider", &"<provider>")
77            .field(
78                "event_publisher",
79                &self.event_publisher.as_ref().map(|_| "<publisher>"),
80            )
81            .finish_non_exhaustive()
82    }
83}
84
85impl SessionCreationService {
86    pub fn new(
87        analytics_provider: Arc<dyn AnalyticsProvider>,
88        user_provider: Arc<dyn UserProvider>,
89    ) -> Self {
90        Self {
91            analytics_provider,
92            user_provider,
93            fingerprint_locks: Arc::new(RwLock::new(HashMap::new())),
94            event_publisher: None,
95            fingerprint_provider: None,
96        }
97    }
98
99    pub fn with_event_publisher(mut self, publisher: Arc<dyn UserEventPublisher>) -> Self {
100        self.event_publisher = Some(publisher);
101        self
102    }
103
104    pub fn with_fingerprint_provider(mut self, provider: Arc<dyn FingerprintProvider>) -> Self {
105        self.fingerprint_provider = Some(provider);
106        self
107    }
108
109    fn publish_event(&self, event: UserEvent) {
110        if let Some(ref publisher) = self.event_publisher {
111            publisher.publish_user_event(event);
112        }
113    }
114
115    /// Resolves the persisted users row for an unauthenticated caller without
116    /// creating a `user_sessions` row. The fingerprint derives from the
117    /// request analytics; repeated calls with the same fingerprint return the
118    /// same row by virtue of `UserProvider::create_anonymous`'s
119    /// `ON CONFLICT (email) DO UPDATE` upsert.
120    ///
121    /// Used by the session middleware for skip-tracked and bot-classified
122    /// traffic, where the analytics/session-row write is intentionally
123    /// suppressed but downstream handlers still need an FK-valid `user_id`.
124    pub async fn ensure_anonymous_user(
125        &self,
126        headers: &HeaderMap,
127        uri: Option<&Uri>,
128    ) -> OauthResult<(UserId, String)> {
129        let analytics = self.analytics_provider.extract_analytics(headers, uri);
130        let fingerprint = analytics.compute_fingerprint();
131        let user = self
132            .user_provider
133            .create_anonymous(&fingerprint)
134            .await
135            .map_err(|e| crate::error::OauthError::Session(e.to_string()))?;
136        Ok((user.id, fingerprint))
137    }
138
139    pub async fn create_anonymous_session(
140        &self,
141        input: CreateAnonymousSessionInput<'_>,
142    ) -> OauthResult<AnonymousSessionInfo> {
143        let analytics = self
144            .analytics_provider
145            .extract_analytics(input.headers, input.uri);
146        let is_ai_crawler = analytics.is_ai_crawler();
147        let is_bot = analytics.is_bot();
148        let fingerprint = analytics.compute_fingerprint();
149
150        let params = SessionCreationParams {
151            analytics,
152            is_bot,
153            is_ai_crawler,
154            fingerprint,
155            client_id: input.client_id,
156            session_source: input.session_source,
157        };
158        self.create_session_internal(params).await
159    }
160
161    pub async fn create_authenticated_session(
162        &self,
163        user_id: &UserId,
164        headers: &HeaderMap,
165        session_source: SessionSource,
166    ) -> Result<SessionId, SessionCreationError> {
167        let user = self
168            .user_provider
169            .find_by_id(user_id)
170            .await
171            .map_err(|e| SessionCreationError::Internal(e.to_string()))?;
172
173        if user.is_none() {
174            return Err(SessionCreationError::UserNotFound {
175                user_id: user_id.clone(),
176            });
177        }
178
179        let session_id = SessionId::new(format!("sess_{}", Uuid::new_v4()));
180        let analytics = self.analytics_provider.extract_analytics(headers, None);
181        let is_ai_crawler = analytics.is_ai_crawler();
182        let is_bot = analytics.is_bot();
183
184        let global_config = systemprompt_models::Config::get()
185            .map_err(|e| SessionCreationError::Internal(e.to_string()))?;
186        let expires_at = chrono::Utc::now()
187            + chrono::Duration::seconds(global_config.jwt_access_token_expiration);
188
189        self.analytics_provider
190            .create_session(CreateSessionInput {
191                session_id: &session_id,
192                user_id: Some(user_id),
193                analytics: &analytics,
194                session_source,
195                is_bot,
196                is_ai_crawler,
197                expires_at,
198            })
199            .await
200            .map_err(|e| SessionCreationError::Internal(e.to_string()))?;
201
202        self.publish_event(UserEvent::SessionCreated {
203            user_id: user_id.clone(),
204            session_id: session_id.clone(),
205        });
206
207        Ok(session_id)
208    }
209
210    async fn create_session_internal(
211        &self,
212        params: SessionCreationParams<'_>,
213    ) -> OauthResult<AnonymousSessionInfo> {
214        let _guard = self.acquire_fingerprint_lock(&params.fingerprint).await;
215
216        self.update_fingerprint_if_available(&params.fingerprint, &params.analytics)
217            .await;
218
219        if let Some(session) = self
220            .try_reuse_session_at_limit(&params.fingerprint, params.client_id)
221            .await
222        {
223            return Ok(session);
224        }
225
226        if let Some(session) = self
227            .try_find_existing_session(&params.fingerprint, params.client_id)
228            .await
229        {
230            return Ok(session);
231        }
232
233        self.create_new_session(params).await
234    }
235
236    async fn acquire_fingerprint_lock(
237        &self,
238        fingerprint: &str,
239    ) -> tokio::sync::OwnedMutexGuard<()> {
240        let lock = {
241            let mut locks = self.fingerprint_locks.write().await;
242            Arc::clone(
243                locks
244                    .entry(fingerprint.to_owned())
245                    .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))),
246            )
247        };
248        lock.lock_owned().await
249    }
250
251    async fn update_fingerprint_if_available(
252        &self,
253        fingerprint: &str,
254        analytics: &SessionAnalytics,
255    ) {
256        let Some(ref fp_provider) = self.fingerprint_provider else {
257            return;
258        };
259
260        if let Err(e) = fp_provider
261            .upsert_fingerprint(
262                fingerprint,
263                analytics.ip_address.as_deref(),
264                analytics.user_agent.as_deref(),
265                None,
266            )
267            .await
268        {
269            tracing::warn!(error = %e, fingerprint = %fingerprint, "Failed to upsert fingerprint");
270        }
271    }
272}
systemprompt_oauth/services/session/mod.rs

systemprompt_oauth/services/session/
mod.rs
