feat: add crdb

Signed-off-by: kjuulh <contact@kjuulh.io>

crates/flux-releaser/Cargo.toml

@@ -24,6 +24,7 @@ walkdir = "2.4.0"
 tar = "0.4.40"
 tokio-stream = { version = "0.1.14", features = ["full"] }
 rand = "0.8.5"
+sqlx = { version = "0.7.3", features = ["postgres", "runtime-tokio"] }
 
 [build-dependencies]
 tonic-build = "0.11.0"

crates/flux-releaser/migrations/crdb/20240221054936_init.sql

@@ -0,0 +1 @@
+-- Add migration script here

crates/flux-releaser/src/app.rs

@@ -1,5 +1,7 @@
 use std::{ops::Deref, sync::Arc};
 
+use sqlx::{PgPool, Postgres};
+
 use self::infra::{
     aws_s3::s3_client,
     grpc::{new_client, FluxReleaserGrpcClient},
@@ -25,6 +27,7 @@ impl SharedApp {
 pub struct App {
     pub s3_client: aws_sdk_s3::Client,
     pub nats: infra::nats::Nats,
+    pub database: PgPool,
 }
 
 impl App {
@@ -32,6 +35,7 @@ impl App {
         Ok(Self {
             s3_client: s3_client().await?,
             nats: infra::nats::Nats::new().await?,
+            database: infra::database::get_database().await?,
         })
     }
 }

crates/flux-releaser/src/app/infra.rs

@@ -1,23 +1,4 @@
 pub mod aws_s3;
+pub mod database;
+pub mod grpc;
 pub mod nats;
-pub mod grpc {
-    use std::sync::Arc;
-
-    use anyhow::Context;
-    use tokio::sync::Mutex;
-    use tonic::transport::Channel;
-
-    use crate::grpc::gen::flux_releaser_client::FluxReleaserClient;
-
-    pub type FluxReleaserGrpcClient = Arc<Mutex<FluxReleaserClient<Channel>>>;
-
-    pub async fn new_client(registry: impl Into<String>) -> anyhow::Result<FluxReleaserGrpcClient> {
-        let registry = registry.into();
-
-        let client = FluxReleaserClient::connect(registry)
-            .await
-            .context("failed to connect to flux_releaser registry")?;
-
-        Ok(Arc::new(Mutex::new(client)))
-    }
-}

crates/flux-releaser/src/app/infra/database.rs

@@ -0,0 +1,19 @@
+use anyhow::Context;
+use sqlx::{PgPool, Postgres};
+
+pub async fn get_database() -> anyhow::Result<PgPool> {
+    tracing::trace!("initializing database");
+    let db =
+        sqlx::PgPool::connect(&std::env::var("DATABASE_URL").context("DATABASE_URL is not set")?)
+            .await?;
+
+    tracing::trace!("migrating crdb");
+    sqlx::migrate!("migrations/crdb")
+        .set_locking(false)
+        .run(&db)
+        .await?;
+
+    let _ = sqlx::query("SELECT 1;").fetch_one(&db).await?;
+
+    Ok(db)
+}

crates/flux-releaser/src/app/infra/grpc.rs

@@ -0,0 +1,19 @@
+use std::sync::Arc;
+
+use anyhow::Context;
+use tokio::sync::Mutex;
+use tonic::transport::Channel;
+
+use crate::grpc::gen::flux_releaser_client::FluxReleaserClient;
+
+pub type FluxReleaserGrpcClient = Arc<Mutex<FluxReleaserClient<Channel>>>;
+
+pub async fn new_client(registry: impl Into<String>) -> anyhow::Result<FluxReleaserGrpcClient> {
+    let registry = registry.into();
+
+    let client = FluxReleaserClient::connect(registry)
+        .await
+        .context("failed to connect to flux_releaser registry")?;
+
+    Ok(Arc::new(Mutex::new(client)))
+}