ENG-3200: Centralize OAuth logic to a single repository (#7)

Author: sagoez

Cargo.lock

@@ -13,6 +13,7 @@ actix-governor = "0.5.0"
 actix-web = "4.5.1"
 actix-web-lab = "0.20.2"
 anyhow = "1.0.79"
+async-trait = "0.1.80"
 chrono = { version = "0.4.33", features = ["serde"] }
 dotenvy = "0.15.7"
 envconfig = "0.10.0"

Cargo.toml

@@ -1,48 +1,11 @@
 mod parameter;
 mod refresh;
+mod storage;
 mod token;
 mod trigger;
 
 pub use parameter::*;
 pub use refresh::*;
+pub use storage::*;
 pub use token::*;
 pub use trigger::*;
-
-use chrono::{DateTime, Utc};
-use integrationos_domain::{
-    algebra::{MongoStore, StoreExt},
-    Connection, Id, IntegrationOSError,
-};
-use mongodb::bson::doc;
-
-pub async fn get_connections_to_refresh(
-    collection: &MongoStore<Connection>,
-    refresh_before: &DateTime<Utc>,
-    refresh_after: &DateTime<Utc>,
-) -> Result<Vec<Connection>, IntegrationOSError> {
-    collection
-        .get_many(
-            Some(doc! {
-                "oauth.enabled.expires_at": doc! {
-                    "$gt": refresh_before.timestamp(),
-                    "$lte": refresh_after.timestamp(),
-                },
-            }),
-            None,
-            None,
-            None,
-            None,
-        )
-        .await
-}
-
-pub async fn get_connection_to_trigger(
-    collection: &MongoStore<Connection>,
-    id: Id,
-) -> Result<Option<Connection>, IntegrationOSError> {
-    collection
-        .get_one(doc! {
-            "_id": id.to_string(),
-        })
-        .await
-}

src/algebra/mod.rs

@@ -13,13 +13,13 @@ use std::{
 };
 use tracing::warn;
 
-pub trait Parameter {
+pub trait ParameterExt {
     fn headers(&self, computation: Option<&Computation>) -> Result<Option<HeaderMap>, Error>;
     fn body(&self, secret: &OAuthSecret) -> Result<Option<Value>, Error>;
     fn query(&self, computation: Option<&Computation>) -> Result<Option<Value>, Error>;
 }
 
-impl Parameter for ConnectionOAuthDefinition {
+impl ParameterExt for ConnectionOAuthDefinition {
     fn headers(&self, computation: Option<&Computation>) -> Result<Option<HeaderMap>, Error> {
         headers(self, computation)
     }

src/algebra/parameter.rs

@@ -1,5 +1,6 @@
-use crate::prelude::{
-    get_connections_to_refresh, Query, Refresh, StatefulActor, Trigger, TriggerActor, Unit,
+use crate::{
+    algebra::{StorageExt, TriggerActor},
+    domain::{Query, Refresh, StatefulActor, Trigger, Unit},
 };
 use actix::prelude::*;
 use chrono::{Duration, Utc};
@@ -66,9 +67,9 @@ impl Handler<Refresh> for RefreshActor {
         let state = self.state.clone();
 
         Box::pin(async move {
-            let connections =
-                get_connections_to_refresh(&connections_store, &refresh_before, &refresh_after)
-                    .await?;
+            let connections = connections_store
+                .get_by(&refresh_before, &refresh_after)
+                .await?;
 
             tracing::info!("Found {} connections to refresh", connections.len());
 

src/algebra/refresh.rs

@@ -0,0 +1,45 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use integrationos_domain::{Connection, Id, IntegrationOSError, MongoStore, StoreExt};
+use mongodb::bson::doc;
+
+#[async_trait]
+pub trait StorageExt {
+    async fn get_by(
+        &self,
+        refresh_before: &DateTime<Utc>,
+        refresh_after: &DateTime<Utc>,
+    ) -> Result<Vec<Connection>, IntegrationOSError>;
+
+    async fn get(&self, id: Id) -> Result<Option<Connection>, IntegrationOSError>;
+}
+
+#[async_trait]
+impl StorageExt for MongoStore<Connection> {
+    async fn get_by(
+        &self,
+        refresh_before: &DateTime<Utc>,
+        refresh_after: &DateTime<Utc>,
+    ) -> Result<Vec<Connection>, IntegrationOSError> {
+        self.get_many(
+            Some(doc! {
+                "oauth.enabled.expires_at": doc! {
+                    "$gt": refresh_before.timestamp(),
+                    "$lte": refresh_after.timestamp(),
+                },
+            }),
+            None,
+            None,
+            None,
+            None,
+        )
+        .await
+    }
+
+    async fn get(&self, id: Id) -> Result<Option<Connection>, IntegrationOSError> {
+        self.get_one(doc! {
+            "_id": id.to_string(),
+        })
+        .await
+    }
+}

src/algebra/storage.rs

@@ -1,17 +1,17 @@
-use crate::prelude::Config;
+use crate::service::Configuration;
 use chrono::{Duration, Utc};
 use integrationos_domain::{Claims, IntegrationOSError as Error, InternalError};
 use jsonwebtoken::{encode, EncodingKey, Header};
 
-pub trait TokenGenerator {
-    fn generate(&self, configuration: Config, expiration: i64) -> Result<String, Error>;
+pub trait TokenExt {
+    fn generate(&self, configuration: Configuration, expiration: i64) -> Result<String, Error>;
 }
 
 #[derive(Debug, Default)]
-pub struct JwtTokenGenerator;
+pub struct Token;
 
-impl TokenGenerator for JwtTokenGenerator {
-    fn generate(&self, configuration: Config, expiration: i64) -> Result<String, Error> {
+impl TokenExt for Token {
+    fn generate(&self, configuration: Configuration, expiration: i64) -> Result<String, Error> {
         let key = configuration.server().admin_secret();
         let key = key.as_bytes();
         let key = EncodingKey::from_secret(key);

src/algebra/token.rs

@@ -1,5 +1,6 @@
-use super::Parameter;
-use crate::prelude::{Outcome, Trigger};
+use crate::domain::{Outcome, Trigger};
+
+use super::ParameterExt;
 use actix::prelude::*;
 use chrono::Duration;
 use integrationos_domain::{
@@ -256,8 +257,8 @@ impl Handler<Trigger> for TriggerActor {
                     Outcome::success(id.to_string().as_str(), json!({ "id": id.to_string() }))
                 }
                 Err(e) => Outcome::failure(
-                    msg.connection().id.to_string().as_str(),
-                    json!({ "error": e.to_string() }),
+                    e,
+                    json!({ "connectionId": msg.connection().id.to_string() }),
                 ),
             }
         };

src/algebra/trigger.rs

@@ -9,3 +9,9 @@ pub use query::*;
 pub use refresh::*;
 pub use state::*;
 pub use trigger::*;
+
+use futures::Future;
+use std::pin::Pin;
+
+pub type Unit = ();
+pub type Task = Pin<Box<dyn Future<Output = Unit> + Send + Sync>>;

src/domain/mod.rs

@@ -1,11 +1,19 @@
-use serde::{Deserialize, Serialize};
+use integrationos_domain::IntegrationOSError;
+use serde::Serialize;
 use serde_json::Value;
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
+#[serde(tag = "type")]
 pub enum Outcome {
-    Success { message: String, metadata: Value },
-    Failure { message: String, metadata: Value },
+    Success {
+        message: String,
+        metadata: Value,
+    },
+    Failure {
+        error: IntegrationOSError,
+        metadata: Value,
+    },
 }
 
 impl Outcome {
@@ -16,10 +24,7 @@ impl Outcome {
         }
     }
 
-    pub fn failure(message: &str, metadata: Value) -> Self {
-        Self::Failure {
-            message: message.to_string(),
-            metadata,
-        }
+    pub fn failure(error: IntegrationOSError, metadata: Value) -> Self {
+        Self::Failure { error, metadata }
     }
 }