feat: add postgres support in addition to sqlite for recon

.github/workflows/rust-build-and-test.yml

@@ -53,6 +53,7 @@ jobs:
     steps:
     - uses: actions/checkout@v3
     - uses: Swatinem/rust-cache@v2
+    - uses: docker/setup-buildx-action@v2
       with:
         # The prefix cache key, this can be changed to start a new cache manually.
         # default: "v0-rust"
@@ -61,7 +62,15 @@ jobs:
         cache-targets: false
     - uses: mozilla-actions/[email protected]
     - name: Run tests
-      run: make test
+      run: TEST_DATABASE_URL="postgresql://postgres:c3ram1c@postgresci:5432/ceramic_one_tests" PG_TESTS=1 make test
+    services:
+      postgresci:
+        image: postgres:16
+        env:
+          POSTGRES_PASSWORD: c3ram1c
+          POSTGRES_DB: ceramic_one_tests
+        ports:
+          - 5432:5432
   build:
     runs-on: ubuntu-latest
     outputs:

Cargo.toml

@@ -143,7 +143,7 @@ serde_qs = "0.10.1"
 serde_with = "2.1"
 sha2 = { version = "0.10", default-features = false }
 smallvec = "1.10"
-sqlx = { version = "0.7", features = ["sqlite", "runtime-tokio"] }
+sqlx = { version = "0.7", features = ["sqlite", "postgres", "runtime-tokio"] }
 ssh-key = { version = "0.5.1", default-features = false }
 ssi = { version = "0.7", features = ["ed25519"] }
 swagger = { version = "6.1", features = [

Makefile

@@ -54,13 +54,14 @@ gen-kubo-rpc-server:
 check-kubo-rpc-server:
 	./ci-scripts/check_generated_server.sh kubo-rpc-server ./ci-scripts/gen_kubo_rpc_server.sh
 
-.PHONY: check-queries
-check-queries:
-	./ci-scripts/check_queries.sh "sqlite"
-
-.PHONY: check-queries-ci
-check-queries-ci:
-	CI_RUN=1 ./ci-scripts/check_queries.sh "sqlite"
+.PHONY: check-migrations
+check-migrations:
+	MIGRATE_DB=1 ./ci-scripts/check_migrations.sh "sqlite" "postgres"
+
+# Applies migrations to a sqlite and postgres docker container without prompting and removes them afterward
+.PHONY: check-migrations-ci
+check-migrations-ci:
+	MIGRATE_DB=1 CI_RUN=1 MIGRATION_CLEANUP=1 ./ci-scripts/check_migrations.sh "sqlite" "postgres"
 
 .PHONY: release
 release:

beetle/iroh-bitswap/src/lib.rs

@@ -121,6 +121,7 @@ pub trait Store: Send + Sync + 'static {
     async fn get_size(&self, cid: &Cid) -> Result<usize>;
     async fn get(&self, cid: &Cid) -> Result<Block>;
     async fn has(&self, cid: &Cid) -> Result<bool>;
+    async fn put(&self, block: &Block) -> Result<bool>;
 }
 
 #[async_trait::async_trait]
@@ -136,6 +137,10 @@ impl<S: Store> Store for Arc<S> {
     async fn has(&self, cid: &Cid) -> Result<bool> {
         self.as_ref().has(cid).await
     }
+
+    async fn put(&self, block: &Block) -> Result<bool> {
+        self.as_ref().put(block).await
+    }
 }
 
 impl<S: Store> Bitswap<S> {
@@ -655,6 +660,10 @@ mod tests {
         async fn has(&self, _: &Cid) -> Result<bool> {
             todo!()
         }
+
+        async fn put(&self, _: &Block) -> Result<bool> {
+            todo!()
+        }
     }
 
     #[test]
@@ -709,6 +718,15 @@ mod tests {
         async fn has(&self, cid: &Cid) -> Result<bool> {
             Ok(self.store.read().await.contains_key(cid))
         }
+
+        async fn put(&self, block: &Block) -> Result<bool> {
+            Ok(self
+                .store
+                .write()
+                .await
+                .insert(block.cid, block.clone())
+                .is_none())
+        }
     }
 
     #[test(tokio::test)]

ci-scripts/check_migrations.sh

@@ -0,0 +1,120 @@
+#!/usr/bin/env bash
+set -e
+
+# Script to generate the offline sqlx query data
+# Will prompt user if CI_RUN is not set
+# Will run tests if RUN_DB_TESTS is set
+# Will cleanup the db (e.g. stop docker and delete file) if MIGRATION_CLEANUP is set
+
+function prepare_database() {
+    echo "Resetting database at $1"
+    if [ -z "$CI_RUN" ]; then
+        cargo sqlx database reset --database-url "$1"
+    else
+        cargo sqlx database reset --database-url "$1" -y
+    fi
+    echo "Applying migrations at $2"
+    cargo sqlx database setup --database-url "$1" --source "$2"
+    # we no longer support query! since we have multiple sql backends
+    # but we leave it behind a variable in case we want to run it someday
+    if [ -n "$QUERY_MACROS" ]; then
+        cd "$PROJECT_DIR/store"
+        if [ -z "$CI_RUN" ]; then
+            cargo sqlx prepare --database-url "$1"
+        else 
+            cargo sqlx prepare --database-url "$1" --check
+        cd $PROJECT_DIR
+        fi
+    fi
+}
+
+USE_PG=0
+USE_SQLITE=0
+
+check_db_type() {
+    case $1 in
+    postgres*)
+        USE_PG=1
+        ;;
+    sqlite*)
+        USE_SQLITE=1
+        ;;
+    *) ;;
+    esac
+}
+
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+cd "$SCRIPT_DIR/.."
+PROJECT_DIR=$(pwd)
+
+absolute_sqlite_path="$PROJECT_DIR/ceramic_cicddb.sqlite"
+absolute_sqlite_migrations="$PROJECT_DIR/migrations/sqlite"
+
+if [ -n "$TEST_DATABASE_URL" ]; then
+    absolute_pg_path=$TEST_DATABASE_URL
+else
+    absolute_pg_path="postgresql://postgres:c3ram1c@localhost:5432/ceramic_one_tests"
+fi
+absolute_pg_migrations="$PROJECT_DIR/migrations/postgres"
+
+if [ -z "$1" ]; then
+    echo "Parameter required. Should be 'sqlite' or 'postgres'"
+    exit 1
+fi
+
+cargo install sqlx-cli
+
+# Check the first input parameter
+if [ -n "$1" ]; then
+    check_db_type "$1"
+    if [ -n "$2" ]; then
+        check_db_type "$2"
+    fi
+else
+    echo "Must specify 'sqlite' or 'postgres'"
+    exit 1
+fi
+
+if [ -n "$MIGRATE_DB" ]; then
+    if (($USE_SQLITE)); then
+        echo "Using sqlite"
+        prepare_database "sqlite://$absolute_sqlite_path" "$absolute_sqlite_migrations"
+    fi
+
+    if (($USE_PG)); then
+        echo "Using postgres"
+        if [ -z "$TEST_DATABASE_URL" ]; then
+            docker rm ceramic-pg --force 2>/dev/null
+            docker run --name ceramic-pg -e POSTGRES_DB=ceramic_one_tests -e POSTGRES_PASSWORD=c3ram1c -p 5432:5432 -d postgres:16
+            sleep 2
+        fi
+        prepare_database "$absolute_pg_path" "$absolute_pg_migrations"
+    fi
+fi
+
+if [ -n "$RUN_DB_TESTS" ]; then
+    echo "Running tests"
+    PG_TESTS=1 RUSTFLAGS="-D warnings --cfg tokio_unstable" cargo test -p ceramic-store --locked --release
+fi
+
+if [ -n "$MIGRATION_CLEANUP" ]; then
+    if (($USE_SQLITE)); then
+        echo "Cleaning up sqlite"
+        if -w "$absolute_sqlite_path"; then
+            rm $absolute_sqlite_path
+        else 
+            echo "Cannot delete $absolute_sqlite_path (non-existent or not writable)"
+        fi
+    fi
+
+    if (($USE_PG)); then
+        running=$(docker ps -f "name=ceramic-pg" -q)
+        if [ -n "$running" ]; then
+            echo "Cleaning up postgres"
+            docker stop ceramic-pg
+            docker rm ceramic-pg
+        else 
+            echo "No running postgres container found"
+        fi
+    fi
+fi

migrations/postgres/20240321211121_block.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS "ceramic_one_block";

migrations/postgres/20240321211121_block.up.sql

@@ -0,0 +1,9 @@
+-- Add up migration script here
+
+CREATE TABLE IF NOT EXISTS ceramic_one_block (
+    multihash BYTEA NOT NULL,
+    bytes BYTEA NOT NULL,
+    PRIMARY KEY(multihash)
+);
+
+SELECT multihash, bytes FROM ceramic_one_block where false;

migrations/postgres/20240321211303_interest.down.sql

@@ -0,0 +1,3 @@
+-- Add down migration script here
+
+DROP TABLE IF EXISTS ceramic_one_interest;

migrations/postgres/20240321211303_interest.up.sql

@@ -0,0 +1,16 @@
+-- Add up migration script here
+
+CREATE TABLE IF NOT EXISTS ceramic_one_interest (
+    order_key BYTEA NOT NULL, -- network_id sort_value controller StreamID event_cid
+    ahash_0 BIGINT NOT NULL CHECK (ahash_0 >= 0 AND ahash_0 < '4294967296'::BIGINT), -- the ahash is decomposed as [u32; 8]
+    ahash_1 BIGINT NOT NULL CHECK (ahash_1 >= 0 AND ahash_1 < '4294967296'::BIGINT),
+    ahash_2 BIGINT NOT NULL CHECK (ahash_2 >= 0 AND ahash_2 < '4294967296'::BIGINT),
+    ahash_3 BIGINT NOT NULL CHECK (ahash_3 >= 0 AND ahash_3 < '4294967296'::BIGINT),
+    ahash_4 BIGINT NOT NULL CHECK (ahash_4 >= 0 AND ahash_4 < '4294967296'::BIGINT),
+    ahash_5 BIGINT NOT NULL CHECK (ahash_5 >= 0 AND ahash_5 < '4294967296'::BIGINT),
+    ahash_6 BIGINT NOT NULL CHECK (ahash_6 >= 0 AND ahash_6 < '4294967296'::BIGINT),
+    ahash_7 BIGINT NOT NULL CHECK (ahash_7 >= 0 AND ahash_7 < '4294967296'::BIGINT),
+    PRIMARY KEY(order_key)
+);
+
+SELECT order_key, ahash_0, ahash_1, ahash_2, ahash_3, ahash_4, ahash_5, ahash_6, ahash_7 FROM ceramic_one_interest WHERE false;

migrations/postgres/20240321211311_event.down.sql

@@ -0,0 +1,3 @@
+-- Add down migration script here
+
+DROP TABLE IF EXISTS "ceramic_one_event";

migrations/postgres/20240321211311_event.up.sql

@@ -0,0 +1,19 @@
+-- Add up migration script here
+
+CREATE TABLE IF NOT EXISTS ceramic_one_event (
+    order_key BYTEA NOT NULL UNIQUE, -- network_id sep_key sep_value controller stream_id event_cid
+    ahash_0 BIGINT NOT NULL CHECK (ahash_0 >= 0 AND ahash_0 < '4294967296'::BIGINT), -- the ahash is decomposed as [u32; 8]
+    ahash_1 BIGINT NOT NULL CHECK (ahash_1 >= 0 AND ahash_1 < '4294967296'::BIGINT),
+    ahash_2 BIGINT NOT NULL CHECK (ahash_2 >= 0 AND ahash_2 < '4294967296'::BIGINT),
+    ahash_3 BIGINT NOT NULL CHECK (ahash_3 >= 0 AND ahash_3 < '4294967296'::BIGINT),
+    ahash_4 BIGINT NOT NULL CHECK (ahash_4 >= 0 AND ahash_4 < '4294967296'::BIGINT),
+    ahash_5 BIGINT NOT NULL CHECK (ahash_5 >= 0 AND ahash_5 < '4294967296'::BIGINT),
+    ahash_6 BIGINT NOT NULL CHECK (ahash_6 >= 0 AND ahash_6 < '4294967296'::BIGINT),
+    ahash_7 BIGINT NOT NULL CHECK (ahash_7 >= 0 AND ahash_7 < '4294967296'::BIGINT),
+    cid BYTEA NOT NULL, -- the cid of the event as bytes no 0x00 prefix
+    discovered TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    delivered BIGINT UNIQUE, -- monotonic increasing counter indicating this can be delivered to clients
+    PRIMARY KEY(cid)
+);
+
+SELECT order_key, ahash_0, ahash_1, ahash_2, ahash_3, ahash_4, ahash_5, ahash_6, ahash_7, cid, discovered, delivered FROM "ceramic_one_event" WHERE false;

migrations/postgres/20240321211325_event_block.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS ceramic_one_event_block;

migrations/postgres/20240321211325_event_block.up.sql

@@ -0,0 +1,15 @@
+-- Add up migration script here
+CREATE TABLE IF NOT EXISTS ceramic_one_event_block (
+    event_cid BYTEA NOT NULL,
+    block_multihash BYTEA NOT NULL,
+    codec BIGINT NOT NULL, -- the codec of the block
+    idx INTEGER NOT NULL, -- the index of the block in the CAR file
+    "root" BOOL NOT NULL, -- when true the block is a root in the CAR file
+    PRIMARY KEY(event_cid, block_multihash),
+    foreign KEY(event_cid) references ceramic_one_event(cid),
+    foreign KEY(block_multihash) references ceramic_one_block(multihash)
+);
+
+CREATE INDEX IF NOT EXISTS idx_ceramic_one_event_block_block_multihash ON ceramic_one_event_block (block_multihash);
+
+SELECT event_cid, block_multihash, codec, idx, "root" FROM ceramic_one_event_block WHERE false;

migrations/postgres/20240329005642_root.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS "ceramic_one_root";

migrations/postgres/20240329005642_root.up.sql

@@ -0,0 +1,10 @@
+-- Add up migration script here
+CREATE TABLE IF NOT EXISTS "ceramic_one_root" (
+    tx_hash BYTEA NOT NULL, 
+    "root" BYTEA NOT NULL, 
+    block_hash TEXT NOT NULL,
+    "timestamp" BIGINT NOT NULL, 
+    PRIMARY KEY(tx_hash)
+);
+
+SELECT tx_hash, "root", block_hash, "timestamp" FROM "ceramic_one_root" WHERE false;