feat: opendal on postgres output type, bump opendal

common/Cargo.toml

@@ -77,3 +77,6 @@ tracing = ["dep:tracing"]
 
 [dev-dependencies]
 proptest = "1.1.0"
+
+[package.metadata.docs.rs]
+all-features = true

common/src/deployment.rs

@@ -119,7 +119,7 @@ mod tests {
             Environment::Deployment,
             Environment::from_str("production").unwrap()
         );
-        assert!(State::from_str("somewhere_else").is_err());
+        assert!(Environment::from_str("somewhere_else").is_err());
         assert_eq!(format!("{:?}", Environment::Local), "Local".to_owned());
         assert_eq!(format!("{}", Environment::Local), "local".to_owned());
         assert_eq!(Environment::Local.to_string(), "local".to_owned());

resources/opendal/Cargo.toml

@@ -9,7 +9,7 @@ keywords = ["shuttle-service", "opendal"]
 
 [dependencies]
 async-trait = "0.1.56"
-opendal = "0.45"
+opendal = "0.50"
 serde = { version = "1", features = ["derive"] }
 shuttle-service = { path = "../../service", version = "0.49.0" }
 

resources/opendal/src/lib.rs

@@ -66,7 +66,7 @@ impl IntoResource<Operator> for OpendalOutput {
     async fn into_resource(self) -> Result<Operator, shuttle_service::Error> {
         let scheme = Scheme::from_str(&self.scheme).map_err(Error)?;
 
-        Ok(Operator::via_map(scheme, self.cfg).map_err(Error)?)
+        Ok(Operator::via_iter(scheme, self.cfg).map_err(Error)?)
     }
 }
 

resources/shared-db/Cargo.toml

@@ -11,6 +11,7 @@ keywords = ["shuttle-service", "database"]
 async-trait = "0.1.56"
 diesel-async = { version = "0.4.1", optional = true }
 mongodb = { version = "2.3.0", optional = true }
+opendal = { version = "0.50", optional = true }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 shuttle-service = { path = "../../service", version = "0.49.0" }
@@ -31,3 +32,6 @@ diesel-async-deadpool = ["diesel-async", "diesel-async/deadpool"]
 # Postgres with an sqlx PgPool
 sqlx = ["dep:sqlx", "sqlx/runtime-tokio", "sqlx/tls-rustls"]
 sqlx-native-tls = ["dep:sqlx", "sqlx/runtime-tokio", "sqlx/tls-native-tls"]
+
+# OpenDAL backed by Postgres
+opendal = ["dep:opendal", "opendal/services-postgresql", "postgres", "sqlx"]

resources/shared-db/src/postgres.rs

@@ -96,7 +96,8 @@ impl IntoResource<diesel_async::AsyncPgConnection> for OutputWrapper {
     async fn into_resource(self) -> Result<diesel_async::AsyncPgConnection, Error> {
         use diesel_async::{AsyncConnection, AsyncPgConnection};
 
-        let connection_string: String = self.into_resource().await.unwrap();
+        let connection_string: String = self.into_resource().await?;
+
         Ok(AsyncPgConnection::establish(&connection_string)
             .await
             .map_err(shuttle_service::error::CustomError::new)?)
@@ -109,7 +110,7 @@ impl IntoResource<diesel_bb8::Pool<diesel_async::AsyncPgConnection>> for OutputW
     async fn into_resource(
         self,
     ) -> Result<diesel_bb8::Pool<diesel_async::AsyncPgConnection>, Error> {
-        let connection_string: String = self.into_resource().await.unwrap();
+        let connection_string: String = self.into_resource().await?;
 
         Ok(diesel_bb8::Pool::builder()
             .min_idle(Some(MIN_CONNECTIONS))
@@ -126,7 +127,7 @@ impl IntoResource<diesel_deadpool::Pool<diesel_async::AsyncPgConnection>> for Ou
     async fn into_resource(
         self,
     ) -> Result<diesel_deadpool::Pool<diesel_async::AsyncPgConnection>, Error> {
-        let connection_string: String = self.into_resource().await.unwrap();
+        let connection_string: String = self.into_resource().await?;
 
         Ok(
             diesel_deadpool::Pool::builder(AsyncDieselConnectionManager::new(connection_string))
@@ -141,7 +142,7 @@ impl IntoResource<diesel_deadpool::Pool<diesel_async::AsyncPgConnection>> for Ou
 #[async_trait]
 impl IntoResource<sqlx::PgPool> for OutputWrapper {
     async fn into_resource(self) -> Result<sqlx::PgPool, Error> {
-        let connection_string: String = self.into_resource().await.unwrap();
+        let connection_string: String = self.into_resource().await?;
 
         Ok(sqlx::postgres::PgPoolOptions::new()
             .min_connections(MIN_CONNECTIONS)
@@ -151,3 +152,37 @@ impl IntoResource<sqlx::PgPool> for OutputWrapper {
             .map_err(shuttle_service::error::CustomError::new)?)
     }
 }
+
+#[cfg(feature = "opendal")]
+#[async_trait]
+impl IntoResource<opendal::Operator> for OutputWrapper {
+    async fn into_resource(self) -> Result<opendal::Operator, Error> {
+        let connection_string: String = self.into_resource().await?;
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .min_connections(MIN_CONNECTIONS)
+            .max_connections(MAX_CONNECTIONS)
+            .connect(&connection_string)
+            .await
+            .map_err(shuttle_service::error::CustomError::new)?;
+        sqlx::query(
+            "CREATE TABLE IF NOT EXISTS opendal (key TEXT PRIMARY KEY, value BYTEA NOT NULL)",
+        )
+        .execute(&pool)
+        .await
+        .map_err(shuttle_service::error::CustomError::new)?;
+
+        let config = opendal::services::Postgresql::default()
+            .root("/")
+            .connection_string(&connection_string)
+            .table("opendal")
+            // key field type in the table should be compatible with Rust's &str like text
+            .key_field("key")
+            // value field type in the table should be compatible with Rust's Vec<u8> like bytea
+            .value_field("value");
+        let op = opendal::Operator::new(config)
+            .map_err(shuttle_service::error::CustomError::new)?
+            .finish();
+
+        Ok(op)
+    }
+}