fix(analytics): fix active users endpoint SQL type casting

.gitignore

@@ -9,3 +9,4 @@ repl-result-*
 .invoices
 .credit_notes
 .nixos-test-history
+/.idea

academy_api/rest/src/models/session.rs

@@ -1,6 +1,6 @@
 use academy_models::{
     auth::{AccessToken, Login, RefreshToken},
-    session::{DeviceName, Session, SessionId},
+    session::{ActiveUsersBucket, DeviceName, Session, SessionId},
     user::UserId,
 };
 use schemars::JsonSchema;
@@ -49,3 +49,18 @@ impl From<Login> for ApiLogin {
         }
     }
 }
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, JsonSchema)]
+pub struct ApiActiveUsersBucket {
+    pub bucket_start: i64,
+    pub active_users: u64,
+}
+
+impl From<ActiveUsersBucket> for ApiActiveUsersBucket {
+    fn from(value: ActiveUsersBucket) -> Self {
+        Self {
+            bucket_start: value.bucket_start.timestamp(),
+            active_users: value.active_users,
+        }
+    }
+}

academy_api/rest/src/routes/session.rs

@@ -1,9 +1,10 @@
 use std::sync::Arc;
 
 use academy_core_session_contracts::{
-    SessionCreateCommand, SessionCreateError, SessionDeleteByUserError, SessionDeleteCurrentError,
-    SessionDeleteError, SessionFeatureService, SessionGetCurrentError, SessionImpersonateError,
-    SessionListByUserError, SessionRefreshError,
+    ActiveUsersGranularity, ActiveUsersRange, SessionActiveUsersError, SessionCreateCommand,
+    SessionCreateError, SessionDeleteByUserError, SessionDeleteCurrentError, SessionDeleteError,
+    SessionFeatureService, SessionGetCurrentError, SessionImpersonateError, SessionListByUserError,
+    SessionRefreshError,
 };
 use academy_models::{
     RecaptchaResponse,
@@ -18,7 +19,7 @@ use aide::{
 };
 use axum::{
     Json,
-    extract::{Path, State},
+    extract::{Path, Query, State},
     http::StatusCode,
     response::{IntoResponse, Response},
 };
@@ -39,7 +40,7 @@ use crate::{
     extractors::{auth::ApiToken, user_agent::UserAgent},
     models::{
         OkResponse, StringOption,
-        session::{ApiLogin, ApiSession},
+        session::{ApiActiveUsersBucket, ApiLogin, ApiSession},
         user::{ApiUserIdOrSelf, PathUserId, PathUserIdOrSelf},
     },
 };
@@ -55,6 +56,10 @@ pub fn router(service: Arc<impl SessionFeatureService>) -> ApiRouter<()> {
                 .delete_with(delete_current, delete_current_docs),
         )
         .api_route("/auth/sessions", routing::post_with(create, create_docs))
+        .api_route(
+            "/analytics/active-users",
+            routing::get_with(active_users, active_users_docs),
+        )
         .api_route(
             "/auth/sessions/{user_id}",
             routing::get_with(list_by_user, list_by_user_docs)
@@ -87,6 +92,89 @@ fn get_current_docs(op: TransformOperation) -> TransformOperation {
         .with(internal_server_error_docs)
 }
 
+#[derive(Deserialize, JsonSchema)]
+struct ActiveUsersQuery {
+    range: ActiveUsersRangeParam,
+    granularity: ActiveUsersGranularityParam,
+}
+
+#[derive(Debug, Clone, Copy, Deserialize, JsonSchema)]
+enum ActiveUsersRangeParam {
+    #[serde(rename = "1d")]
+    Day1,
+    #[serde(rename = "7d")]
+    Day7,
+    #[serde(rename = "30d")]
+    Day30,
+    #[serde(rename = "90d")]
+    Day90,
+}
+
+impl From<ActiveUsersRangeParam> for ActiveUsersRange {
+    fn from(value: ActiveUsersRangeParam) -> Self {
+        match value {
+            ActiveUsersRangeParam::Day1 => Self::Day1,
+            ActiveUsersRangeParam::Day7 => Self::Day7,
+            ActiveUsersRangeParam::Day30 => Self::Day30,
+            ActiveUsersRangeParam::Day90 => Self::Day90,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Deserialize, JsonSchema)]
+enum ActiveUsersGranularityParam {
+    #[serde(rename = "1h")]
+    Hour1,
+    #[serde(rename = "1d")]
+    Day1,
+    #[serde(rename = "7d")]
+    Day7,
+    #[serde(rename = "30d")]
+    Day30,
+}
+
+impl From<ActiveUsersGranularityParam> for ActiveUsersGranularity {
+    fn from(value: ActiveUsersGranularityParam) -> Self {
+        match value {
+            ActiveUsersGranularityParam::Hour1 => Self::Hour1,
+            ActiveUsersGranularityParam::Day1 => Self::Day1,
+            ActiveUsersGranularityParam::Day7 => Self::Day7,
+            ActiveUsersGranularityParam::Day30 => Self::Day30,
+        }
+    }
+}
+
+async fn active_users(
+    session_service: State<Arc<impl SessionFeatureService>>,
+    token: ApiToken,
+    Query(ActiveUsersQuery { range, granularity }): Query<ActiveUsersQuery>,
+) -> Response {
+    let range = ActiveUsersRange::from(range);
+    let granularity = ActiveUsersGranularity::from(granularity);
+
+    match session_service
+        .active_users(&token.0, range, granularity)
+        .await
+    {
+        Ok(buckets) => Json(
+            buckets
+                .into_iter()
+                .map(ApiActiveUsersBucket::from)
+                .collect::<Vec<_>>(),
+        )
+        .into_response(),
+        Err(SessionActiveUsersError::Auth(err)) => auth_error(err),
+        Err(SessionActiveUsersError::Other(err)) => internal_server_error(err),
+    }
+}
+
+fn active_users_docs(op: TransformOperation) -> TransformOperation {
+    op.summary("Return active user counts grouped by time buckets.")
+        .add_response::<Vec<ApiActiveUsersBucket>>(StatusCode::OK, None)
+        .with(auth_error_docs)
+        .with(internal_server_error_docs)
+}
+
 async fn list_by_user(
     session_service: State<Arc<impl SessionFeatureService>>,
     token: ApiToken,

academy_core/session/contracts/Cargo.toml

@@ -15,5 +15,6 @@ mock = ["dep:mockall"]
 [dependencies]
 academy_models.workspace = true
 anyhow.workspace = true
+chrono.workspace = true
 mockall = { workspace = true, optional = true }
 thiserror.workspace = true

academy_core/session/contracts/src/lib.rs

@@ -4,14 +4,16 @@ use academy_models::{
     RecaptchaResponse,
     auth::{AccessToken, AuthError, Login, RefreshToken},
     mfa::MfaAuthentication,
-    session::{DeviceName, Session, SessionId},
+    session::{ActiveUsersBucket, DeviceName, Session, SessionId},
     user::{UserId, UserIdOrSelf, UserNameOrEmailAddress, UserPassword},
 };
 use thiserror::Error;
 
 pub mod failed_auth_count;
 pub mod session;
 
+pub use session::{ActiveUsersGranularity, ActiveUsersRange};
+
 pub trait SessionFeatureService: Send + Sync + 'static {
     /// Return the currently authenticated session.
     fn get_current_session(
@@ -81,6 +83,16 @@ pub trait SessionFeatureService: Send + Sync + 'static {
         token: &AccessToken,
         user_id: UserIdOrSelf,
     ) -> impl Future<Output = Result<(), SessionDeleteByUserError>> + Send;
+
+    /// Return active user counts for the specified range and granularity.
+    ///
+    /// Requires admin privileges.
+    fn active_users(
+        &self,
+        token: &AccessToken,
+        range: ActiveUsersRange,
+        granularity: ActiveUsersGranularity,
+    ) -> impl Future<Output = Result<Vec<ActiveUsersBucket>, SessionActiveUsersError>> + Send;
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -164,3 +176,11 @@ pub enum SessionDeleteByUserError {
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
+
+#[derive(Debug, Error)]
+pub enum SessionActiveUsersError {
+    #[error(transparent)]
+    Auth(#[from] AuthError),
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}

academy_core/session/contracts/src/session.rs

@@ -2,11 +2,50 @@ use std::future::Future;
 
 use academy_models::{
     auth::Login,
-    session::{DeviceName, SessionId},
+    session::{ActiveUsersBucket, DeviceName, SessionId},
     user::{UserComposite, UserId},
 };
+use chrono::Duration;
 use thiserror::Error;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ActiveUsersRange {
+    Day1,
+    Day7,
+    Day30,
+    Day90,
+}
+
+impl ActiveUsersRange {
+    pub fn duration(self) -> Duration {
+        match self {
+            Self::Day1 => Duration::days(1),
+            Self::Day7 => Duration::days(7),
+            Self::Day30 => Duration::days(30),
+            Self::Day90 => Duration::days(90),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ActiveUsersGranularity {
+    Hour1,
+    Day1,
+    Day7,
+    Day30,
+}
+
+impl ActiveUsersGranularity {
+    pub fn duration(self) -> Duration {
+        match self {
+            Self::Hour1 => Duration::hours(1),
+            Self::Day1 => Duration::days(1),
+            Self::Day7 => Duration::days(7),
+            Self::Day30 => Duration::days(30),
+        }
+    }
+}
+
 #[cfg_attr(feature = "mock", mockall::automock)]
 pub trait SessionService<Txn: Send + Sync + 'static>: Send + Sync + 'static {
     /// Create a new session for the given user.
@@ -41,6 +80,15 @@ pub trait SessionService<Txn: Send + Sync + 'static>: Send + Sync + 'static {
         txn: &mut Txn,
         user_id: UserId,
     ) -> impl Future<Output = anyhow::Result<()>> + Send;
+
+    /// Return the number of active users bucketed by the given range and
+    /// granularity.
+    fn active_users(
+        &self,
+        txn: &mut Txn,
+        range: ActiveUsersRange,
+        granularity: ActiveUsersGranularity,
+    ) -> impl Future<Output = anyhow::Result<Vec<ActiveUsersBucket>>> + Send;
 }
 
 #[derive(Debug, Error)]
@@ -108,4 +156,21 @@ impl<Txn: Send + Sync + 'static> MockSessionService<Txn> {
             .return_once(|_, _| Box::pin(std::future::ready(Ok(()))));
         self
     }
+
+    pub fn with_active_users(
+        mut self,
+        range: ActiveUsersRange,
+        granularity: ActiveUsersGranularity,
+        result: Vec<ActiveUsersBucket>,
+    ) -> Self {
+        self.expect_active_users()
+            .once()
+            .with(
+                mockall::predicate::always(),
+                mockall::predicate::eq(range),
+                mockall::predicate::eq(granularity),
+            )
+            .return_once(move |_, _, _| Box::pin(std::future::ready(Ok(result.clone()))));
+        self
+    }
 }

academy_core/session/impl/Cargo.toml

@@ -21,6 +21,7 @@ academy_persistence_contracts.workspace = true
 academy_shared_contracts.workspace = true
 academy_utils.workspace = true
 anyhow.workspace = true
+chrono.workspace = true
 hex.workspace = true
 tracing.workspace = true
 

academy_core/session/impl/src/lib.rs

@@ -5,16 +5,16 @@ use academy_core_mfa_contracts::authenticate::{
     MfaAuthenticateError, MfaAuthenticateResult, MfaAuthenticateService,
 };
 use academy_core_session_contracts::{
-    SessionCreateCommand, SessionCreateError, SessionDeleteByUserError, SessionDeleteCurrentError,
-    SessionDeleteError, SessionFeatureService, SessionGetCurrentError, SessionImpersonateError,
-    SessionListByUserError, SessionRefreshError, failed_auth_count::SessionFailedAuthCountService,
-    session::SessionService,
+    ActiveUsersGranularity, ActiveUsersRange, SessionActiveUsersError, SessionCreateCommand,
+    SessionCreateError, SessionDeleteByUserError, SessionDeleteCurrentError, SessionDeleteError,
+    SessionFeatureService, SessionGetCurrentError, SessionImpersonateError, SessionListByUserError,
+    SessionRefreshError, failed_auth_count::SessionFailedAuthCountService, session::SessionService,
 };
 use academy_di::Build;
 use academy_models::{
     RecaptchaResponse,
     auth::{AccessToken, Login, RefreshToken},
-    session::{Session, SessionId},
+    session::{ActiveUsersBucket, Session, SessionId},
     user::{UserId, UserIdOrSelf, UserNameOrEmailAddress},
 };
 use academy_persistence_contracts::{
@@ -380,4 +380,25 @@ where
 
         Ok(())
     }
+
+    #[trace_instrument(skip(self))]
+    async fn active_users(
+        &self,
+        token: &AccessToken,
+        range: ActiveUsersRange,
+        granularity: ActiveUsersGranularity,
+    ) -> Result<Vec<ActiveUsersBucket>, SessionActiveUsersError> {
+        let auth = self.auth.authenticate(token).await.map_auth_err()?;
+        auth.ensure_admin().map_auth_err()?;
+
+        let mut txn = self.db.begin_transaction().await?;
+        let buckets = self
+            .session
+            .active_users(&mut txn, range, granularity)
+            .await
+            .context("Failed to load active user statistics")?;
+        txn.commit().await?;
+
+        Ok(buckets)
+    }
 }