cratestack_axum/rpc/
credential_expiry.rs

1//! A subscription lives no longer than the credential that opened it
2//! (`docs/design/subscriptions-websocket.md` §6.3): the context captured at
3//! open is used for every event, so once its `exp` passes the stream ends
4//! with `unauthenticated` / `"credential expired"` and the client
5//! resubscribes with a fresh credential. A context without `exp` never
6//! expires here.
7
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use cratestack_core::{CratestackError, SubscriptionStream};
11use futures_util::future::{Either, select};
12use futures_util::stream::{self, StreamExt};
13
14fn expired() -> CratestackError {
15    CratestackError::Unauthorized("credential expired".to_owned())
16}
17
18/// `items`, ended with `unauthenticated` at `expires_at` (seconds since the
19/// epoch, the `exp` claim). Already expired: the first item is that error.
20pub fn until_credential_expires(
21    items: SubscriptionStream<serde_json::Value>,
22    expires_at: Option<i64>,
23) -> SubscriptionStream<serde_json::Value> {
24    let Some(expires_at) = expires_at else {
25        return items;
26    };
27    let now = SystemTime::now()
28        .duration_since(UNIX_EPOCH)
29        .map(|elapsed| elapsed.as_secs() as i64)
30        .unwrap_or(0);
31    let remaining = Duration::from_secs(expires_at.saturating_sub(now).max(0) as u64);
32    let deadline = tokio::time::Instant::now() + remaining;
33    let state = (items, Box::pin(tokio::time::sleep_until(deadline)), false);
34    Box::pin(stream::unfold(
35        state,
36        |(mut items, mut deadline, ended)| async move {
37            if ended {
38                return None;
39            }
40            match select(items.next(), deadline.as_mut()).await {
41                Either::Left((Some(item), _)) => {
42                    let ended = item.is_err();
43                    Some((item, (items, deadline, ended)))
44                }
45                Either::Left((None, _)) => None,
46                Either::Right(((), _)) => Some((Err(expired()), (items, deadline, true))),
47            }
48        },
49    ))
50}
51
52#[cfg(test)]
53mod tests;
cratestack_axum/rpc/credential_expiry.rs

cratestack_axum/rpc/
credential_expiry.rs
