Shut down the `/api/events` stream when the user logs out or their token expires.

When tokens are revoked (logout or expiry), the server now publishes an internal event via the new `logins` event broadcaster. These events are used to guard the `/api/events` stream. When a token revocation event arrives for the token used to subscribe to the stream, the stream is cut short, disconnecting the client.

In service of this, tokens now have IDs, which are non-confidential values that can be used to discuss tokens without their secrets being passed around unnecessarily. These IDs are not (at this time) exposed to clients, but they could be.

2 files changed, 59 insertions, 73 deletions

diff --git a/src/repo/login/extract.rs b/src/repo/login/extract.rs
index c127078..ab61106 100644
--- a/src/repo/login/extract.rs
+++ b/src/repo/login/extract.rs
@@ -1,67 +1,15 @@
-use axum::{
-    extract::{FromRequestParts, State},
-    http::{request::Parts, StatusCode},
-    response::{IntoResponse, Response},
-};
+use axum::{extract::FromRequestParts, http::request::Parts};
 
 use super::Login;
-use crate::{
-    app::App,
-    clock::RequestedAt,
-    error::Internal,
-    login::{app::ValidateError, extract::IdentityToken},
-};
+use crate::{app::App, login::extract::Identity};
 
 #[async_trait::async_trait]
 impl FromRequestParts<App> for Login {
-    type Rejection = LoginError<Internal>;
+    type Rejection = <Identity as FromRequestParts<App>>::Rejection;
 
     async fn from_request_parts(parts: &mut Parts, state: &App) -> Result<Self, Self::Rejection> {
-        // After Rust 1.82 (and #[feature(min_exhaustive_patterns)] lands on
-        // stable), the following can be replaced:
-        //
-        // ```
-        // let Ok(identity_token) = IdentityToken::from_request_parts(
-        //     parts,
-        //     state,
-        // ).await;
-        // ```
-        let identity_token = IdentityToken::from_request_parts(parts, state).await?;
-        let RequestedAt(used_at) = RequestedAt::from_request_parts(parts, state).await?;
+        let identity = Identity::from_request_parts(parts, state).await?;
 
-        let secret = identity_token.secret().ok_or(LoginError::Unauthorized)?;
-
-        let app = State::<App>::from_request_parts(parts, state).await?;
-        match app.logins().validate(&secret, &used_at).await {
-            Ok(login) => Ok(login),
-            Err(ValidateError::InvalidToken) => Err(LoginError::Unauthorized),
-            Err(other) => Err(other.into()),
-        }
-    }
-}
-
-pub enum LoginError<E> {
-    Failure(E),
-    Unauthorized,
-}
-
-impl<E> IntoResponse for LoginError<E>
-where
-    E: IntoResponse,
-{
-    fn into_response(self) -> Response {
-        match self {
-            Self::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized").into_response(),
-            Self::Failure(e) => e.into_response(),
-        }
-    }
-}
-
-impl<E> From<E> for LoginError<Internal>
-where
-    E: Into<Internal>,
-{
-    fn from(err: E) -> Self {
-        Self::Failure(err.into())
+        Ok(identity.login)
     }
 }
diff --git a/src/repo/token.rs b/src/repo/token.rs
index 15eef48..5f39e1d 100644
--- a/src/repo/token.rs
+++ b/src/repo/token.rs
@@ -1,8 +1,10 @@
+use std::fmt;
+
 use sqlx::{sqlite::Sqlite, SqliteConnection, Transaction};
 use uuid::Uuid;
 
 use super::login::{self, Login};
-use crate::{clock::DateTime, login::extract::IdentitySecret};
+use crate::{clock::DateTime, id::Id as BaseId, login::extract::IdentitySecret};
 
 pub trait Provider {
     fn tokens(&mut self) -> Tokens;
@@ -24,15 +26,17 @@ impl<'c> Tokens<'c> {
         login: &Login,
         issued_at: &DateTime,
     ) -> Result<IdentitySecret, sqlx::Error> {
+        let id = Id::generate();
         let secret = Uuid::new_v4().to_string();
 
         let secret = sqlx::query_scalar!(
             r#"
                 insert
-                into token (secret, login, issued_at, last_used_at)
-                values ($1, $2, $3, $3)
+                into token (id, secret, login, issued_at, last_used_at)
+                values ($1, $2, $3, $4, $4)
                 returning secret as "secret!: IdentitySecret"
             "#,
+            id,
             secret,
             login.id,
             issued_at,
@@ -44,37 +48,38 @@ impl<'c> Tokens<'c> {
     }
 
     // Revoke a token by its secret.
-    pub async fn revoke(&mut self, secret: &IdentitySecret) -> Result<(), sqlx::Error> {
-        sqlx::query!(
+    pub async fn revoke(&mut self, secret: &IdentitySecret) -> Result<Id, sqlx::Error> {
+        let token = sqlx::query_scalar!(
             r#"
                 delete
                 from token
                 where secret = $1
-                returning 1 as "found: u32"
+                returning id as "id: Id"
             "#,
             secret,
         )
         .fetch_one(&mut *self.0)
         .await?;
 
-        Ok(())
+        Ok(token)
     }
 
     // Expire and delete all tokens that haven't been used more recently than
     // `expire_at`.
-    pub async fn expire(&mut self, expire_at: &DateTime) -> Result<(), sqlx::Error> {
-        sqlx::query!(
+    pub async fn expire(&mut self, expire_at: &DateTime) -> Result<Vec<Id>, sqlx::Error> {
+        let tokens = sqlx::query_scalar!(
             r#"
                 delete
                 from token
                 where last_used_at < $1
+                returning id as "id: Id"
             "#,
             expire_at,
         )
-        .execute(&mut *self.0)
+        .fetch_all(&mut *self.0)
         .await?;
 
-        Ok(())
+        Ok(tokens)
     }
 
     // Validate a token by its secret, retrieving the associated Login record.
@@ -84,7 +89,7 @@ impl<'c> Tokens<'c> {
         &mut self,
         secret: &IdentitySecret,
         used_at: &DateTime,
-    ) -> Result<Login, sqlx::Error> {
+    ) -> Result<(Id, Login), sqlx::Error> {
         // I would use `update … returning` to do this in one query, but
         // sqlite3, as of this writing, does not allow an update's `returning`
         // clause to reference columns from tables joined into the update. Two
@@ -101,21 +106,54 @@ impl<'c> Tokens<'c> {
         .execute(&mut *self.0)
         .await?;
 
-        let login = sqlx::query_as!(
-            Login,
+        let login = sqlx::query!(
             r#"
                 select
-                    login.id as "id: login::Id",
-                    name
+                    token.id as "token_id: Id",
+                    login.id as "login_id: login::Id",
+                    name as "login_name"
                 from login
                 join token on login.id = token.login
                 where token.secret = $1
             "#,
             secret,
         )
+        .map(|row| {
+            (
+                row.token_id,
+                Login {
+                    id: row.login_id,
+                    name: row.login_name,
+                },
+            )
+        })
         .fetch_one(&mut *self.0)
         .await?;
 
         Ok(login)
     }
 }
+
+// Stable identifier for a token. Prefixed with `T`.
+#[derive(Clone, Debug, Eq, Hash, PartialEq, sqlx::Type, serde::Deserialize, serde::Serialize)]
+#[sqlx(transparent)]
+#[serde(transparent)]
+pub struct Id(BaseId);
+
+impl From<BaseId> for Id {
+    fn from(id: BaseId) -> Self {
+        Self(id)
+    }
+}
+
+impl Id {
+    pub fn generate() -> Self {
+        BaseId::generate("T")
+    }
+}
+
+impl fmt::Display for Id {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}