feat: define the 0.4.0 storage schema

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

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<dyn std::error::Error>> {
+    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(())
+}

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)
+);

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;