feat(cat-gateway): Signed docs insert query

catalyst-gateway/Earthfile

@@ -124,3 +124,19 @@ check-builder-src-cache:
     RUN diff ../src_fingerprint.txt ../src_fingerprint_uncached.txt \
         || (echo "ERROR: Source fingerprints do not match. Caching Error Detected!!" && exit 1) \
         && echo "Source fingerprints match. Caching OK."
+
+test:
+    FROM +builder-src
+
+    COPY docker-compose.yml .
+
+    ENV EVENT_DB_URL "postgres://catalyst-event-dev:CHANGE_ME@localhost/CatalystEventDev"
+
+    WITH DOCKER \
+        --compose "./docker-compose.yml" \
+        --load ./event-db+build \
+        --pull alpine:3.20.3 \
+        --service event-db-is-running
+        RUN --mount=$EARTHLY_RUST_CARGO_HOME_CACHE --mount=$EARTHLY_RUST_TARGET_CACHE \
+            cargo nextest run --release --run-ignored=only signed_docs
+    END 

catalyst-gateway/bin/Cargo.toml

@@ -43,6 +43,7 @@ tokio-postgres = { version = "0.7.12", features = [
     "with-chrono-0_4",
     "with-serde_json-1",
     "with-time-0_3",
+    "with-uuid-1"
 ] }
 tokio = { version = "1.41.0", features = ["rt", "macros", "rt-multi-thread"] }
 dotenvy = "0.15.7"
@@ -77,7 +78,7 @@ poem-openapi = { version = "5.1.2", features = [
     "url",
     "chrono",
 ] }
-uuid = { version = "1.11.0", features = ["v4", "serde"] }
+uuid = { version = "1.11.0", features = ["v4", "v7", "serde"] }
 ulid = { version = "1.1.3", features = ["serde", "uuid"] }
 blake2b_simd = "1.0.2"
 url = "2.5.3"

catalyst-gateway/bin/src/db/event/mod.rs

@@ -18,6 +18,7 @@ pub(crate) mod config;
 pub(crate) mod error;
 pub(crate) mod legacy;
 pub(crate) mod schema_check;
+pub(crate) mod signed_docs;
 
 /// Database version this crate matches.
 /// Must equal the last Migrations Version Number from `event-db/migrations`.
@@ -212,7 +213,7 @@ impl EventDB {
 ///
 /// The env var "`DATABASE_URL`" can be set directly as an anv var, or in a
 /// `.env` file.
-pub(crate) fn establish_connection() {
+pub fn establish_connection() {
     let (url, user, pass) = Settings::event_db_settings();
 
     // This was pre-validated and can't fail, but provide default in the impossible case it

catalyst-gateway/bin/src/db/event/signed_docs/mod.rs

@@ -0,0 +1,27 @@
+//! Signed docs queries
+
+#[cfg(test)]
+mod tests;
+
+use super::EventDB;
+
+/// Upsert sql query
+const UPSERT_SIGNED_DOCS: &str = include_str!("./sql/upsert_signed_documents.sql");
+
+/// Make an upsert query into the `event-db` by adding data into the `signed_docs` table
+///
+/// * IF the record primary key (id,ver) does not exist, then add the new record. Return
+///   success.
+/// * IF the record does exist, but all values are the same as stored, return Success.
+/// * Otherwise return an error. (Can not over-write an existing record with new data).
+#[allow(dead_code)]
+pub(crate) async fn upsert_signed_docs(
+    id: &uuid::Uuid, ver: &uuid::Uuid, doc_type: &uuid::Uuid, author: &String,
+    metadata: &serde_json::Value, payload: &serde_json::Value, raw: &Vec<u8>,
+) -> anyhow::Result<()> {
+    EventDB::modify(UPSERT_SIGNED_DOCS, &[
+        id, ver, doc_type, author, metadata, payload, raw,
+    ])
+    .await?;
+    Ok(())
+}

catalyst-gateway/bin/src/db/event/signed_docs/sql/upsert_signed_documents.sql

@@ -0,0 +1,22 @@
+INSERT INTO signed_docs
+(
+  id,
+  ver,
+  type,
+  author,
+  metadata,
+  payload,
+  raw
+)
+VALUES
+($1, $2, $3, $4, $5, $6, $7)
+ON CONFLICT (id, ver) DO UPDATE
+SET
+type = signed_docs.type
+
+WHERE
+signed_docs.type = excluded.type
+AND signed_docs.author = excluded.author
+AND signed_docs.metadata = excluded.metadata
+AND signed_docs.payload = excluded.payload
+AND signed_docs.raw = excluded.raw

catalyst-gateway/bin/src/db/event/signed_docs/tests/mod.rs

@@ -0,0 +1,41 @@
+//! Integration tests of the `signed docs` queries
+
+use super::*;
+use crate::db::event::establish_connection;
+
+#[ignore = "An integration test which requires a running EventDB instance, disabled from `testunit` CI run"]
+#[tokio::test]
+async fn some_test() {
+    establish_connection();
+
+    let docs = [
+        (
+            uuid::Uuid::now_v7(),
+            uuid::Uuid::now_v7(),
+            uuid::Uuid::new_v4(),
+            "Alex".to_string(),
+            serde_json::Value::Null,
+            serde_json::Value::Null,
+            vec![1, 2, 3, 4],
+        ),
+        (
+            uuid::Uuid::now_v7(),
+            uuid::Uuid::now_v7(),
+            uuid::Uuid::new_v4(),
+            "Steven".to_string(),
+            serde_json::Value::Null,
+            serde_json::Value::Null,
+            vec![5, 6, 7, 8],
+        ),
+    ];
+
+    for (id, ver, doc_type, author, metadata, payload, raw) in &docs {
+        upsert_signed_docs(id, ver, doc_type, author, metadata, payload, raw)
+            .await
+            .unwrap();
+        // try to insert the same data again
+        upsert_signed_docs(id, ver, doc_type, author, metadata, payload, raw)
+            .await
+            .unwrap();
+    }
+}

catalyst-gateway/blueprint.cue

@@ -12,4 +12,9 @@ project: {
 			}
 		}
 	}
+	ci: {
+    targets: {
+      test: privileged: true
+    }
+  }
 }

catalyst-gateway/docker-compose.yml

@@ -1,10 +1,7 @@
-version: "3"
-
 services:
   event-db:
     image: event-db:latest
     environment:
-      # Required environment variables for migrations
       - DB_HOST=localhost
       - DB_PORT=5432
       - DB_NAME=CatalystEventDev
@@ -16,7 +13,6 @@ services:
 
       - INIT_AND_DROP_DB=true
       - WITH_MIGRATIONS=true
-      - WITH_SEED_DATA=true
     ports:
       - 5432:5432
     healthcheck:
@@ -25,6 +21,14 @@ services:
       timeout: 5s
       retries: 10
 
+# it is a helper service to wait until the event-db will be ready
+# mainly its a trick for Earthly how to wait until service will be fully functional
+  event-db-is-running:
+    image: alpine:3.20.3
+    depends_on:
+      event-db:
+        condition: service_healthy
+
   cat-gateway:
     image: cat-gateway:latest
     environment: