database support

Signed-off-by: Alexey Aristov <aav@acm.org>

Cargo.lock

@@ -2765,6 +2765,8 @@ dependencies = [
  "bytes",
  "fallible-iterator",
  "postgres-protocol",
+ "serde",
+ "serde_json",
  "uuid",
 ]
 

Cargo.toml

@@ -13,7 +13,10 @@ serde = "1.0.219"
 actix-web = "4.11.0"
 actix-cors = "0.7.1"
 refinery = { version = "0.8.16", features = ["tokio-postgres"] }
-tokio-postgres = "0.7.13"
+tokio-postgres = { version = "0.7.13", features = [
+    "with-uuid-1",
+    "with-serde_json-1",
+] }
 bb8 = "0.9.0"
 bb8-postgres = { version = "0.9.0", features = ["with-uuid-1"] }
 md5 = "0.8.0"

etc/migrations/V1__initial.sql

@@ -0,0 +1,17 @@
+create table blob(
+    key text not null, 
+    hash text not null
+);
+
+create unique index blob_key on blob(key);
+create unique index blob_hash on blob(hash);
+
+
+create table object(
+    workspace uuid not null,
+    key text not null,
+    part int not null,
+    data jsonb not null,
+
+    primary key (workspace, key, part)
+)

src/main.rs

@@ -6,7 +6,7 @@ use actix_web::{
     body::MessageBody,
     dev::{ServiceRequest, ServiceResponse},
     middleware::{Next, from_fn},
-    web::{Data, Path, delete, get, post, put, scope},
+    web::{Data, Path, get, scope},
 };
 use tracing::*;
 use tracing_actix_web::TracingLogger;
@@ -15,9 +15,7 @@ use uuid::Uuid;
 use hulyrs::services::jwt::actix::ServiceRequestExt;
 
 mod config;
-mod handlers;
 mod postgres;
-mod s3;
 
 use config::CONFIG;
 
@@ -45,7 +43,6 @@ async fn main() -> anyhow::Result<()> {
         env!("CARGO_PKG_VERSION")
     );
 
-    let s3 = s3::client().await;
     let postgres = postgres::pool().await?;
 
     let bind_to = SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);
@@ -84,20 +81,17 @@ async fn main() -> anyhow::Result<()> {
             .supports_credentials()
             .max_age(3600);
 
-        const KEY_PATH: &str = "/{key:.*}";
+        //const KEY_PATH: &str = "/{key:.*}";
 
         App::new()
             .app_data(Data::new(postgres.clone()))
-            .app_data(Data::new(s3.clone()))
             .wrap(TracingLogger::default())
             .wrap(cors)
             .service(
-                scope("/api/{workspace}")
-                    .wrap(from_fn(auth))
-                    .route(KEY_PATH, get().to(handlers::get))
-                    .route(KEY_PATH, put().to(handlers::put))
-                    .route(KEY_PATH, post().to(handlers::post))
-                    .route(KEY_PATH, delete().to(handlers::delete)),
+                scope("/api/{workspace}").wrap(from_fn(auth)), //.route(KEY_PATH, get().to(handlers::get))
+                                                               //.route(KEY_PATH, put().to(handlers::put))
+                                                               //.route(KEY_PATH, post().to(handlers::post))
+                                                               //.route(KEY_PATH, delete().to(handlers::delete)),
             )
             .route("/status", get().to(async || "ok"))
     })
@@ -110,37 +104,3 @@ async fn main() -> anyhow::Result<()> {
 
     Ok(())
 }
-
-#[tokio::main]
-async fn main_() -> anyhow::Result<()> {
-    use crate::{postgres::Pool, s3};
-    use aws_sdk_s3::{presigning::PresigningConfig, primitives::ByteStream};
-
-    initialize_tracing();
-
-    let expires_in: std::time::Duration = std::time::Duration::from_secs(600);
-    let expires_in: aws_sdk_s3::presigning::PresigningConfig =
-        PresigningConfig::expires_in(expires_in).unwrap();
-
-    let s3 = s3::client().await;
-
-    let presigned_request = s3
-        .put_object()
-        .set_bucket(Some("hulylake".into()))
-        .set_key(Some("myobject".into()))
-        .presigned(expires_in)
-        .await
-        .unwrap();
-
-    let url = presigned_request.uri();
-
-    debug!(?url, "presigned request");
-
-    let client = reqwest::Client::new();
-    let res = client.put(url).body("hello world").send().await.unwrap();
-
-    debug!(?res, "response");
-    debug!("body: {:?}", res.text().await.unwrap());
-
-    Ok(())
-}

src/postgres.rs

@@ -0,0 +1,184 @@
+use std::pin::Pin;
+
+use bb8_postgres::PostgresConnectionManager;
+use tokio_postgres::NoTls;
+use tokio_postgres::{self as pg};
+use tracing::*;
+
+use crate::config::CONFIG;
+
+pub type Pool = bb8::Pool<PostgresConnectionManager<NoTls>>;
+
+pub async fn pool() -> anyhow::Result<Pool> {
+    tracing::debug!(
+        connection = CONFIG.db_connection,
+        "database connection string"
+    );
+
+    let manager = bb8_postgres::PostgresConnectionManager::new_from_stringlike(
+        &CONFIG.db_connection,
+        tokio_postgres::NoTls,
+    )?;
+
+    #[derive(Debug)]
+    struct ConnectionCustomizer;
+
+    impl bb8::CustomizeConnection<pg::Client, pg::Error> for ConnectionCustomizer {
+        fn on_acquire<'a>(
+            &'a self,
+            client: &'a mut pg::Client,
+        ) -> Pin<Box<dyn Future<Output = Result<(), pg::Error>> + Send + 'a>> {
+            Box::pin(async {
+                client
+                    .execute("set search_path to $1", &[&CONFIG.db_scheme])
+                    .await
+                    .unwrap();
+                Ok(())
+            })
+        }
+    }
+
+    let pool = bb8::Pool::builder()
+        .max_size(15)
+        .connection_customizer(Box::new(ConnectionCustomizer))
+        .build(manager)
+        .await?;
+
+    {
+        let mut connection = pool.dedicated_connection().await?;
+
+        // query params cannot be bound in ddl statements
+        connection
+            .execute(
+                &format!("create schema if not exists {}", CONFIG.db_scheme),
+                &[],
+            )
+            .await?;
+
+        refinery::embed_migrations!("etc/migrations");
+
+        let report = migrations::runner()
+            .set_migration_table_name("migrations")
+            .run_async(&mut connection)
+            .await?;
+
+        for m in report.applied_migrations().iter() {
+            info!(migration = m.to_string(), "apply migration");
+        }
+    }
+
+    Ok(pool)
+}
+
+#[instrument(level = "debug", skip_all)]
+pub async fn find_blob_by_hash(pool: &Pool, hash: &str) -> anyhow::Result<Option<String>> {
+    let connection = pool.get().await?;
+
+    let blob = connection
+        .query("select key from blob where hash = $1", &[&hash])
+        .await?;
+
+    Ok(match blob.as_slice() {
+        [found] => Some(found.get::<_, String>("key")),
+        [] => None,
+
+        _ => panic!(),
+    })
+}
+
+#[instrument(level = "debug", skip_all)]
+pub async fn insert_blob(pool: &Pool, key: &str, hash: &str) -> anyhow::Result<()> {
+    let connection = pool.get().await?;
+
+    connection
+        .execute(
+            "insert into blob (key, hash) values ($1, $2)",
+            &[&key, &hash],
+        )
+        .await?;
+
+    Ok(())
+}
+
+pub struct Object {
+    part: u32,
+    data: serde_json::Value,
+}
+
+pub async fn find_parts(
+    pool: &Pool,
+    workspace: uuid::Uuid,
+    key: &str,
+) -> anyhow::Result<Vec<Object>> {
+    let connection = pool.get().await?;
+
+    let parts = connection
+        .query(
+            "select part, data from object where workspace = $1 and key = $1 order by part",
+            &[&workspace, &key],
+        )
+        .await?;
+
+    let parts = parts
+        .into_iter()
+        .map(|row| {
+            let part = row.get::<_, u32>("part");
+            let data = row.get::<_, serde_json::Value>("data");
+            Object { part, data }
+        })
+        .collect();
+
+    Ok(parts)
+}
+
+pub async fn insert_part<D: serde::Serialize>(
+    pool: &Pool,
+    workspace: uuid::Uuid,
+    key: &str,
+    part: u32,
+    data: D,
+) -> anyhow::Result<()> {
+    let connection = pool.get().await?;
+
+    let data = serde_json::to_value(data)?;
+
+    connection
+        .execute(
+            "insert into object (workspace, key, part, data) values ($1, $2, $3, $4)",
+            &[&workspace, &key, &part, &data],
+        )
+        .await?;
+
+    Ok(())
+}
+
+pub async fn shrink<D: serde::Serialize>(
+    pool: &Pool,
+    workspace: uuid::Uuid,
+    key: &str,
+    data: D,
+) -> anyhow::Result<()> {
+    let mut connection = pool.get().await?;
+
+    let transaction = connection.transaction().await?;
+
+    transaction
+        .execute(
+            "delete from object where workspace = $1 and key = $2 and part > 0",
+            &[&workspace, &key],
+        )
+        .await?;
+
+    let data = serde_json::to_value(data)?;
+
+    transaction
+        .execute(
+            "update object set data=$1 where workspace = $2 and key = $3 and part = 0",
+            &[&data, &workspace, &key],
+        )
+        .await?;
+
+    transaction.commit().await?;
+
+    Ok(())
+}