feat: use tower service as impl of authenticate layer

Author: ymgyt

Cargo.lock

@@ -13,7 +13,7 @@ anyhow = { version = "1" }
 clap = { version = "4.4" }
 chrono = { version = "0.4.31" }
 feed-rs = { version = "1.4" }
-futures = { version = "0.3" }
+futures-util = { version = "0.3.30" }
 graphql_client = { version = "0.13.0", default-features = false }
 moka = { version = "0.12.4", features = ["future"] }
 reqwest = { version = "0.11.23", default-features = false, features = [

Cargo.toml

@@ -12,7 +12,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 chrono = { workspace = true }
 feed-rs = { workspace = true }
-futures = { workspace = true }
+futures-util = { workspace = true }
 moka = { workspace = true, features = ["future"] }
 reqwest = { workspace = true, features = ["stream"] }
 thiserror = "1.0.56"

synd/Cargo.toml

@@ -49,7 +49,7 @@ pub struct FeedService {
 #[async_trait]
 impl FetchFeed for FeedService {
     async fn fetch_feed(&self, url: String) -> ParseResult<Feed> {
-        use futures::StreamExt;
+        use futures_util::StreamExt;
         let mut stream = self
             .http
             .get(&url)

synd/src/feed/parser.rs

@@ -16,8 +16,10 @@ axum = { version = "0.7.4" }
 clap = { workspace = true, features = ["derive"] }
 chrono = { workspace = true }
 feed-rs = { workspace = true }
+futures-util = { workspace = true }
 graphql_client = { workspace = true }
 kvsd = { version = "0.1.2" }
+moka = { workspace = true, features = ["future"]}
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = "1.0.111"

syndapi/Cargo.toml

@@ -1,9 +1,6 @@
-use axum::{
-    extract::{Request, State},
-    http::{self, StatusCode},
-    middleware::Next,
-    response::Response,
-};
+use std::time::Duration;
+
+use moka::future::Cache;
 use tracing::warn;
 
 use crate::{
@@ -14,60 +11,48 @@ use crate::{
 #[derive(Clone)]
 pub struct Authenticator {
     github: GithubClient,
+    cache: Cache<String, Principal>,
 }
 
 impl Authenticator {
     pub fn new() -> anyhow::Result<Self> {
+        let cache = Cache::builder()
+            .max_capacity(1024 * 1024)
+            .time_to_live(Duration::from_secs(60 * 60))
+            .build();
+
         Ok(Self {
             github: GithubClient::new()?,
+            cache,
         })
     }
 
     /// Authenticate from given token
-    async fn authenticate(&self, token: &str) -> Result<Principal, ()> {
+    pub async fn authenticate(&self, token: impl AsRef<str>) -> Result<Principal, ()> {
+        let token = token.as_ref();
         let mut split = token.splitn(2, ' ');
         match (split.next(), split.next()) {
             (Some("github"), Some(access_token)) => {
-                // TODO: configure cache to reduce api call
+                if let Some(principal) = self.cache.get(token).await {
+                    tracing::info!("Principal cache hit");
+                    return Ok(principal);
+                }
 
                 match self.github.authenticate(access_token).await {
-                    Ok(email) => Ok(Principal::User(User::from_email(email))),
+                    Ok(email) => {
+                        let principal = Principal::User(User::from_email(email));
+
+                        self.cache.insert(token.to_owned(), principal.clone()).await;
+
+                        Ok(principal)
+                    }
                     Err(err) => {
                         warn!("Failed to authenticate github {err}");
                         Err(())
                     }
                 }
             }
-            // TODO: remove
-            (Some("me"), None) => Ok(Principal::User(User::from_email("me@ymgyt.io"))),
             _ => Err(()),
         }
     }
 }
-
-/// Check authorization header and inject Authentication
-pub async fn authenticate(
-    State(authenticator): State<Authenticator>,
-    mut req: Request,
-    next: Next,
-) -> Result<Response, StatusCode> {
-    let header = req
-        .headers()
-        .get(http::header::AUTHORIZATION)
-        .and_then(|header| header.to_str().ok());
-
-    let Some(token) = header else {
-        return Err(StatusCode::UNAUTHORIZED);
-    };
-    let principal = match authenticator.authenticate(token).await {
-        Ok(principal) => principal,
-        Err(_) => {
-            warn!("Invalid token");
-            return Err(StatusCode::UNAUTHORIZED);
-        }
-    };
-
-    req.extensions_mut().insert(principal);
-
-    Ok(next.run(req).await)
-}

syndapi/src/serve/auth.rs

@@ -0,0 +1,86 @@
+use std::task::{Context, Poll};
+
+use axum::http::{self, StatusCode};
+use axum::response::IntoResponse;
+use futures_util::future::BoxFuture;
+use tower::{Layer, Service};
+
+use crate::serve::auth::Authenticator;
+
+#[derive(Clone)]
+pub struct AuthenticateLayer {
+    authenticator: Authenticator,
+}
+
+impl AuthenticateLayer {
+    pub fn new(authenticator: Authenticator) -> Self {
+        Self { authenticator }
+    }
+}
+
+impl<S> Layer<S> for AuthenticateLayer {
+    type Service = AuthenticateService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        AuthenticateService {
+            inner,
+            authenticator: self.authenticator.clone(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct AuthenticateService<S> {
+    authenticator: Authenticator,
+    inner: S,
+}
+
+impl<S> Service<axum::extract::Request> for AuthenticateService<S>
+where
+    S: Service<axum::extract::Request, Response = axum::response::Response>
+        + Send
+        + 'static
+        + Clone,
+    S::Future: Send + 'static,
+{
+    type Response = S::Response;
+
+    type Error = S::Error;
+
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut request: axum::extract::Request) -> Self::Future {
+        let header = request
+            .headers()
+            .get(http::header::AUTHORIZATION)
+            .and_then(|header| header.to_str().ok());
+
+        let Some(token) = header else {
+            return Box::pin(async { Ok(StatusCode::UNAUTHORIZED.into_response()) });
+        };
+
+        let Self {
+            authenticator,
+            mut inner,
+        } = self.clone();
+        let token = token.to_owned();
+
+        Box::pin(async move {
+            let principal = match authenticator.authenticate(token).await {
+                Ok(principal) => principal,
+                Err(_) => {
+                    tracing::warn!("Invalid token");
+                    return Ok(StatusCode::UNAUTHORIZED.into_response());
+                }
+            };
+
+            request.extensions_mut().insert(principal);
+
+            inner.call(request).await
+        })
+    }
+}

syndapi/src/serve/layer/authenticate.rs

@@ -1,2 +1,3 @@
 pub mod audit;
+pub mod authenticate;
 pub mod trace;

syndapi/src/serve/layer/mod.rs

@@ -4,7 +4,6 @@ use async_graphql::{extensions::Tracing, EmptySubscription, Schema};
 use axum::{
     error_handling::HandleErrorLayer,
     http::{header::AUTHORIZATION, StatusCode},
-    middleware,
     routing::{get, post},
     BoxError, Extension, Router,
 };
@@ -19,7 +18,7 @@ use crate::{
     config,
     dependency::Dependency,
     gql::{self, Mutation, Query},
-    serve::layer::trace,
+    serve::layer::{authenticate, trace},
 };
 
 pub mod auth;
@@ -53,11 +52,8 @@ pub async fn serve(listener: TcpListener, dep: Dependency) -> anyhow::Result<()>
     let service = Router::new()
         .route("/graphql", post(gql::handler::graphql))
         .layer(Extension(schema))
-        .route_layer(middleware::from_fn_with_state(
-            authenticator,
-            auth::authenticate,
-        ))
         .route("/graphql", get(gql::handler::graphiql))
+        .layer(authenticate::AuthenticateLayer::new(authenticator))
         .layer(
             ServiceBuilder::new()
                 .layer(SetSensitiveHeadersLayer::new(std::iter::once(