diff --git a/Cargo.lock b/Cargo.lock index 540ea4f76f..0f7b341ab1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2765,6 +2765,8 @@ dependencies = [ "bytes", "fallible-iterator", "postgres-protocol", + "serde", + "serde_json", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index 6d4d4e2200..607377de37 100644 --- a/Cargo.toml +++ b/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" diff --git a/etc/migrations/V1__initial.sql b/etc/migrations/V1__initial.sql new file mode 100644 index 0000000000..8d8b097679 --- /dev/null +++ b/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) +) \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index aac9c860af..17edb0f014 100644 --- a/src/main.rs +++ b/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(()) -} diff --git a/src/postgres.rs b/src/postgres.rs new file mode 100644 index 0000000000..408e389765 --- /dev/null +++ b/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>; + +pub async fn pool() -> anyhow::Result { + 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 for ConnectionCustomizer { + fn on_acquire<'a>( + &'a self, + client: &'a mut pg::Client, + ) -> Pin> + 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> { + 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> { + 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( + 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( + 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(()) +}