feat(sessions): identify online sessions (connected users)

identify online session by writing the `last_seen_at`
information is stored in a map and flush each 30s to prevent performance impact on pgsql

src/infrastructure/services/jwt_service.rs

@@ -55,6 +55,17 @@ struct JwtClaims {
     /// Serialised as `{"cnf": {"jkt": "..."}}` to match RFC 9449.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub cnf: Option<CnfClaim>,
+    /// OIDC-style `sid` claim (RFC 8417 §4.1) — carries the
+    /// `auth.sessions.id` this access token was minted for so the
+    /// auth middleware can stamp per-session liveness without a DB
+    /// round trip. `None` on tokens minted by pre-`sid` builds so
+    /// deserialisation stays backward-compatible during rollout.
+    /// Kept as `String` on the wire (Uuid parses at the port
+    /// boundary) so a malformed value fails at token-decode time
+    /// with a clear parse error instead of poisoning the field
+    /// silently.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sid: Option<String>,
 }
 
 /// RFC 9449 §5 confirmation-key wrapper. Only the `jkt` member is
@@ -72,6 +83,17 @@ impl From<JwtClaims> for TokenClaims {
         // signed always carries a UUID `sub`; nil is a safe sentinel the
         // middleware rejects. See benches/ROUND14.md §A3.
         let sub_id = uuid::Uuid::parse_str(&claims.sub).unwrap_or_else(|_| uuid::Uuid::nil());
+        // Parse `sid` at the boundary — same amortization rationale
+        // as `sub_id` above, and gives us a clean `Option<Uuid>` in
+        // `TokenClaims`. A parse failure (mint-time bug or hand-
+        // crafted claim) drops the sid to `None`; the middleware
+        // then simply skips the stamp — token still authenticates.
+        // Legitimate tokens minted by this codebase always carry a
+        // valid Uuid, so this only masks external drift.
+        let sid = claims
+            .sid
+            .as_deref()
+            .and_then(|s| uuid::Uuid::parse_str(s).ok());
         TokenClaims {
             sub_id,
             sub: claims.sub,
@@ -82,6 +104,7 @@ impl From<JwtClaims> for TokenClaims {
             email: claims.email,
             role: claims.role,
             dpop_jkt: claims.cnf.map(|c| c.jkt),
+            sid,
         }
     }
 }
@@ -196,6 +219,7 @@ impl TokenServicePort for JwtTokenService {
     fn generate_access_token(
         &self,
         user: &User,
+        session_id: Option<Uuid>,
         dpop_jkt: Option<&str>,
     ) -> Result<String, DomainError> {
         let now = Utc::now().timestamp();
@@ -219,6 +243,7 @@ impl TokenServicePort for JwtTokenService {
             cnf: dpop_jkt.map(|jkt| CnfClaim {
                 jkt: jkt.to_string(),
             }),
+            sid: session_id.map(|id| id.to_string()),
         };
 
         // Log JWT claims for debugging
@@ -331,7 +356,7 @@ mod tests {
 
         let user = create_test_user();
         let token = service
-            .generate_access_token(&user, None)
+            .generate_access_token(&user, Some(Uuid::new_v4()), None)
             .expect("Should generate token");
 
         let claims = service
@@ -370,7 +395,7 @@ mod tests {
 
         let user = create_test_user();
         let token = service
-            .generate_access_token(&user, None)
+            .generate_access_token(&user, Some(Uuid::new_v4()), None)
             .expect("Should generate token");
 
         // First call: cache miss — performs full HMAC verification
@@ -397,7 +422,7 @@ mod tests {
             86400,
         );
         let token = service
-            .generate_access_token(&create_test_user(), None)
+            .generate_access_token(&create_test_user(), Some(Uuid::new_v4()), None)
             .expect("Should generate token");
 
         // Miss populates the cache; hit must hand back the very same
@@ -425,4 +450,46 @@ mod tests {
         let (hits, _misses) = service.cache_stats();
         assert_eq!(hits, 0, "Invalid tokens should never produce cache hits");
     }
+
+    /// Regression for the `sid` claim wiring — the auth middleware
+    /// stamps per-session liveness by reading this exact field. If
+    /// the mint stops setting the claim or the port stops parsing
+    /// it, every `LastSeenTracker::stamp` call goes silent and the
+    /// Prometheus gauges freeze at zero.
+    #[test]
+    fn access_token_round_trips_session_id_as_sid_claim() {
+        let service = JwtTokenService::new(
+            "test_secret_key_at_least_32_bytes_long".to_string(),
+            3600,
+            86400,
+        );
+        let user = create_test_user();
+        let session_id = Uuid::new_v4();
+        let token = service
+            .generate_access_token(&user, Some(session_id), None)
+            .expect("Should generate token");
+        let claims = service.validate_token(&token).expect("Should validate");
+        assert_eq!(claims.sid, Some(session_id));
+    }
+
+    /// Backward-compatibility guard: a mint call with `None`
+    /// omits the `sid` claim entirely (matches the pre-`sid`
+    /// on-wire shape), and the validated claims surface `None`
+    /// on the port. The middleware's `if let (Some(sid), ...)`
+    /// then simply skips the stamp — critical during rollout
+    /// where old tokens are still in flight.
+    #[test]
+    fn access_token_without_session_id_omits_sid_claim() {
+        let service = JwtTokenService::new(
+            "test_secret_key_at_least_32_bytes_long".to_string(),
+            3600,
+            86400,
+        );
+        let user = create_test_user();
+        let token = service
+            .generate_access_token(&user, None, None)
+            .expect("Should generate token");
+        let claims = service.validate_token(&token).expect("Should validate");
+        assert_eq!(claims.sid, None);
+    }
 }

src/infrastructure/services/last_seen_tracker.rs

@@ -0,0 +1,239 @@
+//! Per-session liveness tracker — the hot path of the "how many
+//! sessions are active right now?" observation loop.
+//!
+//! **Contract.** Every authenticated request calls
+//! [`LastSeenTracker::stamp`] with the session id it resolved. The
+//! call is O(1) — a DashMap upsert of `(session_id → Utc::now())` —
+//! and hits no I/O. The map data structure IS the dedup: 100
+//! requests against the same session in a flush window contribute
+//! ONE row to the batched UPDATE with the latest timestamp.
+//!
+//! A background task ([`flush_loop`](Self::flush_loop), spawned at
+//! construction) drains the map every 30 s and issues one
+//! `UPDATE ... FROM UNNEST($1::uuid[], $2::timestamptz[])` covering
+//! every distinct session_id observed in the window. The
+//! `greatest(s.last_seen_at, t.seen_at)` guard makes the write
+//! idempotent under any retry / race / clock skew — replaying the
+//! same batch never moves the column backward.
+//!
+//! **Failure model.** A flush that hits a transient PG error does
+//! NOT drop the accumulated set — the map is not cleared until the
+//! UPDATE succeeds. Next tick overlays new activity on the retry
+//! set and the whole thing gets flushed together. Bounded loss
+//! window under a hard crash is one flush interval; graceful
+//! shutdown calls [`flush_now`](Self::flush_now) synchronously (see
+//! `main.rs`) so rolling restarts drop nothing.
+//!
+//! **Non-goals.** No per-session locking, no ordering guarantees
+//! across sessions, no back-pressure on the flusher (the loop
+//! swallows errors and keeps ticking). The workload is
+//! observation-only — losing a stamp under contention is a
+//! correctness no-op, the next request re-stamps.
+//!
+//! See `docs/plan/sessions.md` for the full design (why DashMap
+//! over Mutex<HashMap>, why not NOTIFY/LISTEN today, migration
+//! path to a multi-instance cluster).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use chrono::{DateTime, Utc};
+use dashmap::DashMap;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+/// Cadence of the batched UPDATE. Hardcoded — 30 s balances DB
+/// write load against gauge freshness (the Prometheus scrape
+/// interval is typically 15 s, so at worst two scrapes see the
+/// same value before the next flush). Deliberately NOT exposed as
+/// an env var — tuning it is a deployment-shape question we've
+/// never had to answer in practice.
+const FLUSH_INTERVAL: Duration = Duration::from_secs(30);
+
+/// In-process session-liveness tracker. See [module docs](self) for
+/// the full contract; the two entry points are:
+///
+/// - [`stamp`](Self::stamp) — called from the auth middleware on
+///   every authenticated request.
+/// - [`flush_now`](Self::flush_now) — called from the graceful-
+///   shutdown handler.
+///
+/// The periodic flush task is spawned on the tokio runtime by
+/// [`start`](Self::start) at construction. The struct keeps no
+/// handle to it — the task holds the `Arc<Self>` and observes the
+/// runtime shutting down naturally.
+pub struct LastSeenTracker {
+    /// (session_id → last observed time). DashMap's sharded locking
+    /// parallelises writes across distinct session_ids — different
+    /// users' requests never contend.
+    seen: DashMap<Uuid, DateTime<Utc>>,
+    /// Maintenance pool — the tracker is a background writer and
+    /// must not compete with request-serving connections.
+    pool: Arc<PgPool>,
+}
+
+impl LastSeenTracker {
+    /// Construct + spawn the flush loop. Returns the shared
+    /// handle; callers store it on `AppState` and pass it to the
+    /// auth middleware.
+    ///
+    /// The background task lives for the runtime's lifetime — no
+    /// cancellation handle is exposed because there is no
+    /// mid-process reason to stop tracking (a stopped flusher is
+    /// indistinguishable from a wedged one, and both are bugs).
+    /// Graceful shutdown calls [`flush_now`](Self::flush_now)
+    /// separately BEFORE the runtime tears down.
+    pub fn start(pool: Arc<PgPool>) -> Arc<Self> {
+        let this = Arc::new(Self {
+            seen: DashMap::new(),
+            pool,
+        });
+        tokio::spawn(this.clone().flush_loop());
+        this
+    }
+
+    /// Record that `session_id` was observed serving a request
+    /// right now. Overwrites any prior stamp for the same session
+    /// in the current window — the flusher uses the latest value.
+    ///
+    /// O(1) DashMap upsert. No I/O. Never fails.
+    pub fn stamp(&self, session_id: Uuid) {
+        self.seen.insert(session_id, Utc::now());
+    }
+
+    /// Drain the accumulated stamps and write them in one batched
+    /// UPDATE. Idempotent — the `greatest(...)` guard means
+    /// replaying the same batch (or overlapping batches from a
+    /// retry) never moves the column backward.
+    ///
+    /// Errors are surfaced to the caller so `flush_loop`'s
+    /// warn-and-continue policy is a deliberate choice made in one
+    /// place, and the shutdown flusher in `main.rs` can decide
+    /// whether to log or panic.
+    ///
+    /// On PG error the accumulated set is NOT cleared — the next
+    /// tick retries with fresh activity overlaid.
+    pub async fn flush_now(&self) -> Result<usize, sqlx::Error> {
+        if self.seen.is_empty() {
+            return Ok(0);
+        }
+
+        // Drain into two parallel vectors — one UNNEST arg each.
+        // `retain(|_,_| false)` clears every shard in-place; the
+        // pull-and-drop order doesn't matter (we upserted the
+        // latest wins per key already).
+        let mut ids: Vec<Uuid> = Vec::with_capacity(self.seen.len());
+        let mut seen_at: Vec<DateTime<Utc>> = Vec::with_capacity(self.seen.len());
+        for entry in self.seen.iter() {
+            ids.push(*entry.key());
+            seen_at.push(*entry.value());
+        }
+
+        let result = sqlx::query(
+            r#"
+            UPDATE auth.sessions AS s
+               SET last_seen_at = greatest(s.last_seen_at, t.seen_at)
+              FROM UNNEST($1::uuid[], $2::timestamptz[]) AS t(id, seen_at)
+             WHERE s.id = t.id
+            "#,
+        )
+        .bind(&ids)
+        .bind(&seen_at)
+        .execute(&*self.pool)
+        .await?;
+
+        // Only clear the drained keys on success. A key inserted
+        // BETWEEN our copy above and the clear below survives
+        // (retain drops only those whose value we already flushed,
+        // by timestamp equality). Same-key re-stamp with a newer
+        // timestamp gets kept for the next flush.
+        let flushed: std::collections::HashMap<Uuid, DateTime<Utc>> =
+            ids.iter().copied().zip(seen_at.iter().copied()).collect();
+        self.seen
+            .retain(|k, v| flushed.get(k).is_none_or(|ts| ts != v));
+
+        let updated = result.rows_affected() as usize;
+        tracing::debug!(
+            target: "oxicloud::sessions",
+            batched = ids.len(),
+            updated,
+            "last_seen flush",
+        );
+        Ok(updated)
+    }
+
+    /// The periodic drain loop. Runs forever; every failed flush
+    /// is logged at WARN and the accumulated set is preserved for
+    /// the next tick.
+    async fn flush_loop(self: Arc<Self>) {
+        let mut ticker = tokio::time::interval(FLUSH_INTERVAL);
+        // Skip the "first tick fires immediately" behaviour — the
+        // map is empty at spawn time, so a same-tick flush is
+        // wasted work.
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        ticker.tick().await;
+
+        loop {
+            ticker.tick().await;
+            if let Err(err) = self.flush_now().await {
+                tracing::warn!(
+                    target: "oxicloud::sessions",
+                    error = %err,
+                    "last_seen flush failed; will retry next tick",
+                );
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Ten stamps of the same session_id must collapse to ONE
+    /// entry with the newest timestamp — the whole point of the
+    /// DashMap-as-dedup pattern. Guards against a future refactor
+    /// that swaps to an append-only channel and doubles the DB
+    /// write rate.
+    #[test]
+    fn stamps_dedup_by_session_id() {
+        // No pool needed — we're only exercising the map. Build
+        // the tracker directly without spawning the loop.
+        let seen = DashMap::new();
+        let session = Uuid::new_v4();
+
+        for _ in 0..10 {
+            seen.insert(session, Utc::now());
+        }
+
+        assert_eq!(seen.len(), 1);
+    }
+
+    /// Latest-wins semantics: two stamps for the same session
+    /// leave the newer timestamp in place, matching the flusher's
+    /// `greatest(...)` guard so a request that beats the flush
+    /// keeps its more recent stamp.
+    #[test]
+    fn stamp_keeps_latest_timestamp() {
+        let seen: DashMap<Uuid, DateTime<Utc>> = DashMap::new();
+        let session = Uuid::new_v4();
+
+        let t1 = Utc::now();
+        seen.insert(session, t1);
+        let t2 = t1 + chrono::Duration::seconds(5);
+        seen.insert(session, t2);
+
+        assert_eq!(*seen.get(&session).unwrap(), t2);
+    }
+
+    /// Distinct sessions never collide — sharded map, no dedup
+    /// across keys.
+    #[test]
+    fn different_sessions_are_independent() {
+        let seen: DashMap<Uuid, DateTime<Utc>> = DashMap::new();
+        for _ in 0..100 {
+            seen.insert(Uuid::new_v4(), Utc::now());
+        }
+        assert_eq!(seen.len(), 100);
+    }
+}

src/infrastructure/services/mod.rs

@@ -27,6 +27,7 @@ pub mod folders_consistency_service;
 pub mod grant_cleanup_service;
 pub mod image_transcode_service;
 pub mod jwt_service;
+pub mod last_seen_tracker;
 pub mod local_blob_backend;
 pub mod local_fs_mount_provider;
 pub mod login_lockout_service;
@@ -51,6 +52,7 @@ pub mod retry_blob_backend;
 pub mod s3_blob_backend;
 pub mod search_index;
 pub mod session_cleanup_service;
+pub mod session_liveness_gauges;
 pub mod share_unlock_cookie;
 pub mod smtp_email_sender;
 pub mod swappable_blob_backend;

src/infrastructure/services/session_liveness_gauges.rs

@@ -0,0 +1,169 @@
+//! Prometheus session-liveness gauges — periodic polling of
+//! `auth.sessions` to publish three gauges the `/metrics` scraper
+//! reads:
+//!
+//! - `oxicloud_sessions_online` — non-revoked rows observed in the
+//!   last [`ONLINE_WINDOW`](crate::application::dtos::session_dto::ONLINE_WINDOW).
+//!   **Per-session count**, not per-user — one user with three
+//!   devices contributes three.
+//! - `oxicloud_sessions_online_users` — DISTINCT `user_id` behind
+//!   those online sessions. The multi-device factor is exactly
+//!   `sessions_online / sessions_online_users`.
+//! - `oxicloud_sessions_total_non_revoked` — long-tail total,
+//!   including mobile clients still holding a refresh token they
+//!   haven't used in weeks. Useful sanity signal on the dashboard.
+//!
+//! **Naming — "online" vs "active".** The word "active" is already
+//! spoken for by the session *lifecycle* (Active | Expired |
+//! Revoked in the admin panel). Presence (recently-seen) is
+//! orthogonal and uses "online" throughout the UI, DTO
+//! (`SessionSummaryDto::is_online`), and these gauges — so a
+//! dashboard graph and a per-row green-dot badge have the same
+//! label root. Terminology decided 2026-08-18; see
+//! `docs/plan/sessions.md`.
+//!
+//! **Cadence.** Poller ticks every [`POLL_INTERVAL`] (30 s). Three
+//! `COUNT(*)` reads on the maintenance pool per tick — negligible
+//! load on tens-of-thousands-of-rows tables thanks to the partial
+//! index `idx_sessions_last_seen_at` (partial on `revoked = FALSE`,
+//! which every query below filters on).
+//!
+//! **When it runs.** Spawned from DI only when auth is enabled AND
+//! `OXICLOUD_METRICS_LISTEN` is set (recorder installed). Without
+//! the recorder, `metrics::gauge!(...)` is a no-op — spawning
+//! anyway would still hit PG every 30 s for values nobody reads.
+//!
+//! See `docs/plan/sessions.md` for the full design.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use sqlx::PgPool;
+
+use crate::application::dtos::session_dto::ONLINE_WINDOW;
+
+/// Poll cadence. Matches the [`LastSeenTracker`](super::last_seen_tracker)
+/// flush cadence so the gauges converge one tick after the tracker
+/// flushes — no need to sync the two.
+const POLL_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Spawn the session-liveness poller. Detached — the task lives
+/// for the runtime's lifetime; there's no mid-process reason to
+/// stop reporting gauges.
+///
+/// Emits an initial poll on spawn so the very first `/metrics`
+/// scrape after boot returns real values instead of the recorder's
+/// zero-initialised default.
+pub fn spawn(maintenance_pool: Arc<PgPool>) {
+    tokio::spawn(async move {
+        // Immediate first tick — a scraper hitting `/metrics` in
+        // the first 30 s otherwise sees `oxicloud_sessions_online
+        // 0` even on a busy server. Warmup query is cheap.
+        if let Err(err) = poll_once(&maintenance_pool).await {
+            tracing::warn!(
+                target: "oxicloud::sessions",
+                error = %err,
+                "initial session-liveness poll failed",
+            );
+        }
+
+        let mut ticker = tokio::time::interval(POLL_INTERVAL);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        // Consume the first tick — `interval` fires immediately on
+        // creation and we've already done the warmup above.
+        ticker.tick().await;
+
+        loop {
+            ticker.tick().await;
+            if let Err(err) = poll_once(&maintenance_pool).await {
+                tracing::warn!(
+                    target: "oxicloud::sessions",
+                    error = %err,
+                    "session-liveness poll failed; keeping last-known gauge values",
+                );
+            }
+        }
+    });
+    tracing::info!(
+        target: "oxicloud::sessions",
+        poll_interval_secs = POLL_INTERVAL.as_secs(),
+        online_window_secs = ONLINE_WINDOW.as_secs(),
+        "📊 session-liveness gauges spawned",
+    );
+}
+
+/// One poll cycle. Three lightweight `COUNT` reads → three gauge
+/// updates. Errors propagate to the caller (loop logs + retries
+/// next tick; gauges keep their last-known value in the interim,
+/// which is the honest thing to publish — a temporary PG blip is
+/// not a "sessions dropped to zero" event).
+async fn poll_once(pool: &PgPool) -> Result<(), sqlx::Error> {
+    // NOTE: `ONLINE_WINDOW` is a Duration; PG expects the interval
+    // in seconds via `make_interval` (portable across sqlx driver
+    // versions). Casting once at bind time is cheaper than an
+    // `INTERVAL '$1 seconds'` string interp and keeps the query
+    // parameterised.
+    let online_secs: f64 = ONLINE_WINDOW.as_secs_f64();
+
+    let online: i64 = sqlx::query_scalar(
+        r#"
+        SELECT COUNT(*) FROM auth.sessions
+        WHERE revoked = FALSE
+          AND last_seen_at > NOW() - make_interval(secs => $1)
+        "#,
+    )
+    .bind(online_secs)
+    .fetch_one(pool)
+    .await?;
+
+    let online_users: i64 = sqlx::query_scalar(
+        r#"
+        SELECT COUNT(DISTINCT user_id) FROM auth.sessions
+        WHERE revoked = FALSE
+          AND last_seen_at > NOW() - make_interval(secs => $1)
+        "#,
+    )
+    .bind(online_secs)
+    .fetch_one(pool)
+    .await?;
+
+    let total_non_revoked: i64 = sqlx::query_scalar(
+        r#"
+        SELECT COUNT(*) FROM auth.sessions WHERE revoked = FALSE
+        "#,
+    )
+    .fetch_one(pool)
+    .await?;
+
+    // metrics-exporter-prometheus takes f64 gauges; the raw COUNT
+    // fits into f64 precisely up to 2^53, well past any realistic
+    // session-row count. `describe_gauge!` is called once at first
+    // emission and cached in the recorder — the second/third tick
+    // just updates the value.
+    metrics::describe_gauge!(
+        "oxicloud_sessions_online",
+        "Non-revoked sessions observed in the last ONLINE_WINDOW."
+    );
+    metrics::gauge!("oxicloud_sessions_online").set(online as f64);
+
+    metrics::describe_gauge!(
+        "oxicloud_sessions_online_users",
+        "Distinct users behind sessions observed in the last ONLINE_WINDOW."
+    );
+    metrics::gauge!("oxicloud_sessions_online_users").set(online_users as f64);
+
+    metrics::describe_gauge!(
+        "oxicloud_sessions_total_non_revoked",
+        "Total non-revoked sessions regardless of last-seen recency."
+    );
+    metrics::gauge!("oxicloud_sessions_total_non_revoked").set(total_non_revoked as f64);
+
+    tracing::debug!(
+        target: "oxicloud::sessions",
+        online,
+        online_users,
+        total_non_revoked,
+        "session-liveness gauges updated",
+    );
+    Ok(())
+}