From 0374cd70b90c8a1faa9fff883476e98f01fe0eef Mon Sep 17 00:00:00 2001 From: 0xgleb <39841057+0xgleb@users.noreply.github.com> Date: Tue, 14 Jul 2026 23:56:21 -0300 Subject: [PATCH] feat: define the 0.4.0 storage schema --- Cargo.lock | 2 + crates/event-sorcery-engine/Cargo.toml | 4 ++ .../event-sorcery-engine/tests/sqlite_ddl.rs | 28 +++++++++ ddl/postgres/0001_initial.sql | 63 +++++++++++++++++++ ddl/sqlite/0001_initial.sql | 63 +++++++++++++++++++ 5 files changed, 160 insertions(+) create mode 100644 crates/event-sorcery-engine/tests/sqlite_ddl.rs create mode 100644 ddl/postgres/0001_initial.sql create mode 100644 ddl/sqlite/0001_initial.sql diff --git a/Cargo.lock b/Cargo.lock index 2cf511e..d381c73 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -532,6 +532,8 @@ name = "event-sorcery-engine" version = "0.4.0" dependencies = [ "ciborium", + "sqlx", + "tokio", ] [[package]] diff --git a/crates/event-sorcery-engine/Cargo.toml b/crates/event-sorcery-engine/Cargo.toml index 40e83ab..fd849b9 100644 --- a/crates/event-sorcery-engine/Cargo.toml +++ b/crates/event-sorcery-engine/Cargo.toml @@ -8,5 +8,9 @@ homepage.workspace = true [dependencies] ciborium.workspace = true +[dev-dependencies] +sqlx.workspace = true +tokio.workspace = true + [lints] workspace = true diff --git a/crates/event-sorcery-engine/tests/sqlite_ddl.rs b/crates/event-sorcery-engine/tests/sqlite_ddl.rs new file mode 100644 index 0000000..3119222 --- /dev/null +++ b/crates/event-sorcery-engine/tests/sqlite_ddl.rs @@ -0,0 +1,28 @@ +use sqlx::SqlitePool; + +#[tokio::test] +async fn sqlite_ddl_creates_the_complete_engine_schema() -> Result<(), Box> { + let pool = SqlitePool::connect(":memory:").await?; + sqlx::migrate!("../../ddl/sqlite").run(&pool).await?; + + for table in [ + "events", + "jobs", + "outbox", + "delivery_receipts", + "reactors", + "projections", + "snapshots", + "schemas", + ] { + let count = sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM sqlite_schema WHERE type = 'table' AND name = ?", + ) + .bind(table) + .fetch_one(&pool) + .await?; + assert_eq!(count, 1, "missing engine table {table}"); + } + + Ok(()) +} diff --git a/ddl/postgres/0001_initial.sql b/ddl/postgres/0001_initial.sql new file mode 100644 index 0000000..9d9898c --- /dev/null +++ b/ddl/postgres/0001_initial.sql @@ -0,0 +1,63 @@ +CREATE TABLE events ( + global_offset BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + sequence BIGINT NOT NULL, + event_type TEXT NOT NULL, + event_version BIGINT NOT NULL, + payload BYTEA NOT NULL, + UNIQUE (aggregate_type, aggregate_id, sequence) +); + +CREATE TABLE jobs ( + job_id TEXT PRIMARY KEY, + record BYTEA NOT NULL, + kind TEXT NOT NULL, + status TEXT NOT NULL, + run_at BIGINT, + lease_expires BIGINT +); + +CREATE INDEX jobs_poll ON jobs (kind, status, run_at, lease_expires); + +CREATE TABLE outbox ( + delivery_id BYTEA PRIMARY KEY, + record BYTEA NOT NULL, + status TEXT NOT NULL, + run_at BIGINT, + lease_expires BIGINT +); + +CREATE INDEX outbox_poll ON outbox (status, run_at, lease_expires); + +CREATE TABLE delivery_receipts ( + delivery_id BYTEA PRIMARY KEY +); + +CREATE TABLE reactors ( + name TEXT PRIMARY KEY, + event_offset BIGINT NOT NULL +); + +CREATE TABLE projections ( + name TEXT PRIMARY KEY, + event_offset BIGINT NOT NULL, + view BYTEA NOT NULL +); + +CREATE TABLE snapshots ( + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + sequence BIGINT NOT NULL, + schema_version BIGINT NOT NULL, + history TEXT NOT NULL CHECK (history IN ('retained', 'compacted')), + payload BYTEA NOT NULL, + PRIMARY KEY (aggregate_type, aggregate_id) +); + +CREATE TABLE schemas ( + kind TEXT NOT NULL CHECK (kind IN ('aggregate', 'projection')), + name TEXT NOT NULL, + schema_version BIGINT NOT NULL, + PRIMARY KEY (kind, name) +); diff --git a/ddl/sqlite/0001_initial.sql b/ddl/sqlite/0001_initial.sql new file mode 100644 index 0000000..e419805 --- /dev/null +++ b/ddl/sqlite/0001_initial.sql @@ -0,0 +1,63 @@ +CREATE TABLE events ( + global_offset INTEGER PRIMARY KEY AUTOINCREMENT, + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + event_type TEXT NOT NULL, + event_version INTEGER NOT NULL, + payload BLOB NOT NULL, + UNIQUE (aggregate_type, aggregate_id, sequence) +) STRICT; + +CREATE TABLE jobs ( + job_id TEXT PRIMARY KEY, + record BLOB NOT NULL, + kind TEXT NOT NULL, + status TEXT NOT NULL, + run_at INTEGER, + lease_expires INTEGER +) STRICT; + +CREATE INDEX jobs_poll ON jobs (kind, status, run_at, lease_expires); + +CREATE TABLE outbox ( + delivery_id BLOB PRIMARY KEY, + record BLOB NOT NULL, + status TEXT NOT NULL, + run_at INTEGER, + lease_expires INTEGER +) STRICT; + +CREATE INDEX outbox_poll ON outbox (status, run_at, lease_expires); + +CREATE TABLE delivery_receipts ( + delivery_id BLOB PRIMARY KEY +) STRICT; + +CREATE TABLE reactors ( + name TEXT PRIMARY KEY, + event_offset INTEGER NOT NULL +) STRICT; + +CREATE TABLE projections ( + name TEXT PRIMARY KEY, + event_offset INTEGER NOT NULL, + view BLOB NOT NULL +) STRICT; + +CREATE TABLE snapshots ( + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + schema_version INTEGER NOT NULL, + history TEXT NOT NULL CHECK (history IN ('retained', 'compacted')), + payload BLOB NOT NULL, + PRIMARY KEY (aggregate_type, aggregate_id) +) STRICT; + +CREATE TABLE schemas ( + kind TEXT NOT NULL CHECK (kind IN ('aggregate', 'projection')), + name TEXT NOT NULL, + schema_version INTEGER NOT NULL, + PRIMARY KEY (kind, name) +) STRICT;