diff --git a/README.md b/README.md index ae1f501..e3bf45d 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ Events are first written to a durable store (in-memory or SQL), then dispatched - **Status audit trail** — every event status transition is recorded in `event_outbox_status` for ops visibility - **Scheduled delivery** — set `publishAt` in the future; the processor only picks up events whose time has come - **Worker-safe** — MySQL store uses `FOR UPDATE SKIP LOCKED` to allow multiple workers without double-processing -- **Auto schema** — `SqlEventStore` and `SqlRedeliveryStore` each create their own tables on first boot, no migrations needed +- **Schema templates** — versioned DDL files in `migrations/{mysql,sqlite}/` to drop into your migration tool, or a one-line `Schema::create()` helper for projects that want boot-time setup - **Clean architecture boundaries** — `EventSerializer` and `EventHydrator` keep `RawEvent` out of your application layer --- @@ -335,16 +335,31 @@ $store = new InMemoryEventStore(); ### SqlEventStore -Production-ready persistent store. Requires a PDO connection to **MySQL** or **SQLite**. Creates the `event_outbox` table automatically on first boot. +Production-ready persistent store. Requires a PDO connection to **MySQL** or **SQLite**. The store assumes its tables already exist — see [Setting up the schema](#setting-up-the-schema) below. ```php -use Tcds\Io\Raw\Infrastructure\SqlEventStore; +use Vesper\Tool\Event\Infrastructure\SqlEventStore; $pdo = new PDO('mysql:host=localhost;dbname=myapp', 'user', 'pass'); -$store = new SqlEventStore($pdo); // schema created here if not present +$store = new SqlEventStore($pdo); ``` -**Schema created automatically:** +> MySQL workers use `SELECT … FOR UPDATE SKIP LOCKED` on the `event_outbox` table for safe +> concurrent processing. + +If you wire up a `RedeliveryStore` (see [Automatic retry & failure tracking](#automatic-retry--failure-tracking)), +a third table `event_outbox_redelivery` holds per-listener retry state — install it from the same `migrations/` directory. + +### Setting up the schema + +Two options. Pick whichever matches how the host application manages schema. + +**Option 1 — your migration tool runs the shipped DDL (recommended).** Versioned templates live in `migrations/{mysql,sqlite}/` in this package. Each shipped file is immutable; future schema changes ship as new numbered files. The two initial files are: + +- `0001_create_event_outbox.sql` — required for `SqlEventStore` (creates `event_outbox` and `event_outbox_status`). +- `0001_create_event_outbox_redelivery.sql` — required if you use `SqlRedeliveryStore` (creates `event_outbox_redelivery`). + +Either copy the SQL into your own migration file, or have your migration read it from `vendor/hivesper/php-events/migrations/{driver}/`. The DDL the package ships for MySQL looks like: ```sql CREATE TABLE event_outbox ( @@ -368,11 +383,17 @@ CREATE TABLE event_outbox_status ( ); ``` -> MySQL workers use `SELECT … FOR UPDATE SKIP LOCKED` on the `event_outbox` table for safe -> concurrent processing. +**Option 2 — call the shipped `Schema::create()` helper at boot.** Convenient for prototypes, single-app deployments, or projects without their own migration tool. The helper is idempotent (`CREATE … IF NOT EXISTS` on SQLite, `information_schema` check on MySQL), so it's safe to call on every boot — but be aware: if the host application also runs migrations, this can race with them. Prefer Option 1 in that case. -If you wire up a `RedeliveryStore` (see [Automatic retry & failure tracking](#automatic-retry--failure-tracking)), -a third table `event_outbox_redelivery` is created on demand for per-listener retry state. +```php +use Vesper\Tool\Event\Infrastructure\Schema\MysqlEventStoreSchema; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; + +// run once at boot +MysqlEventStoreSchema::create($pdo); // or SqliteEventStoreSchema::create($pdo) +``` + +The redelivery table has matching `MysqlRedeliverySchema::create()` and `SqliteRedeliverySchema::create()` helpers. ### Event lifecycle @@ -675,14 +696,14 @@ many attempts have been made, when the next one should run, what the last error implementations: - **`InMemoryRedeliveryStore`** — array-backed, for tests and dev. -- **`SqlRedeliveryStore`** — durable, MySQL/SQLite-compatible. Auto-creates its - `event_outbox_redelivery` table on first construction. Worker-safe via `FOR UPDATE SKIP LOCKED` - on MySQL. +- **`SqlRedeliveryStore`** — durable, MySQL/SQLite-compatible. Worker-safe via `FOR UPDATE SKIP LOCKED` + on MySQL. Assumes its `event_outbox_redelivery` table already exists — see + [Setting up the schema](#setting-up-the-schema). ```php use Vesper\Tool\Event\Infrastructure\Redelivery\SqlRedeliveryStore; -$store = new SqlRedeliveryStore($pdo); // schema created here if not present +$store = new SqlRedeliveryStore($pdo); ``` ### Wiring it together diff --git a/ROADMAP.md b/ROADMAP.md index 327d3cb..1a07f7c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -52,17 +52,6 @@ column on MySQL). **Trigger:** first post-mortem where a permanently-failed row is unreadable. -### Schema migrations - -Auto-create-on-construct is great for first boot but provides no path for -schema evolution. Adding a column (e.g. `claimed_until` for a lease-based -redelivery claim) won't backfill existing installations. - -**Likely shape:** a `schema_version` column on each table + a -`migrate()` method that runs idempotent ALTERs ordered by version. - -**Trigger:** the first time we need to change an existing column. - ## Scale concerns These matter once volume grows past a single moderate-sized service. diff --git a/migrations/mysql/0001_create_event_outbox.sql b/migrations/mysql/0001_create_event_outbox.sql new file mode 100644 index 0000000..e541aca --- /dev/null +++ b/migrations/mysql/0001_create_event_outbox.sql @@ -0,0 +1,20 @@ +CREATE TABLE event_outbox ( + id VARCHAR(36) NOT NULL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + status VARCHAR(255) NOT NULL, + payload JSON NOT NULL, + created_at DATETIME(6) NOT NULL, + publish_at DATETIME(6) NOT NULL, + + INDEX idx_event_outbox_status_publish (status, publish_at), + INDEX idx_event_outbox_created_at (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +CREATE TABLE event_outbox_status ( + event_id VARCHAR(36) NOT NULL, + status VARCHAR(255) NOT NULL, + error_message TEXT, + created_at DATETIME(6) NOT NULL, + + INDEX idx_event_outbox_status_event_created (event_id, created_at DESC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/migrations/mysql/0001_create_event_outbox_redelivery.sql b/migrations/mysql/0001_create_event_outbox_redelivery.sql new file mode 100644 index 0000000..b033181 --- /dev/null +++ b/migrations/mysql/0001_create_event_outbox_redelivery.sql @@ -0,0 +1,14 @@ +CREATE TABLE event_outbox_redelivery ( + event_id VARCHAR(36) NOT NULL, + listener VARCHAR(500) NOT NULL, + status VARCHAR(32) NOT NULL, + attempt_number INT NOT NULL, + next_retry_at DATETIME(6) NOT NULL, + last_error TEXT NULL, + created_at DATETIME(6) NOT NULL, + updated_at DATETIME(6) NOT NULL, + + PRIMARY KEY (event_id, listener), + INDEX idx_redelivery_due (status, next_retry_at), + INDEX idx_redelivery_event_id (event_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/migrations/sqlite/0001_create_event_outbox.sql b/migrations/sqlite/0001_create_event_outbox.sql new file mode 100644 index 0000000..da34d0d --- /dev/null +++ b/migrations/sqlite/0001_create_event_outbox.sql @@ -0,0 +1,27 @@ +CREATE TABLE event_outbox ( + id TEXT NOT NULL PRIMARY KEY, + name TEXT NOT NULL, + status TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + publish_at TEXT NOT NULL +); + +CREATE INDEX idx_event_outbox_status_publish + ON event_outbox (status, publish_at); + +CREATE INDEX idx_event_outbox_created_at + ON event_outbox (created_at); + +CREATE TABLE event_outbox_status ( + event_id TEXT NOT NULL, + status TEXT NOT NULL, + error_message TEXT, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_event_outbox_status_event + ON event_outbox_status (event_id); + +CREATE INDEX idx_event_outbox_status_event_created + ON event_outbox_status (event_id, created_at); diff --git a/migrations/sqlite/0001_create_event_outbox_redelivery.sql b/migrations/sqlite/0001_create_event_outbox_redelivery.sql new file mode 100644 index 0000000..3321d5f --- /dev/null +++ b/migrations/sqlite/0001_create_event_outbox_redelivery.sql @@ -0,0 +1,17 @@ +CREATE TABLE event_outbox_redelivery ( + event_id TEXT NOT NULL, + listener TEXT NOT NULL, + status TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + next_retry_at TEXT NOT NULL, + last_error TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (event_id, listener) +); + +CREATE INDEX idx_redelivery_due + ON event_outbox_redelivery (status, next_retry_at); + +CREATE INDEX idx_redelivery_event_id + ON event_outbox_redelivery (event_id); diff --git a/src/Infrastructure/Redelivery/SqlRedeliveryStore.php b/src/Infrastructure/Redelivery/SqlRedeliveryStore.php index b75f043..6719df5 100644 --- a/src/Infrastructure/Redelivery/SqlRedeliveryStore.php +++ b/src/Infrastructure/Redelivery/SqlRedeliveryStore.php @@ -8,8 +8,6 @@ use PDO; use RuntimeException; use Throwable; -use Vesper\Tool\Event\Infrastructure\Schema\MysqlRedeliverySchema; -use Vesper\Tool\Event\Infrastructure\Schema\SqliteRedeliverySchema; use Vesper\Tool\Event\RawEvent; use Vesper\Tool\Event\RawEventStatus; use Vesper\Tool\Event\Redelivery\Redelivery; @@ -18,10 +16,7 @@ readonly class SqlRedeliveryStore implements RedeliveryStore { - public function __construct(private PDO $connection) - { - $this->ensureRedeliverySchema(); - } + public function __construct(private PDO $connection) {} #[Override] public function schedule(Redelivery $redelivery): void @@ -248,17 +243,6 @@ private function rollBackIfActive(): void } } - private function ensureRedeliverySchema(): void - { - $driver = $this->driverName(); - - match ($driver) { - 'mysql' => MysqlRedeliverySchema::create($this->connection), - 'sqlite' => SqliteRedeliverySchema::create($this->connection), - default => throw new RuntimeException('Unsupported database driver: ' . $driver), - }; - } - private function lockingClause(): string { return match ($this->driverName()) { diff --git a/src/Infrastructure/SqlEventStore.php b/src/Infrastructure/SqlEventStore.php index 2a10143..78ebd9f 100644 --- a/src/Infrastructure/SqlEventStore.php +++ b/src/Infrastructure/SqlEventStore.php @@ -6,20 +6,14 @@ use Carbon\CarbonInterval; use Override; use PDO; -use RuntimeException; use Throwable; use Vesper\Tool\Event\EventStore; -use Vesper\Tool\Event\Infrastructure\Schema\MysqlEventStoreSchema; -use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; use Vesper\Tool\Event\RawEvent; use Vesper\Tool\Event\RawEventStatus; readonly class SqlEventStore implements EventStore { - public function __construct(private PDO $connection) - { - $this->ensureOutboxSchema(); - } + public function __construct(private PDO $connection) {} #[Override] public function add(RawEvent $event): void @@ -223,17 +217,6 @@ private function rollBackIfActive(): void } } - private function ensureOutboxSchema(): void - { - $driver = $this->driverName(); - - match ($driver) { - 'mysql' => MysqlEventStoreSchema::create($this->connection), - 'sqlite' => SqliteEventStoreSchema::create($this->connection), - default => throw new RuntimeException('Unsupported database driver: ' . $driver), - }; - } - private function lockingClause(): string { return match ($this->driverName()) { diff --git a/tests/Feature/Redelivery/SqlRedeliveryStoreTest.php b/tests/Feature/Redelivery/SqlRedeliveryStoreTest.php index 40ffad0..971faa1 100644 --- a/tests/Feature/Redelivery/SqlRedeliveryStoreTest.php +++ b/tests/Feature/Redelivery/SqlRedeliveryStoreTest.php @@ -9,6 +9,8 @@ use RuntimeException; use Test\Vesper\Tool\Event\_Fixtures\TestEventFactory; use Vesper\Tool\Event\Infrastructure\Redelivery\SqlRedeliveryStore; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteRedeliverySchema; use Vesper\Tool\Event\Infrastructure\SqlEventStore; use Vesper\Tool\Event\RawEvent; use Vesper\Tool\Event\Redelivery\Redelivery; @@ -25,14 +27,17 @@ protected function setUp(): void $this->pdo = new PDO('sqlite::memory:'); $this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + SqliteEventStoreSchema::create($this->pdo); + SqliteRedeliverySchema::create($this->pdo); + $this->eventStore = new SqlEventStore($this->pdo); $this->store = new SqlRedeliveryStore($this->pdo); } - public function test_schema_is_idempotent_across_multiple_instantiations(): void + public function test_schema_create_is_idempotent(): void { - new SqlRedeliveryStore($this->pdo); - new SqlRedeliveryStore($this->pdo); + SqliteRedeliverySchema::create($this->pdo); + SqliteRedeliverySchema::create($this->pdo); self::assertTrue(true); } diff --git a/tests/Feature/SqlEventStoreAuditTest.php b/tests/Feature/SqlEventStoreAuditTest.php index 15250a1..a04290f 100644 --- a/tests/Feature/SqlEventStoreAuditTest.php +++ b/tests/Feature/SqlEventStoreAuditTest.php @@ -5,6 +5,7 @@ use Carbon\CarbonImmutable; use PDO; use PHPUnit\Framework\TestCase; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; use Vesper\Tool\Event\Infrastructure\SqlEventStore; use Vesper\Tool\Event\RawEvent; use Vesper\Tool\Event\RawEventStatus; @@ -19,6 +20,8 @@ protected function setUp(): void $this->pdo = new PDO('sqlite::memory:'); $this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + SqliteEventStoreSchema::create($this->pdo); + $this->store = new SqlEventStore($this->pdo); } diff --git a/tests/Feature/SqlEventStoreRecoverStuckTest.php b/tests/Feature/SqlEventStoreRecoverStuckTest.php index e8d5539..e7d0e09 100644 --- a/tests/Feature/SqlEventStoreRecoverStuckTest.php +++ b/tests/Feature/SqlEventStoreRecoverStuckTest.php @@ -6,6 +6,7 @@ use Carbon\CarbonInterval; use PDO; use PHPUnit\Framework\TestCase; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; use Vesper\Tool\Event\Infrastructure\SqlEventStore; use Vesper\Tool\Event\RawEvent; @@ -19,6 +20,8 @@ protected function setUp(): void $this->pdo = new PDO('sqlite::memory:'); $this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + SqliteEventStoreSchema::create($this->pdo); + $this->store = new SqlEventStore($this->pdo); } diff --git a/tests/Feature/SqlEventStoreTest.php b/tests/Feature/SqlEventStoreTest.php index caa810e..71d54b0 100644 --- a/tests/Feature/SqlEventStoreTest.php +++ b/tests/Feature/SqlEventStoreTest.php @@ -5,6 +5,7 @@ use Carbon\CarbonImmutable; use PDO; use PHPUnit\Framework\TestCase; +use Vesper\Tool\Event\Infrastructure\Schema\SqliteEventStoreSchema; use Vesper\Tool\Event\Infrastructure\SqlEventStore; use Vesper\Tool\Event\RawEvent; @@ -18,6 +19,8 @@ protected function setUp(): void $this->pdo = new PDO('sqlite::memory:'); $this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); + SqliteEventStoreSchema::create($this->pdo); + $this->store = new SqlEventStore($this->pdo); } @@ -130,10 +133,10 @@ public function test_next_returns_event_with_pending_status_from_db(): void self::assertNull($anotherStore->next()); } - public function test_schema_is_idempotent_across_multiple_instantiations(): void + public function test_schema_create_is_idempotent(): void { - new SqlEventStore($this->pdo); - new SqlEventStore($this->pdo); + SqliteEventStoreSchema::create($this->pdo); + SqliteEventStoreSchema::create($this->pdo); self::assertTrue(true); }