From c60cf0f47607a0ff12c639fa5b8edbf10c5dd240 Mon Sep 17 00:00:00 2001 From: Gustavo Freze Date: Thu, 14 May 2026 01:03:42 -0300 Subject: [PATCH 1/3] feat: Add initial implementation of the Outbox pattern with database integration and serializers. --- .claude/CLAUDE.md | 31 + .claude/rules/github-workflows.md | 78 ++ .claude/rules/php-library-code-style.md | 154 ++ .claude/rules/php-library-documentation.md | 40 + .claude/rules/php-library-modeling.md | 163 +++ .claude/rules/php-library-testing.md | 116 ++ .editorconfig | 18 + .gitattributes | 22 + .github/copilot-instructions.md | 11 + .github/dependabot.yml | 31 + .github/workflows/auto-assign.yml | 25 + .github/workflows/ci.yml | 105 ++ .github/workflows/codeql.yml | 35 + .gitignore | 20 + .idea/.gitignore | 5 + .idea/misc.xml | 6 + Makefile | 81 ++ README.md | 459 +++++- composer.json | 84 ++ infection.json.dist | 23 + phpstan.neon.dist | 9 + phpunit.xml | 37 + src/DoctrineOutboxRepository.php | 80 ++ src/Exceptions/DuplicateAggregateSequence.php | 24 + src/Exceptions/DuplicateOutboxEvent.php | 21 + src/Exceptions/InvalidPayloadJson.php | 17 + src/Exceptions/InvalidSnapshotJson.php | 15 + .../OutboxRequiresActiveTransaction.php | 15 + .../PayloadSerializerNotConfigured.php | 17 + .../SnapshotSerializerNotConfigured.php | 17 + src/Internal/BinaryIdentityColumn.php | 20 + src/Internal/ColumnsBuilder.php | 112 ++ src/Internal/IdentityColumn.php | 14 + src/Internal/OutboxInsert.php | 61 + src/Internal/StringIdentityColumn.php | 18 + src/Internal/TableLayoutBuilder.php | 54 + src/OutboxRepository.php | 40 + src/Schema/Columns.php | 61 + src/Schema/IdentityColumnType.php | 23 + src/Schema/TableLayout.php | 39 + src/Schema/UniqueConstraint.php | 29 + src/Serialization/PayloadSerializer.php | 26 + .../PayloadSerializerReflection.php | 20 + src/Serialization/PayloadSerializers.php | 20 + src/Serialization/SerializedPayload.php | 35 + src/Serialization/SerializedSnapshot.php | 35 + src/Serialization/SnapshotSerializer.php | 26 + .../SnapshotSerializerReflection.php | 20 + src/Serialization/SnapshotSerializers.php | 20 + tests/Database.php | 78 ++ tests/DoctrineOutboxRepositoryTest.php | 1239 +++++++++++++++++ tests/InMemoryOutboxRepositoryTest.php | 330 +++++ tests/IntegrationTestCase.php | 37 + tests/Mocks/CustomOrderSnapshotSerializer.php | 22 + tests/Mocks/DriverExceptionStub.php | 11 + tests/Mocks/FallbackOrderPlacedSerializer.php | 23 + tests/Mocks/InMemoryOutboxRepositoryMock.php | 106 ++ tests/Mocks/InvalidPayloadSerializer.php | 23 + tests/Mocks/InvalidSnapshotSerializer.php | 22 + tests/Mocks/OrderPlacedSerializer.php | 23 + tests/Mocks/RefundIssuedSerializer.php | 23 + tests/Models/EventRecordFactory.php | 46 + tests/Models/Order.php | 29 + tests/Models/OrderId.php | 17 + tests/Models/OrderPlaced.php | 13 + tests/Models/RefundIssued.php | 13 + tests/OutboxTableFactory.php | 103 ++ tests/bootstrap.php | 9 + 68 files changed, 4597 insertions(+), 2 deletions(-) create mode 100644 .claude/CLAUDE.md create mode 100644 .claude/rules/github-workflows.md create mode 100644 .claude/rules/php-library-code-style.md create mode 100644 .claude/rules/php-library-documentation.md create mode 100644 .claude/rules/php-library-modeling.md create mode 100644 .claude/rules/php-library-testing.md create mode 100644 .editorconfig create mode 100644 .gitattributes create mode 100644 .github/copilot-instructions.md create mode 100644 .github/dependabot.yml create mode 100644 .github/workflows/auto-assign.yml create mode 100644 .github/workflows/ci.yml create mode 100644 .github/workflows/codeql.yml create mode 100644 .gitignore create mode 100644 .idea/.gitignore create mode 100644 .idea/misc.xml create mode 100644 Makefile create mode 100644 composer.json create mode 100644 infection.json.dist create mode 100644 phpstan.neon.dist create mode 100644 phpunit.xml create mode 100644 src/DoctrineOutboxRepository.php create mode 100644 src/Exceptions/DuplicateAggregateSequence.php create mode 100644 src/Exceptions/DuplicateOutboxEvent.php create mode 100644 src/Exceptions/InvalidPayloadJson.php create mode 100644 src/Exceptions/InvalidSnapshotJson.php create mode 100644 src/Exceptions/OutboxRequiresActiveTransaction.php create mode 100644 src/Exceptions/PayloadSerializerNotConfigured.php create mode 100644 src/Exceptions/SnapshotSerializerNotConfigured.php create mode 100644 src/Internal/BinaryIdentityColumn.php create mode 100644 src/Internal/ColumnsBuilder.php create mode 100644 src/Internal/IdentityColumn.php create mode 100644 src/Internal/OutboxInsert.php create mode 100644 src/Internal/StringIdentityColumn.php create mode 100644 src/Internal/TableLayoutBuilder.php create mode 100644 src/OutboxRepository.php create mode 100644 src/Schema/Columns.php create mode 100644 src/Schema/IdentityColumnType.php create mode 100644 src/Schema/TableLayout.php create mode 100644 src/Schema/UniqueConstraint.php create mode 100644 src/Serialization/PayloadSerializer.php create mode 100644 src/Serialization/PayloadSerializerReflection.php create mode 100644 src/Serialization/PayloadSerializers.php create mode 100644 src/Serialization/SerializedPayload.php create mode 100644 src/Serialization/SerializedSnapshot.php create mode 100644 src/Serialization/SnapshotSerializer.php create mode 100644 src/Serialization/SnapshotSerializerReflection.php create mode 100644 src/Serialization/SnapshotSerializers.php create mode 100644 tests/Database.php create mode 100644 tests/DoctrineOutboxRepositoryTest.php create mode 100644 tests/InMemoryOutboxRepositoryTest.php create mode 100644 tests/IntegrationTestCase.php create mode 100644 tests/Mocks/CustomOrderSnapshotSerializer.php create mode 100644 tests/Mocks/DriverExceptionStub.php create mode 100644 tests/Mocks/FallbackOrderPlacedSerializer.php create mode 100644 tests/Mocks/InMemoryOutboxRepositoryMock.php create mode 100644 tests/Mocks/InvalidPayloadSerializer.php create mode 100644 tests/Mocks/InvalidSnapshotSerializer.php create mode 100644 tests/Mocks/OrderPlacedSerializer.php create mode 100644 tests/Mocks/RefundIssuedSerializer.php create mode 100644 tests/Models/EventRecordFactory.php create mode 100644 tests/Models/Order.php create mode 100644 tests/Models/OrderId.php create mode 100644 tests/Models/OrderPlaced.php create mode 100644 tests/Models/RefundIssued.php create mode 100644 tests/OutboxTableFactory.php create mode 100644 tests/bootstrap.php diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md new file mode 100644 index 0000000..233598b --- /dev/null +++ b/.claude/CLAUDE.md @@ -0,0 +1,31 @@ +# Project + +PHP library (tiny-blocks ecosystem). Self-contained package: immutable models, zero infrastructure +dependencies in core, small public surface area. Public API at `src/` root; implementation details +under `src/Internal/`. + +## Rules + +All coding standards, architecture, naming, testing, and documentation conventions +are defined in `rules/`. Read the applicable rule files before generating any code or documentation. + +## Commands + +- `make test` — runs PHPUnit (with coverage) and Infection (mutation testing) sequentially via `composer tests`. +- `make review` — run lint. +- `make help` — list all available commands. + +## Post-change validation + +After any code change, run `make review`, `make test`, and `make mutation-test`. +If any fails, iterate on the fix while respecting all project rules until all pass. +Never deliver code that breaks lint, tests, or leaves surviving mutants. + +## File formatting + +Every file produced or modified must: + +- Use **LF** line endings. Never CRLF. +- Have no trailing whitespace on any line. +- End with a single trailing newline. +- Have no consecutive blank lines (max one blank line between blocks). diff --git a/.claude/rules/github-workflows.md b/.claude/rules/github-workflows.md new file mode 100644 index 0000000..a369ba4 --- /dev/null +++ b/.claude/rules/github-workflows.md @@ -0,0 +1,78 @@ +--- +description: Naming, ordering, inputs, security, and structural rules for all GitHub Actions workflow files. +paths: + - ".github/workflows/**/*.yml" + - ".github/workflows/**/*.yaml" +--- + +# Workflows + +Structural and stylistic rules for GitHub Actions workflow files. Refer to `shell-scripts.md` for Bash conventions used +inside `run:` steps, and to `terraforms.md` for Terraform conventions used in `terraform/`. + +## Pre-output checklist + +Verify every item before producing any workflow YAML. If any item fails, revise before outputting. + +1. File name follows the convention: `ci-.yml` for reusable CI, `cd-.yml` for dispatch CD. +2. `name` field follows the pattern `CI — ` or `CD — `, using sentence case after the dash + (e.g., `CD — Run migration`, not `CD — Run Migration`). +3. Reusable workflows use `workflow_call` trigger. CD workflows use `workflow_dispatch` trigger. +4. Each workflow has a single responsibility. CI tests code. CD deploys it. Never combine both. +5. Every input has a `description` field. Descriptions use American English and end with a period. +6. Input names use `kebab-case`: `service-name`, `dry-run`, `skip-build`. +7. Inputs are ordered: required first, then optional. Each group by **name length ascending**. +8. Choice input options are in **alphabetical order**. +9. `env`, `outputs`, and `with` entries are ordered by **key length ascending**. +10. `permissions` keys are ordered by **key length ascending** (`contents` before `id-token`). +11. Top-level workflow keys follow canonical order: `name`, `on`, `concurrency`, `permissions`, `env`, `jobs`. +12. Job-level properties follow canonical order: `if`, `name`, `needs`, `uses`, `with`, `runs-on`, + `environment`, `timeout-minutes`, `strategy`, `outputs`, `permissions`, `env`, `steps`. +13. All other YAML property names within a block are ordered by **name length ascending**. +14. Jobs follow execution order: `load-config` → `lint` → `test` → `build` → `deploy`. +15. Step names start with a verb and use sentence case: `Setup PHP`, `Run lint`, `Resolve image tag`. +16. Runtime versions are resolved from the service repo's native dependency file (`composer.json`, `go.mod`, + `package.json`). No version is hardcoded in any workflow. +17. Service-specific overrides live in a pipeline config file (e.g., `.pipeline.yml`) in the service repo, + not in the workflows repository. +18. The `load-config` job reads the pipeline config file at runtime with safe fallback to defaults when absent. +19. Top-level `permissions` defaults to read-only (`contents: read`). Jobs escalate only the permissions they + need. +20. AWS authentication uses OIDC federation exclusively. Static access keys are forbidden. +21. Secrets are passed via `secrets: inherit` from callers. No secret is hardcoded. +22. Sensitive values fetched from SSM are masked with `::add-mask::` before assignment. +23. Third-party actions are pinned to the latest available full commit SHA with a version comment: + `uses: aws-actions/configure-aws-credentials@ # v4.0.2`. Always verify the latest + version before generating a workflow. +24. First-party actions (`actions/*`) are pinned to the latest major version tag available: + `actions/checkout@v4`. Always check for the most recent major version before generating a workflow. +25. Production deployments require GitHub Environments protection rules (manual approval). +26. Every job sets `timeout-minutes` to prevent indefinite hangs. CI jobs: 10–15 minutes. CD jobs: 20–30 + minutes. Adjust only with justification in a comment. +27. CI workflows set `concurrency` with `group` scoped to the PR and `cancel-in-progress: true` to avoid + redundant runs. +28. CD workflows set `concurrency` with `group` scoped to the environment and `cancel-in-progress: false` to + prevent interrupted deployments. +29. CD workflows use `if: ${{ !cancelled() }}` to allow to deploy after optional build steps. +30. Inline logic longer than 3 lines is extracted to a script in `scripts/ci/` or `scripts/cd/`. + +## Style + +- All text (workflow names, step names, input descriptions, comments) uses American English with correct + spelling and punctuation. Sentences and descriptions end with a period. + +## Callers + +- Callers trigger on `pull_request` targeting `main` only. No `push` trigger. +- Callers in service repos are static (~10 lines) and pass only `service-name` or `app-name`. +- Callers reference workflows with `@main` during development. Pin to a tag or SHA for production. + +## Image tagging + +- CD deploy builds: `-sha-` + `latest`. + +## Migrations + +- Migrations run **before** service deployment (schema first, code second). +- `cd-migrate.yml` supports `dry-run` mode (`flyway validate`) for pre-flight checks. +- Database credentials are fetched from SSM at runtime, never stored in workflow files. diff --git a/.claude/rules/php-library-code-style.md b/.claude/rules/php-library-code-style.md new file mode 100644 index 0000000..7ec196e --- /dev/null +++ b/.claude/rules/php-library-code-style.md @@ -0,0 +1,154 @@ +--- +description: Pre-output checklist, naming, typing, complexity, and PHPDoc rules for all PHP files in libraries. +paths: + - "src/**/*.php" + - "tests/**/*.php" +--- + +# Code style + +Semantic code rules for all PHP files. Formatting rules (PSR-1, PSR-4, PSR-12, line length) are enforced by `phpcs.xml` +and are not repeated here. Refer to `php-library-modeling.md` for library modeling rules. + +## Pre-output checklist + +Verify every item before producing any PHP code. If any item fails, revise before outputting. + +1. `declare(strict_types=1)` is present. +2. All classes are `final readonly` by default. Use `class` (without `final` or `readonly`) only when the class is + designed as an extension point for consumers (e.g., `Collection`, `ValueObject`). Use `final class` without + `readonly` only when the parent class is not readonly (e.g., extending a third-party abstract class). +3. All parameters, return types, and properties have explicit types. +4. Constructor property promotion is used. +5. Named arguments are used at call sites for own code, tests, and third-party library methods (e.g., tiny-blocks). + Never use named arguments on native PHP functions (`array_map`, `in_array`, `preg_match`, `is_null`, + `iterator_to_array`, `sprintf`, `implode`, etc.) or PHPUnit assertions (`assertEquals`, `assertSame`, + `assertTrue`, `expectException`, etc.). +6. No `else` or `else if` exists anywhere. Use early returns, polymorphism, or map dispatch instead. +7. No abbreviations appear in identifiers. Use `$index` instead of `$i`, `$account` instead of `$acc`. +8. No generic identifiers exist. Use domain-specific names instead: + `$data` → `$payload`, `$value` → `$totalAmount`, `$item` → `$element`, + `$info` → `$currencyDetails`, `$result` → `$conversionOutcome`. +9. No raw arrays exist where a typed collection or value object is available. Use the `tiny-blocks/collection` + fluent API (`Collection`, `Collectible`) when data is `Collectible`. Use `createLazyFrom` when elements are + consumed once. Raw arrays are acceptable only for primitive configuration data, variadic pass-through, and + interop at system boundaries. See "Collection usage" below for the full rule and example. +10. No private methods exist except private constructors for factory patterns. Inline trivial logic at the call site + or extract it to a collaborator or value object. +11. Members are ordered: constants first, then constructor, then static methods, then instance methods. Within each + group, order by body size ascending (number of lines between `{` and `}`). Constants and enum cases, which have + no body, are ordered by name length ascending. +12. Constructor parameters are ordered by parameter name length ascending (count the name only, without `$` or type), + except when parameters have an implicit semantic order (e.g., `$start/$end`, `$from/$to`, `$startAt/$endAt`), + which takes precedence. Parameters with default values go last, regardless of name length. The same rule + applies to named arguments at call sites. + Example: `$id` (2) → `$value` (5) → `$status` (6) → `$precision` (9). +13. Time and space complexity are first-class design concerns. + - No `O(N²)` or worse time complexity exists unless the problem inherently requires it and the cost is + documented in PHPDoc on the interface method. + - Space complexity is kept minimal: prefer lazy/streaming pipelines (`createLazyFrom`) over materializing + intermediate collections. + - Never re-iterate the same source; fuse stages when possible. + - Public interface methods document time and space complexity in Big O form (see "PHPDoc" section). +14. No logic is duplicated across two or more places (DRY). +15. No abstraction exists without real duplication or isolation need (KISS). +16. All identifiers, comments, and documentation are written in American English. +17. No justification comments exist (`// NOTE:`, `// REASON:`, etc.). Code speaks for itself. +18. `// TODO: ` is used when implementation is unknown, uncertain, or intentionally deferred. + Never leave silent gaps. +19. All class references use `use` imports at the top of the file. Fully qualified names inline are prohibited. +20. No dead or unused code exists. Remove unreferenced classes, methods, constants, and imports. +21. Never create public methods, constants, or classes in `src/` solely to serve tests. If production code does not + need it, it does not exist. +22. Always use the most current and clean syntax available in the target PHP version. Prefer match to switch, + first-class callables over `Closure::fromCallable()`, readonly promotion over manual assignment, enum methods + over external switch/if chains, named arguments over positional ambiguity (except where excluded by rule 5), + and `Collection::map` over foreach accumulation. +23. No vertical alignment of types in parameter lists or property declarations. Use a single space between + type and variable name. Never pad with extra spaces to align columns: + `public OrderId $id` — not `public OrderId $id`. +24. Opening brace `{` follows PSR-12: on a **new line** for classes, interfaces, traits, enums, and methods + (including constructors); on the **same line** for closures and control structures (`if`, `for`, `foreach`, + `while`, `switch`, `match`, `try`). +25. Never pass an argument whose value equals the parameter's default. Omit the argument entirely. + Example — `toArray(KeyPreservation $keyPreservation = KeyPreservation::PRESERVE)`: + `$collection->toArray(keyPreservation: KeyPreservation::PRESERVE)` → `$collection->toArray()`. + Only pass the argument when the value differs from the default. +26. No trailing comma in any multi-line list. This applies to parameter lists (constructors, methods, + closures), argument lists at call sites, array literals, match arms, and any other comma-separated + multi-line structure. The last element never has a comma after it. PHP accepts trailing commas in + parameter lists, but this project prohibits them for visual consistency. + Example — correct: + ``` + new Precision( + value: 2, + rounding: RoundingMode::HALF_UP + ); + ``` + Example — prohibited: + ``` + new Precision( + value: 2, + rounding: RoundingMode::HALF_UP, + ); + ``` + +## Casing conventions + +- Internal code (variables, methods, classes): **`camelCase`**. +- Constants and enum-backed values when representing codes: **`SCREAMING_SNAKE_CASE`**. + +## Naming + +- Names describe **what** in domain terms, not **how** technically: `$monthlyRevenue` instead of `$calculatedValue`. +- Generic technical verbs are avoided. See `php-library-modeling.md` — Nomenclature. +- Booleans use predicate form: `isActive`, `hasPermission`, `wasProcessed`. +- Collections are always plural: `$orders`, `$lines`. +- Methods returning bool use prefixes: `is`, `has`, `can`, `was`, `should`. + +## Comparisons + +1. Null checks: use `is_null($variable)`, never `$variable === null`. +2. Empty string checks on typed `string` parameters: use `$variable === ''`. Avoid `empty()` on typed strings + because `empty('0')` returns `true`. +3. Mixed or untyped checks (value may be `null`, empty string, `0`, or `false`): use `empty($variable)`. + +## American English + +All identifiers, enum values, comments, and error codes use American English spelling: +`canceled` (not `cancelled`), `organization` (not `organisation`), `initialize` (not `initialise`), +`behavior` (not `behaviour`), `modeling` (not `modelling`), `labeled` (not `labelled`), +`fulfill` (not `fulfil`), `color` (not `colour`). + +## PHPDoc + +- PHPDoc is restricted to interfaces only, documenting obligations, `@throws`, and complexity. +- Never add PHPDoc to concrete classes. +- Document `@throws` for every exception the method may raise. +- Document time and space complexity in Big O form. When a method participates in a fused pipeline (e.g., collection + pipelines), express cost as a two-part form: call-site cost + fused-pass contribution. Include a legend defining + variables (e.g., `N` for input size, `K` for number of stages). + +## Collection usage + +When a property or parameter is `Collectible`, use its fluent API. Never break out to raw array functions such as +`array_map`, `array_filter`, `iterator_to_array`, or `foreach` + accumulation. The same applies to `filter()`, +`reduce()`, `each()`, and all other `Collectible` operations. Chain them fluently. Never materialize with +`iterator_to_array` to then pass into a raw `array_*` function. + +**Prohibited — `array_map` + `iterator_to_array` on a Collectible:** + +```php +$names = array_map( + static fn(Element $element): string => $element->name(), + iterator_to_array($collection) +); +``` + +**Correct — fluent chain with `map()` + `toArray()`:** + +```php +$names = $collection + ->map(transformations: static fn(Element $element): string => $element->name()) + ->toArray(keyPreservation: KeyPreservation::DISCARD); +``` diff --git a/.claude/rules/php-library-documentation.md b/.claude/rules/php-library-documentation.md new file mode 100644 index 0000000..4791cb9 --- /dev/null +++ b/.claude/rules/php-library-documentation.md @@ -0,0 +1,40 @@ +--- +description: Standards for README files and all project documentation in PHP libraries. +paths: + - "**/*.md" +--- + +# Documentation + +## README + +1. Include an anchor-linked table of contents. +2. Start with a concise one-line description of what the library does. +3. Include a **license** badge. Do not include any other badges. +4. Provide an **Overview** section explaining the problem the library solves and its design philosophy. +5. **Installation** section: Composer command (`composer require vendor/package`). +6. **How to use** section: complete, runnable code examples covering the primary use cases. Each example + includes a brief heading describing what it demonstrates. +7. If the library exposes multiple entry points, strategies, or container types, document each with its own + subsection and example. +8. **FAQ** section: include entries for common pitfalls, non-obvious behaviors, or design decisions that users + frequently ask about. Each entry is a numbered question as heading (e.g., `### 01. Why does X happen?`) + followed by a concise explanation. Only include entries that address real confusion points. +9. **License** and **Contributing** sections at the end. +10. Write strictly in American English. See `php-library-code-style.md` American English section for spelling + conventions. + +## Structured data + +1. When documenting constructors, factory methods, or configuration options with more than 3 parameters, + use tables with columns: Parameter, Type, Required, Description. +2. Prefer tables to prose for any structured information. + +## Style + +1. Keep language concise and scannable. +2. Never include placeholder content (`TODO`, `TBD`). +3. Code examples must be syntactically correct and self-contained. +4. Code examples include every `use` statement needed to compile. Each example stands alone — copyable into + a fresh file without modification. +5. Do not document `Internal/` classes or private API. Only document what consumers interact with. diff --git a/.claude/rules/php-library-modeling.md b/.claude/rules/php-library-modeling.md new file mode 100644 index 0000000..bedb733 --- /dev/null +++ b/.claude/rules/php-library-modeling.md @@ -0,0 +1,163 @@ +--- +description: Library modeling rules — folder structure, public API boundary, naming, value objects, exceptions, enums, extension points, and complexity. +paths: + - "src/**/*.php" +--- + +# Library modeling + +Libraries are self-contained packages. The core has no dependency on frameworks, databases, or I/O. Refer to +`php-library-code-style.md` for the pre-output checklist applied to all PHP code. + +## Folder structure + +``` +src/ +├── .php # Primary contract for consumers +├── .php # Main implementation or extension point +├── .php # Public enum +├── Contracts/ # Interfaces for data returned to consumers +├── Internal/ # Implementation details (not part of public API) +│ ├── .php +│ └── Exceptions/ # Internal exception classes +├── / # Feature-specific subdirectory when needed +└── Exceptions/ # Public exception classes (when part of the API) +``` + +Never use `Models/`, `Entities/`, `ValueObjects/`, `Enums/`, or `Domain/` as folder names. + +## Public API boundary + +Only interfaces, extension points, enums, and thin orchestration classes live at the `src/` root. These classes +define the contract consumers interact with and delegate all real work to collaborators inside `src/Internal/`. +If a class contains substantial logic (algorithms, state machines, I/O), it belongs in `Internal/`, not at the root. + +The `Internal/` namespace signals classes that are implementation details. Consumers must not depend on them. +Breaking changes inside `Internal/` are not semver-breaking for the library. + +## Nomenclature + +1. Every class, property, method, and exception name reflects the **concept** the library represents. A math library + uses `Precision`, `RoundingMode`; a money library uses `Currency`, `Amount`; a collection library uses + `Collectible`, `Order`. +2. Name classes after what they represent: `Money`, `Color`, `Pipeline` — not after what they do technically. +3. Name methods after the operation in the library's vocabulary: `add()`, `convertTo()`, `splitAt()`. + +### Always banned + +These names carry zero semantic content. Never use them anywhere, as class suffixes, prefixes, or method names: + +- `Data`, `Info`, `Utils`, `Item`, `Record`, `Entity`. +- `Exception` as a class suffix (e.g., `FooException` — use `Foo` when it already extends a native exception). + +### Anemic verbs (banned by default) + +These verbs hide what is actually happening behind a generic action. Banned unless the verb **is** the operation +that constitutes the library's reason to exist (e.g., a JSON parser may have `parse()`; a hashing library may +have `compute()`): + +- `ensure`, `validate`, `check`, `verify`, `assert`, `mark`, `enforce`, `sanitize`, `normalize`, `compute`, + `transform`, `parse`. + +When in doubt, prefer the domain operation name. `Password::hash()` beats `Password::compute()`; `Email::parse()` +is fine in a parser library but suspicious elsewhere (use `Email::from()` instead). + +### Architectural roles (allowed with justification) + +These names describe a role the library offers as a building block. Acceptable when the class **is** that role +(e.g., `EventHandler` in an events library, `CacheManager` in a cache library, `Upcaster` in an event-sourcing +library). Not acceptable on domain objects inside the library (value objects, enums, contract interfaces): + +- `Manager`, `Handler`, `Processor`, `Service`, and their verb forms `process`, `handle`, `execute`. + +The test: if the consumer instantiates or extends this class to integrate with the library, the role name is +legitimate. If the class models a concept the consumer manipulates (a money amount, a country code, a color), +the role name is wrong. + +## Value objects + +1. Are immutable: no setters, no mutation after construction. Operations return new instances. +2. Compare by value, not by reference. +3. Validate invariants in the constructor and throw on invalid input. +4. Have no identity field. +5. Use static factory methods (e.g., `from`, `of`, `zero`) with a private constructor when multiple creation paths + exist. The factory name communicates the semantic intent. + +## Exceptions + +1. Every failure throws a **dedicated exception class** named after the invariant it guards — never + `throw new DomainException('...')`, `throw new InvalidArgumentException('...')`, + `throw new RuntimeException('...')`, or any other generic native exception thrown directly. If the invariant + is worth throwing for, it is worth a named class. +2. Dedicated exception classes **extend** the appropriate native PHP exception (`DomainException`, + `InvalidArgumentException`, `OverflowException`, etc.) — the native class is the parent, never the thing that + is thrown. Consumers that catch the broad standard types continue to work; consumers that need precise handling + can catch the specific classes. +3. Exceptions are pure: no transport-specific fields (`code` populated with HTTP status, formatted `message` meant + for end-user display). Formatting to any transport happens at the consumer's boundary, not inside the library. +4. Exceptions signal invariant violations only, not control flow. +5. Name the class after the invariant violated, never after the technical type: + - `PrecisionOutOfRange` — not `InvalidPrecisionException`. + - `CurrencyMismatch` — not `BadCurrencyException`. + - `ContainerWaitTimeout` — not `TimeoutException`. +6. A descriptive `message` argument is allowed and encouraged when it carries **debugging context** — the violating + value, the boundary that was crossed, the state the library was in. The class name identifies the invariant; + the message describes the specific violation for stack traces and test assertions. Do not build messages meant + for end-user display or transport rendering. Keep them short, factual, and in American English. +7. Public exceptions live in `src/Exceptions/`. Internal exceptions live in `src/Internal/Exceptions/`. + +**Prohibited** — throwing a native exception directly: + +```php +if ($value < 0) { + throw new InvalidArgumentException('Precision cannot be negative.'); +} +``` + +**Correct** — dedicated class, no message (class name is sufficient): + +```php +// src/Exceptions/PrecisionOutOfRange.php +final class PrecisionOutOfRange extends InvalidArgumentException +{ +} + +// at the callsite +if ($value < 0) { + throw new PrecisionOutOfRange(); +} +``` + +**Correct** — dedicated class with debugging context: + +```php +if ($value < 0 || $value > 16) { + throw new PrecisionOutOfRange(sprintf('Precision must be between 0 and 16, got %d.', $value)); +} +``` + +## Enums + +1. Are PHP backed enums. +2. Include methods when they carry vocabulary meaning (e.g., `Order::ASCENDING_KEY`, `RoundingMode::apply()`). +3. Live at the `src/` root when public. Enums used only by internals live in `src/Internal/`. + +## Extension points + +1. When a class is designed to be extended by consumers (e.g., `Collection`, `ValueObject`), it uses `class` instead + of `final readonly class`. All other classes use `final readonly class`. +2. Extension point classes use a private constructor with static factory methods (`createFrom`, `createFromEmpty`) + as the only creation path. +3. Internal state is injected via the constructor and stored in a `private readonly` property. + +## Time and space complexity + +1. Every public method has predictable, documented complexity. Document Big O in PHPDoc on the interface + (see `php-library-code-style.md`, "PHPDoc" section). +2. Algorithms run in `O(N)` or `O(N log N)` unless the problem inherently requires worse. `O(N²)` or worse must + be justified and documented. +3. Prefer lazy/streaming evaluation over materializing intermediate results. In pipeline-style libraries, fuse + stages so a single pass suffices. +4. Memory usage is bounded and proportional to the output, not to the sum of intermediate stages. +5. Validate complexity claims with benchmarks against a reference implementation when optimizing critical paths. + Parity testing against the reference library is the validation standard for optimization work. diff --git a/.claude/rules/php-library-testing.md b/.claude/rules/php-library-testing.md new file mode 100644 index 0000000..610b928 --- /dev/null +++ b/.claude/rules/php-library-testing.md @@ -0,0 +1,116 @@ +--- +description: BDD Given/When/Then structure, PHPUnit conventions, test organization, and fixture rules for PHP libraries. +paths: + - "tests/**/*.php" +--- + +# Testing conventions + +Framework: **PHPUnit**. Refer to `php-library-code-style.md` for the code style checklist, which also applies to +test files. + +## Structure: Given/When/Then (BDD) + +Every test uses `/** @Given */`, `/** @And */`, `/** @When */`, `/** @Then */` doc comments without exception. + +### Happy path example + +```php +public function testAddMoneyWhenSameCurrencyThenAmountsAreSummed(): void +{ + /** @Given two money instances in the same currency */ + $ten = Money::of(amount: 1000, currency: Currency::BRL); + $five = Money::of(amount: 500, currency: Currency::BRL); + + /** @When adding them together */ + $total = $ten->add(other: $five); + + /** @Then the result contains the sum of both amounts */ + self::assertEquals(expected: 1500, actual: $total->amount()); +} +``` + +### Exception example + +When testing that an exception is thrown, place `@Then` (expectException) **before** `@When`. PHPUnit requires this +ordering. + +```php +public function testAddMoneyWhenDifferentCurrenciesThenCurrencyMismatch(): void +{ + /** @Given two money instances in different currencies */ + $brl = Money::of(amount: 1000, currency: Currency::BRL); + $usd = Money::of(amount: 500, currency: Currency::USD); + + /** @Then an exception indicating currency mismatch should be thrown */ + $this->expectException(CurrencyMismatch::class); + + /** @When trying to add money with different currencies */ + $brl->add(other: $usd); +} +``` + +Use `@And` for complementary preconditions or actions within the same scenario, avoiding consecutive `@Given` or +`@When` tags. + +## Rules + +1. Include exactly one `@When` per test. Two actions require two tests. +2. Test only the public API. Never assert on private state or `Internal/` classes directly. +3. Never mock internal collaborators. Use real objects. Use test doubles only at system boundaries (filesystem, + clock, network) when the library interacts with external resources. +4. Name tests to describe behavior, not method names. +5. Never include conditional logic inside tests. +6. Include one logical concept per `@Then` block. +7. Maintain strict independence between tests. No inherited state. +8. Use domain-specific model classes in `tests/Models/` for test fixtures that represent domain concepts + (e.g., `Amount`, `Invoice`, `Order`). +9. Use mock classes in `tests/Mocks/` (or `tests/Unit/Mocks/`) for test doubles of system boundaries + (e.g., `ClientMock`, `ExecutionCompletedMock`). +10. Exercise invariants and edge cases through the library's public entry point. Create a dedicated test class + for an internal model only when the condition cannot be reached through the public API. +11. Never use `/** @test */` annotation. Test methods are discovered by the `test` prefix in the method name. +12. Never use named arguments on PHPUnit assertions (`assertEquals`, `assertSame`, `assertTrue`, + `expectException`, etc.). Pass arguments positionally. + +## Test setup and fixtures + +1. **One annotation = one statement.** Each `@Given` or `@And` block contains exactly one annotation line + followed by one expression or assignment. Never place multiple variable declarations or object + constructions under a single annotation. +2. **No intermediate variables used only once.** If a value is consumed in a single place, inline it at the + call site. Chain method calls when the intermediate state is not referenced elsewhere + (e.g., `Money::of(...)->add(...)` instead of `$money = Money::of(...); $money->add(...);`). +3. **No private or helper methods in test classes.** The only non-test methods allowed are data providers. + If setup logic is complex enough to extract, it belongs in a dedicated fixture class, not in a + private method on the test class. +4. **Domain terms in variables and annotations.** Never use technical testing jargon (`$spy`, `$mock`, + `$stub`, `$fake`, `$dummy`) as variable or property names. Use the domain concept the object + represents: `$collection`, `$amount`, `$currency`, `$sortedElements`. Class names like + `ClientMock` or `GatewaySpy` are acceptable — the variable holding the instance is what matters. +5. **Annotations use domain language.** Write `/** @Given a collection of amounts */`, not + `/** @Given a mocked collection in test state */`. The annotation describes the domain + scenario, not the technical setup. + +## Test organization + +``` +tests/ +├── Models/ # Domain-specific fixtures reused across tests +├── Mocks/ # Test doubles for system boundaries +├── Unit/ # Unit tests for public API +│ └── Mocks/ # Alternative location for test doubles +├── Integration/ # Tests requiring real external resources (Docker, filesystem) +└── bootstrap.php # Test bootstrap when needed +``` + +`tests/Integration/` is only present when the library interacts with infrastructure. + +## Coverage and mutation testing + +1. Line and branch coverage must be **100%**. No annotations (`@codeCoverageIgnore`), attributes, or configuration + that exclude code from coverage are allowed. +2. All mutations reported by Infection must be **killed**. Never ignore or suppress mutants via `infection.json.dist` + or any other mechanism. +3. If a line or mutation cannot be covered or killed, it signals a design problem in the production code. Refactor + the code to make it testable, do not work around the tool. diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..73e3c9a --- /dev/null +++ b/.editorconfig @@ -0,0 +1,18 @@ +root = true + +[*] +charset = utf-8 +end_of_line = lf +indent_size = 4 +indent_style = space +insert_final_newline = true +trim_trailing_whitespace = true + +[*.{yml,yaml}] +indent_size = 2 + +[Makefile] +indent_style = tab + +[*.md] +trim_trailing_whitespace = false diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..744a43b --- /dev/null +++ b/.gitattributes @@ -0,0 +1,22 @@ +* text=auto eol=lf + +*.php text diff=php + +# Dev-only — excluded from the Packagist tarball +/.github export-ignore +/tests export-ignore +/.claude export-ignore +/.editorconfig export-ignore +/.gitattributes export-ignore +/.gitignore export-ignore +/phpunit.xml export-ignore +/phpunit.xml.dist export-ignore +/phpstan.neon export-ignore +/phpstan.neon.dist export-ignore +/phpcs.xml export-ignore +/phpcs.xml.dist export-ignore +/infection.json export-ignore +/infection.json.dist export-ignore +/Makefile export-ignore +/CONTRIBUTING.md export-ignore +/CHANGES.md export-ignore diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 0000000..77c2bb8 --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,11 @@ +# Copilot instructions + +## Context + +PHP library (tiny-blocks). Immutable domain models, zero infrastructure dependencies in core. + +## Mandatory pre-task step + +Before starting any task, read and strictly follow all instruction files located in `.claude/CLAUDE.md` and +`.claude/rules/`. These files are the absolute source of truth for code generation. Apply every rule strictly. Do not +deviate from the patterns, folder structure, or naming conventions defined in them. diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..f0ce8fc --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,31 @@ +version: 2 + +updates: + - package-ecosystem: "composer" + directory: "/" + schedule: + interval: "weekly" + open-pull-requests-limit: 0 + labels: + - "php" + - "security" + - "dependencies" + groups: + php-security: + applies-to: security-updates + patterns: + - "*" + + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + commit-message: + prefix: "build" + labels: + - "dependencies" + - "github-actions" + groups: + github-actions: + patterns: + - "*" diff --git a/.github/workflows/auto-assign.yml b/.github/workflows/auto-assign.yml new file mode 100644 index 0000000..d0ba49e --- /dev/null +++ b/.github/workflows/auto-assign.yml @@ -0,0 +1,25 @@ +name: Auto assign issues and pull requests + +on: + issues: + types: + - opened + pull_request: + types: + - opened + +jobs: + run: + runs-on: ubuntu-latest + permissions: + issues: write + pull-requests: write + steps: + - name: Assign issues and pull requests + uses: gustavofreze/auto-assign@2.1.0 + with: + assignees: '${{ vars.ASSIGNEES }}' + github_token: '${{ secrets.GITHUB_TOKEN }}' + allow_self_assign: 'true' + allow_no_assignees: 'true' + assignment_options: 'ISSUE,PULL_REQUEST' \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..81df43a --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,105 @@ +name: CI + +on: + pull_request: + +concurrency: + group: pr-${{ github.event.pull_request.number }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + load-config: + name: Load config + runs-on: ubuntu-latest + timeout-minutes: 5 + outputs: + php-version: ${{ steps.config.outputs.php-version }} + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Resolve PHP version from composer.json + id: config + run: | + version=$(jq -r '.require.php' composer.json | grep -oP '\d+\.\d+' | head -1) + echo "php-version=$version" >> "$GITHUB_OUTPUT" + + build: + name: Build + needs: load-config + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + tools: composer:2 + php-version: ${{ needs.load-config.outputs.php-version }} + + - name: Validate composer.json + run: composer validate --no-interaction + + - name: Install dependencies + run: composer install --no-progress --optimize-autoloader --prefer-dist --no-interaction + + - name: Upload vendor and composer.lock as artifact + uses: actions/upload-artifact@v4 + with: + name: vendor-artifact + path: | + vendor + composer.lock + + auto-review: + name: Auto review + needs: [load-config, build] + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + tools: composer:2 + php-version: ${{ needs.load-config.outputs.php-version }} + + - name: Download vendor artifact from build + uses: actions/download-artifact@v4 + with: + name: vendor-artifact + path: . + + - name: Run review + run: composer review + + tests: + name: Tests + needs: [load-config, auto-review] + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + tools: composer:2 + php-version: ${{ needs.load-config.outputs.php-version }} + + - name: Download vendor artifact from build + uses: actions/download-artifact@v4 + with: + name: vendor-artifact + path: . + + - name: Run tests + run: composer tests diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..4c6d7f7 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,35 @@ +name: Security checks + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + schedule: + - cron: "0 0 * * *" + +permissions: + actions: read + contents: read + security-events: write + +jobs: + analyze: + name: Analyze + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + language: [ "actions" ] + + steps: + - name: Checkout repository + uses: actions/checkout@v6 + + - name: Initialize CodeQL + uses: github/codeql-action/init@v4 + with: + languages: ${{ matrix.language }} + + - name: Perform CodeQL analysis + uses: github/codeql-action/analyze@v4 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5baa3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,20 @@ +# Agent/IDE +.claude/ +.idea/ +.vscode/ +.cursor/ + +# Composer +/vendor/ +composer.lock + +# PHPUnit / coverage +.phpunit.cache/ +.phpunit.result.cache +report/ +coverage/ +build/ + +# OS +.DS_Store +Thumbs.db diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..b58b603 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,5 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Editor-based HTTP Client requests +/httpRequests/ diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..13515d1 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,6 @@ + + + + + \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..6d6ada2 --- /dev/null +++ b/Makefile @@ -0,0 +1,81 @@ +PWD := $(CURDIR) +ARCH := $(shell uname -m) +PLATFORM := + +ifeq ($(ARCH),arm64) + PLATFORM := --platform=linux/amd64 +endif + +DOCKER_RUN = docker run ${PLATFORM} --rm -it --net=host \ + -e DATABASE_HOST=outbox-test-db \ + -e TEST_DB_HOST_PORT=33306 \ + -e DATABASE_NAME=outbox_test \ + -e DATABASE_USER=outbox \ + -e DATABASE_PASSWORD=outbox \ + -v ${PWD}:/app -w /app \ + -v /var/run/docker.sock:/var/run/docker.sock \ + gustavofreze/php:8.5-alpine + +RESET := \033[0m +GREEN := \033[0;32m +YELLOW := \033[0;33m + +.DEFAULT_GOAL := help + +.PHONY: configure +configure: ## Configure development environment + @${DOCKER_RUN} composer update --optimize-autoloader + @${DOCKER_RUN} composer normalize + +.PHONY: test +test: ## Run all tests with coverage + @${DOCKER_RUN} composer tests + +.PHONY: test-file +test-file: ## Run tests for a specific file (usage: make test-file FILE=ClassNameTest) + @${DOCKER_RUN} composer test-file ${FILE} + +.PHONY: test-no-coverage +test-no-coverage: ## Run all tests without coverage + @${DOCKER_RUN} composer tests-no-coverage + +.PHONY: review +review: ## Run static code analysis + @${DOCKER_RUN} composer review + +.PHONY: show-reports +show-reports: ## Open static analysis reports (e.g., coverage, lints) in the browser + @sensible-browser report/coverage/coverage-html/index.html report/coverage/mutation-report.html + +.PHONY: show-outdated +show-outdated: ## Show outdated direct dependencies + @${DOCKER_RUN} composer outdated --direct + +.PHONY: clean +clean: ## Remove dependencies and generated artifacts + @sudo chown -R ${USER}:${USER} ${PWD} + @rm -rf report vendor .phpunit.cache *.lock + +.PHONY: help +help: ## Display this help message + @echo "Usage: make [target]" + @echo "" + @echo "$$(printf '$(GREEN)')Setup$$(printf '$(RESET)')" + @grep -E '^(configure):.*?## .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":.*? ## "}; {printf "$(YELLOW)%-25s$(RESET) %s\n", $$1, $$2}' + @echo "" + @echo "$$(printf '$(GREEN)')Testing$$(printf '$(RESET)')" + @grep -E '^(test|test-file|test-no-coverage):.*?## .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":.*?## "}; {printf "$(YELLOW)%-25s$(RESET) %s\n", $$1, $$2}' + @echo "" + @echo "$$(printf '$(GREEN)')Quality$$(printf '$(RESET)')" + @grep -E '^(review):.*?## .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":.*?## "}; {printf "$(YELLOW)%-25s$(RESET) %s\n", $$1, $$2}' + @echo "" + @echo "$$(printf '$(GREEN)')Reports$$(printf '$(RESET)')" + @grep -E '^(show-reports|show-outdated):.*?## .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":.*?## "}; {printf "$(YELLOW)%-25s$(RESET) %s\n", $$1, $$2}' + @echo "" + @echo "$$(printf '$(GREEN)')Cleanup$$(printf '$(RESET)')" + @grep -E '^(clean):.*?## .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":.*?## "}; {printf "$(YELLOW)%-25s$(RESET) %s\n", $$1, $$2}' diff --git a/README.md b/README.md index d8a5a2b..c1ad7f2 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,457 @@ -# outbox -Implements the transactional outbox engine for PHP: relay loop, transactional batching, back-off strategies, and observability hooks. +# Outbox + +[![License](https://img.shields.io/badge/license-MIT-green)](https://github.com/tiny-blocks/outbox/blob/main/LICENSE) + +* [Overview](#overview) +* [Installation](#installation) +* [How to use](#how-to-use) + + [Expected table schema](#expected-table-schema) + + [Wiring the repository](#wiring-the-repository) + + [Producing events from an aggregate](#producing-events-from-an-aggregate) + + [Customizing the table layout](#customizing-the-table-layout) + + [Writing a custom payload serializer](#writing-a-custom-payload-serializer) + + [Writing a custom snapshot serializer](#writing-a-custom-snapshot-serializer) + + [Event schema versioning](#event-schema-versioning) +* [FAQ](#faq) +* [License](#license) +* [Contributing](#contributing) + +## Overview + +The **Transactional Outbox** pattern solves the dual-write problem: persisting an aggregate state change and publishing +a domain event must happen atomically. Doing both independently risks a crash leaving one side committed and the other +lost. The outbox pattern records both in the same database transaction, delegating event delivery to a separate relay +process. + +This library is the write-side adapter. It persists outbox records via Doctrine DBAL and is opinionated on correctness. +Transactions are always required and JSON validity is always checked, while leaving every schema decision to you: table +name, column names, and identity column storage type are all configurable. + +The library composes with [`tiny-blocks/building-blocks`](https://github.com/tiny-blocks/building-blocks), which +contributes `DomainEvent`, `DomainEventBehavior`, `EventRecord`, `EventRecords`, `EventType`, `Revision`, +`SequenceNumber`, `SnapshotData`, and the `EventualAggregateRoot` family. This library provides the persistence step +only. + +## Installation + +``` +composer require tiny-blocks/outbox +``` + +## How to use + +### Expected table schema + +The library does not create or manage the outbox table. Add it in your own migration. + +**Default schema (BINARY(16) identity columns, recommended for UUID-based aggregates):** + +```sql +CREATE TABLE outbox_events ( + sequence BIGINT NOT NULL AUTO_INCREMENT UNIQUE, + id BINARY(16) NOT NULL PRIMARY KEY, + aggregate_type VARCHAR(255) NOT NULL, + aggregate_id BINARY(16) NOT NULL, + event_type VARCHAR(255) NOT NULL, + revision INT NOT NULL, + sequence_number BIGINT NOT NULL, + payload JSON NOT NULL, + snapshot JSON NOT NULL, + occurred_at DATETIME(6) NOT NULL, + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + INDEX idx_aggregate (aggregate_id), + UNIQUE KEY uniq_aggregate_sequence (aggregate_type, aggregate_id, sequence_number) +); +``` + +The library writes to `id`, `aggregate_id`, `aggregate_type`, `event_type`, `revision`, `sequence_number`, `payload`, +`snapshot`, and `occurred_at`. It never writes to `sequence` or `created_at`. The database fills those automatically. + +For aggregates whose identities are not UUID v4 strings, use VARCHAR columns and configure `IdentityColumnType::STRING` +(see [Customizing the table layout](#customizing-the-table-layout)): + +```sql +-- For non-UUID identities: VARCHAR(36) or wider. +id VARCHAR(36) NOT NULL PRIMARY KEY, +aggregate_id VARCHAR(36) NOT NULL, +``` + +### Wiring the repository + +`DoctrineOutboxRepository` requires a Doctrine DBAL `Connection` and a `PayloadSerializers` collection. The snapshot +serializer collection defaults to a `SnapshotSerializers` containing `SnapshotSerializerReflection`, and the table +layout defaults to table `outbox_events` with BINARY(16) identity columns. + +```php +beginTransaction(); + +try { + $order = Order::place(orderId: 'order-123'); + $orderRepository->save(order: $order); + $outboxRepository->push(records: $order->recordedEvents()); + $connection->commit(); +} catch (Throwable $exception) { + $connection->rollBack(); + throw $exception; +} +``` + +The aggregate instance is use-once: its recorded-events buffer is never cleared by the library. Discard the instance +after `push()`. Re-saving the same instance pushes the same records again and throws `DuplicateOutboxEvent`. + +### Customizing the table layout + +`TableLayout::builder()` controls the table name. `Columns::builder()` renames individual columns and +switches identity column storage between BINARY(16) and STRING. + +```php +withTableName(tableName: 'domain_events') + ->withColumns(columns: Columns::builder() + ->withId(name: 'id', type: IdentityColumnType::STRING) + ->withEventType(name: 'kind') + ->withAggregateId(name: 'aggregate_id', type: IdentityColumnType::STRING) + ->withAggregateType(name: 'entity_class') + ->withSequenceNumber(name: 'position') + ->build()) + ->build(); + +$repository = new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new PayloadSerializerReflection()]), + tableLayout: $tableLayout +); +``` + +All `Columns::builder()` methods are optional. Omit any method to keep its default. +`withId` and `withAggregateId` require both `name:` and `type:`; all other methods require only `name:`. + +| Method | Default column name | Default type | Description | +|---------------------------------|---------------------|--------------|-----------------------------------------------------------------| +| `withId(name:, type:)` | `id` | `BINARY` | Renames the event id column and/or changes its storage type | +| `withAggregateId(name:, type:)` | `aggregate_id` | `BINARY` | Renames the aggregate id column and/or changes its storage type | +| `withAggregateType(name:)` | `aggregate_type` | — | Renames the aggregate type column | +| `withEventType(name:)` | `event_type` | — | Renames the event type column | +| `withRevision(name:)` | `revision` | — | Renames the schema revision column | +| `withSequenceNumber(name:)` | `sequence_number` | — | Renames the aggregate sequence number column | +| `withPayload(name:)` | `payload` | — | Renames the event payload column | +| `withSnapshot(name:)` | `snapshot` | — | Renames the aggregate snapshot column | +| `withOccurredAt(name:)` | `occurred_at` | — | Renames the event timestamp column | +| `withCreatedAt(name:)` | `created_at` | — | Renames the record creation timestamp column | + +`TableLayout::builder()` controls the table name, columns, and unique constraint name. + +| Method | Default | Description | +|-------------------------------|---------------------------|--------------------------------------------------------------------| +| `withTableName(tableName:)` | `outbox_events` | Sets the outbox table name | +| `withColumns(columns:)` | Default column names | Provides a custom `Columns` configuration | +| `withUniqueConstraint(name:)` | `uniq_aggregate_sequence` | Sets the unique constraint name used to detect duplicate sequences | + +The DDL example uses `uniq_aggregate_sequence` as the unique constraint name. The library expects this name by +default; if you rename it in your DDL, configure it via +`TableLayout::builder()->withUniqueConstraint(name: 'your_name')->build()`. + +Constraint violation detection works with MySQL, MariaDB, PostgreSQL, and SQL Server. These DBMSs include the +constraint name in their violation messages. SQLite is not supported because it omits the constraint name. +All unique violations with SQLite fall under `DuplicateOutboxEvent`. + +### Writing a custom payload serializer + +`PayloadSerializerReflection` covers events whose public properties are scalars or `JsonSerializable`. Implement +`PayloadSerializer` explicitly for events that contain value objects or domain types that need custom JSON shaping. + +Both `supports()` and `serialize()` receive the full `EventRecord`, giving access to `$record->event`, +`$record->aggregateType`, `$record->snapshotData`, and all other fields when routing or shaping the payload. + +Use `match (true)` in `serialize()` to handle multiple event types from the same aggregate in a single serializer: + +```php +event instanceof OrderPlaced || $record->event instanceof OrderShipped; + } + + public function serialize(EventRecord $record): SerializedPayload + { + return match (true) { + $record->event instanceof OrderPlaced => SerializedPayload::from( + payload: json_encode(['orderId' => $record->event->orderId], JSON_THROW_ON_ERROR) + ), + $record->event instanceof OrderShipped => SerializedPayload::from( + payload: json_encode( + ['orderId' => $record->event->orderId, 'shippedAt' => $record->event->shippedAt], + JSON_THROW_ON_ERROR + ) + ) + }; + } +} + +# Register custom serializers before PayloadSerializerReflection. +# PayloadSerializerReflection always returns true from supports(), so it must come last. +$repository = new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [ + new OrderEventSerializer(), + new PayloadSerializerReflection() + ]) +); +``` + +`SerializedPayload::from()` validates the JSON string at construction time and throws `InvalidPayloadJson` if the JSON +is malformed, before the INSERT is attempted. When building the JSON from an array, prefer +`SerializedPayload::fromArray($array)` over `SerializedPayload::from(json_encode($array, JSON_THROW_ON_ERROR))`. The +library handles encoding internally. + +### Writing a custom snapshot serializer + +The aggregate snapshot records the state at the time of each event. By default, `SnapshotSerializerReflection` +serializes `$record->snapshotData->toArray()` using `json_encode`. Provide a custom `SnapshotSerializer` when the +snapshot payload contains value objects or domain types that are not directly JSON-encodable. + +Both `supports()` and `serialize()` receive the full `EventRecord`, giving access to `$record->snapshotData`, +`$record->aggregateType`, `$record->event`, and all other fields when routing or shaping the snapshot. + +```php +aggregateType === 'Order'; + } + + public function serialize(EventRecord $record): SerializedSnapshot + { + $state = $record->snapshotData->toArray(); + + return SerializedSnapshot::from( + snapshot: json_encode( + ['orderId' => $state['orderId']->value, 'status' => $state['status']], + JSON_THROW_ON_ERROR + ) + ); + } +} + +# Register custom snapshot serializers before SnapshotSerializerReflection. +# SnapshotSerializerReflection always returns true from supports(), so it must come last. +$repository = new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new PayloadSerializerReflection()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [ + new OrderSnapshotSerializer(), + new SnapshotSerializerReflection() + ]) +); +``` + +`SerializedSnapshot::from()` validates the JSON string at construction time and throws `InvalidSnapshotJson` if the +JSON is malformed, before the INSERT is attempted. When building the JSON from an array, prefer +`SerializedSnapshot::fromArray($array)` over `SerializedSnapshot::from(json_encode($array, JSON_THROW_ON_ERROR))`. The +library handles encoding internally. + +### Event schema versioning + +Each domain event declares its schema revision via `DomainEvent::revision()`. `DomainEventBehavior` provides the +default implementation, returning revision 1. Override `revision()` when the event's payload structure changes: + +```php +aggregateType === 'Order'`), or the snapshot shaping may vary +based on which event triggered the state change (`$record->event`). Receiving the full `EventRecord` in both +`supports()` and `serialize()` gives serializers access to all available context without requiring any additional +indirection. + +### 09. How does the library handle transient database errors? + +The library catches `UniqueConstraintViolationException` to differentiate +`DuplicateAggregateSequence` from `DuplicateOutboxEvent`. All other DBAL +exceptions, including transient errors like deadlocks (`DeadlockException`), +lock wait timeouts (`LockWaitTimeoutException`), and connection failures +(`ConnectionLost`), propagate unchanged to the caller. + +The consumer is responsible for any retry policy. A common pattern is to wrap +the unit of work (aggregate save + outbox push) in a retry loop that catches +transient exceptions and re-executes the entire transaction. + +## License + +Outbox is licensed under [MIT](LICENSE). + +## Contributing + +Please follow the [contributing guidelines](https://github.com/tiny-blocks/tiny-blocks/blob/main/CONTRIBUTING.md) to +contribute to the project. diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..8b11ba1 --- /dev/null +++ b/composer.json @@ -0,0 +1,84 @@ +{ + "name": "tiny-blocks/outbox", + "description": "Implements the Transactional Outbox pattern. Persists domain events atomically with aggregate state changes through a customizable table schema, reflection-based payload serialization, and built-in support for event schema versioning.", + "license": "MIT", + "type": "library", + "keywords": [ + "ddd", + "outbox", + "tiny-blocks", + "event-store", + "event-driven", + "domain-events", + "event-sourcing", + "event-versioning", + "event-persistence", + "event-serialization", + "transactional-outbox" + ], + "authors": [ + { + "name": "Gustavo Freze de Araujo Santos", + "homepage": "https://github.com/gustavofreze" + } + ], + "homepage": "https://github.com/tiny-blocks/outbox", + "support": { + "issues": "https://github.com/tiny-blocks/outbox/issues", + "source": "https://github.com/tiny-blocks/outbox" + }, + "require": { + "php": "^8.5", + "doctrine/dbal": "^4.0", + "ramsey/uuid": "^4.9", + "tiny-blocks/building-blocks": "^2.2", + "tiny-blocks/collection": "^2.3" + }, + "require-dev": { + "ergebnis/composer-normalize": "^2.51", + "infection/infection": "^0.32", + "phpstan/phpstan": "^2.1", + "phpunit/phpunit": "^13.1", + "squizlabs/php_codesniffer": "^4.0", + "tiny-blocks/docker-container": "^2.0", + "tiny-blocks/environment-variable": "^1.2" + }, + "minimum-stability": "stable", + "prefer-stable": true, + "autoload": { + "psr-4": { + "TinyBlocks\\Outbox\\": "src/" + } + }, + "autoload-dev": { + "psr-4": { + "Test\\TinyBlocks\\Outbox\\": "tests/" + } + }, + "config": { + "allow-plugins": { + "ergebnis/composer-normalize": true, + "infection/extension-installer": true + }, + "sort-packages": true + }, + "scripts": { + "mutation-test": "php ./vendor/bin/infection --threads=max --logger-html=report/coverage/mutation-report.html --coverage=report/coverage", + "phpcs": "php ./vendor/bin/phpcs --standard=PSR12 --extensions=php ./src", + "phpstan": "php ./vendor/bin/phpstan analyse -c phpstan.neon.dist --quiet --no-progress", + "review": [ + "@phpcs", + "@phpstan" + ], + "test": "php -d memory_limit=2G ./vendor/bin/phpunit --configuration phpunit.xml tests", + "test-file": "php ./vendor/bin/phpunit --configuration phpunit.xml --no-coverage --filter", + "test-no-coverage": "php ./vendor/bin/phpunit --configuration phpunit.xml --no-coverage tests", + "tests": [ + "@test", + "@mutation-test" + ], + "tests-no-coverage": [ + "@test-no-coverage" + ] + } +} diff --git a/infection.json.dist b/infection.json.dist new file mode 100644 index 0000000..45c49fc --- /dev/null +++ b/infection.json.dist @@ -0,0 +1,23 @@ +{ + "logs": { + "text": "report/infection/logs/infection-text.log", + "summary": "report/infection/logs/infection-summary.log" + }, + "tmpDir": "report/infection/", + "minMsi": 100, + "timeout": 30, + "source": { + "directories": [ + "src" + ] + }, + "phpUnit": { + "configDir": "", + "customPath": "./vendor/bin/phpunit" + }, + "mutators": { + "@default": true + }, + "minCoveredMsi": 100, + "testFramework": "phpunit" +} diff --git a/phpstan.neon.dist b/phpstan.neon.dist new file mode 100644 index 0000000..297c011 --- /dev/null +++ b/phpstan.neon.dist @@ -0,0 +1,9 @@ +parameters: + paths: + - src + level: 9 + tmpDir: report/phpstan + ignoreErrors: + - identifier: cast.string + - identifier: missingType.iterableValue + reportUnmatchedIgnoredErrors: true diff --git a/phpunit.xml b/phpunit.xml new file mode 100644 index 0000000..97ff87c --- /dev/null +++ b/phpunit.xml @@ -0,0 +1,37 @@ + + + + + + src + + + + + + tests + + + + + + + + + + + + + + + + + diff --git a/src/DoctrineOutboxRepository.php b/src/DoctrineOutboxRepository.php new file mode 100644 index 0000000..d2260f3 --- /dev/null +++ b/src/DoctrineOutboxRepository.php @@ -0,0 +1,80 @@ +tableLayout = $tableLayout ?? TableLayout::default(); + $this->snapshotSerializers = $snapshotSerializers + ?? SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]); + } + + public function push(EventRecords $records): void + { + if (!$this->connection->isTransactionActive()) { + throw OutboxRequiresActiveTransaction::asMissing(); + } + + $records->each(actions: function (EventRecord $record): void { + $payloadSerializer = $this->payloadSerializers->findFor(record: $record); + + if (is_null($payloadSerializer)) { + throw new PayloadSerializerNotConfigured(eventClass: $record->event::class); + } + + $snapshotSerializer = $this->snapshotSerializers->findFor(record: $record); + + if (is_null($snapshotSerializer)) { + throw SnapshotSerializerNotConfigured::for(aggregateType: $record->aggregateType); + } + + $insert = OutboxInsert::from( + record: $record, + payload: $payloadSerializer->serialize(record: $record), + snapshot: $snapshotSerializer->serialize(record: $record), + tableLayout: $this->tableLayout + ); + + try { + $this->connection->executeStatement($insert->sql, $insert->parameters); + } catch (UniqueConstraintViolationException $exception) { + if ($this->tableLayout->uniqueConstraint->isViolatedBy(exception: $exception)) { + throw new DuplicateAggregateSequence( + aggregateId: (string)$record->identity->identityValue(), + aggregateType: $record->aggregateType, + sequenceNumber: $record->sequenceNumber->value, + previous: $exception + ); + } + + throw DuplicateOutboxEvent::forRecord(eventId: $record->id, previous: $exception); + } + }); + } +} diff --git a/src/Exceptions/DuplicateAggregateSequence.php b/src/Exceptions/DuplicateAggregateSequence.php new file mode 100644 index 0000000..cd14141 --- /dev/null +++ b/src/Exceptions/DuplicateAggregateSequence.php @@ -0,0 +1,24 @@ + at sequence number <%d>.'; + + parent::__construct(sprintf($template, $aggregateType, $aggregateId, $sequenceNumber), self::CODE, $previous); + } +} diff --git a/src/Exceptions/DuplicateOutboxEvent.php b/src/Exceptions/DuplicateOutboxEvent.php new file mode 100644 index 0000000..7febc2c --- /dev/null +++ b/src/Exceptions/DuplicateOutboxEvent.php @@ -0,0 +1,21 @@ +.'; + + parent::__construct(sprintf($template, $eventClass)); + } +} diff --git a/src/Exceptions/SnapshotSerializerNotConfigured.php b/src/Exceptions/SnapshotSerializerNotConfigured.php new file mode 100644 index 0000000..94f33a2 --- /dev/null +++ b/src/Exceptions/SnapshotSerializerNotConfigured.php @@ -0,0 +1,17 @@ +getBytes(); + } +} diff --git a/src/Internal/ColumnsBuilder.php b/src/Internal/ColumnsBuilder.php new file mode 100644 index 0000000..eaaf600 --- /dev/null +++ b/src/Internal/ColumnsBuilder.php @@ -0,0 +1,112 @@ +payload = $name; + return $this; + } + + public function withRevision(string $name): ColumnsBuilder + { + $this->revision = $name; + return $this; + } + + public function withCreatedAt(string $name): ColumnsBuilder + { + $this->createdAt = $name; + return $this; + } + + public function withSnapshot(string $name): ColumnsBuilder + { + $this->snapshot = $name; + return $this; + } + + public function withEventType(string $name): ColumnsBuilder + { + $this->eventType = $name; + return $this; + } + + public function withOccurredAt(string $name): ColumnsBuilder + { + $this->occurredAt = $name; + return $this; + } + + public function withAggregateType(string $name): ColumnsBuilder + { + $this->aggregateType = $name; + return $this; + } + + public function withSequenceNumber(string $name): ColumnsBuilder + { + $this->sequenceNumber = $name; + return $this; + } + + public function withId(string $name, IdentityColumnType $type): ColumnsBuilder + { + $this->idName = $name; + $this->idType = $type; + return $this; + } + + public function withAggregateId(string $name, IdentityColumnType $type): ColumnsBuilder + { + $this->aggregateIdName = $name; + $this->aggregateIdType = $type; + return $this; + } + + public function build(): Columns + { + return Columns::from( + id: $this->idType->toColumn(name: $this->idName), + payload: $this->payload, + revision: $this->revision, + snapshot: $this->snapshot, + createdAt: $this->createdAt, + eventType: $this->eventType, + occurredAt: $this->occurredAt, + aggregateId: $this->aggregateIdType->toColumn(name: $this->aggregateIdName), + aggregateType: $this->aggregateType, + sequenceNumber: $this->sequenceNumber + ); + } +} diff --git a/src/Internal/IdentityColumn.php b/src/Internal/IdentityColumn.php new file mode 100644 index 0000000..3718618 --- /dev/null +++ b/src/Internal/IdentityColumn.php @@ -0,0 +1,14 @@ +columns; + $idValue = $columns->id->convert(identityValue: $record->id); + $aggregateIdValue = $columns->aggregateId->convert(identityValue: $record->identity->identityValue()); + + return new OutboxInsert( + sql: sprintf( + $template, + $tableLayout->tableName, + $columns->id->name, + $columns->aggregateId->name, + $columns->aggregateType, + $columns->eventType, + $columns->revision, + $columns->sequenceNumber, + $columns->payload, + $columns->snapshot, + $columns->occurredAt + ), + parameters: [ + 'id' => $idValue, + 'aggregateId' => $aggregateIdValue, + 'aggregateType' => $record->aggregateType, + 'eventType' => $record->type->value, + 'revision' => $record->revision->value, + 'sequenceNumber' => $record->sequenceNumber->value, + 'payload' => $payload->toJson(), + 'snapshot' => $snapshot->toJson(), + 'occurredAt' => $record->occurredOn->toIso8601() + ] + ); + } +} diff --git a/src/Internal/StringIdentityColumn.php b/src/Internal/StringIdentityColumn.php new file mode 100644 index 0000000..4118075 --- /dev/null +++ b/src/Internal/StringIdentityColumn.php @@ -0,0 +1,18 @@ +columns = Columns::default(); + $this->uniqueConstraint = UniqueConstraint::default(); + } + + public static function create(): TableLayoutBuilder + { + return new TableLayoutBuilder(); + } + + public function withColumns(Columns $columns): TableLayoutBuilder + { + $this->columns = $columns; + return $this; + } + + public function withTableName(string $tableName): TableLayoutBuilder + { + $this->tableName = $tableName; + return $this; + } + + public function withUniqueConstraint(string $name): TableLayoutBuilder + { + $this->uniqueConstraint = UniqueConstraint::named(name: $name); + return $this; + } + + public function build(): TableLayout + { + return TableLayout::from( + columns: $this->columns, + tableName: $this->tableName, + uniqueConstraint: $this->uniqueConstraint + ); + } +} diff --git a/src/OutboxRepository.php b/src/OutboxRepository.php new file mode 100644 index 0000000..54305be --- /dev/null +++ b/src/OutboxRepository.php @@ -0,0 +1,40 @@ +Used by aggregate repositories during the business unit of work. The implementation must not + * open or commit a transaction. Atomicity with the aggregate state change is the caller's responsibility.

+ */ +interface OutboxRepository +{ + /** + * Persists the given records as part of the caller's open transaction. + * + *

The implementation must not open or commit a transaction. It is the caller's responsibility + * to ensure this call happens inside the same unit of work as the aggregate state change.

+ * + * @param EventRecords $records The records to persist. + * @throws OutboxRequiresActiveTransaction When called outside an active transaction. + * @throws PayloadSerializerNotConfigured When no serializer supports the event class. + * @throws SnapshotSerializerNotConfigured When no serializer supports the aggregate type. + * @throws InvalidPayloadJson When a serializer produces an invalid JSON payload. + * @throws InvalidSnapshotJson When a serializer produces an invalid JSON snapshot. + * @throws DuplicateOutboxEvent When a record with a duplicate id already exists in the outbox. + * @throws DuplicateAggregateSequence When two records share the same aggregate type, id, and sequence number. + */ + public function push(EventRecords $records): void; +} diff --git a/src/Schema/Columns.php b/src/Schema/Columns.php new file mode 100644 index 0000000..42a51d5 --- /dev/null +++ b/src/Schema/Columns.php @@ -0,0 +1,61 @@ +build(); + } + + public static function from( + IdentityColumn $id, + string $payload, + string $revision, + string $snapshot, + string $createdAt, + string $eventType, + string $occurredAt, + IdentityColumn $aggregateId, + string $aggregateType, + string $sequenceNumber + ): Columns { + return new Columns( + id: $id, + payload: $payload, + revision: $revision, + snapshot: $snapshot, + createdAt: $createdAt, + eventType: $eventType, + occurredAt: $occurredAt, + aggregateId: $aggregateId, + aggregateType: $aggregateType, + sequenceNumber: $sequenceNumber + ); + } +} diff --git a/src/Schema/IdentityColumnType.php b/src/Schema/IdentityColumnType.php new file mode 100644 index 0000000..8c1bd9f --- /dev/null +++ b/src/Schema/IdentityColumnType.php @@ -0,0 +1,23 @@ + BinaryIdentityColumn::named(name: $name), + IdentityColumnType::STRING => StringIdentityColumn::named(name: $name) + }; + } +} diff --git a/src/Schema/TableLayout.php b/src/Schema/TableLayout.php new file mode 100644 index 0000000..f753235 --- /dev/null +++ b/src/Schema/TableLayout.php @@ -0,0 +1,39 @@ +build(); + } + + public static function builder(): TableLayoutBuilder + { + return TableLayoutBuilder::create(); + } +} diff --git a/src/Schema/UniqueConstraint.php b/src/Schema/UniqueConstraint.php new file mode 100644 index 0000000..96cbf61 --- /dev/null +++ b/src/Schema/UniqueConstraint.php @@ -0,0 +1,29 @@ +getMessage(), $this->name); + } +} diff --git a/src/Serialization/PayloadSerializer.php b/src/Serialization/PayloadSerializer.php new file mode 100644 index 0000000..80408b3 --- /dev/null +++ b/src/Serialization/PayloadSerializer.php @@ -0,0 +1,26 @@ +event)); + } +} diff --git a/src/Serialization/PayloadSerializers.php b/src/Serialization/PayloadSerializers.php new file mode 100644 index 0000000..9406c86 --- /dev/null +++ b/src/Serialization/PayloadSerializers.php @@ -0,0 +1,20 @@ +findBy( + predicates: static fn(PayloadSerializer $serializer): bool => $serializer->supports(record: $record) + ); + + return $serializer instanceof PayloadSerializer ? $serializer : null; + } +} diff --git a/src/Serialization/SerializedPayload.php b/src/Serialization/SerializedPayload.php new file mode 100644 index 0000000..2e96b56 --- /dev/null +++ b/src/Serialization/SerializedPayload.php @@ -0,0 +1,35 @@ +payload; + } +} diff --git a/src/Serialization/SerializedSnapshot.php b/src/Serialization/SerializedSnapshot.php new file mode 100644 index 0000000..1a0a8be --- /dev/null +++ b/src/Serialization/SerializedSnapshot.php @@ -0,0 +1,35 @@ +snapshot; + } +} diff --git a/src/Serialization/SnapshotSerializer.php b/src/Serialization/SnapshotSerializer.php new file mode 100644 index 0000000..357615f --- /dev/null +++ b/src/Serialization/SnapshotSerializer.php @@ -0,0 +1,26 @@ +snapshotData->toArray()); + } +} diff --git a/src/Serialization/SnapshotSerializers.php b/src/Serialization/SnapshotSerializers.php new file mode 100644 index 0000000..7eb51c0 --- /dev/null +++ b/src/Serialization/SnapshotSerializers.php @@ -0,0 +1,20 @@ +findBy( + predicates: static fn(SnapshotSerializer $serializer): bool => $serializer->supports(record: $record) + ); + + return $serializer instanceof SnapshotSerializer ? $serializer : null; + } +} diff --git a/tests/Database.php b/tests/Database.php new file mode 100644 index 0000000..fe6435d --- /dev/null +++ b/tests/Database.php @@ -0,0 +1,78 @@ +toString(); + $port = EnvironmentVariable::from(name: 'TEST_DB_HOST_PORT')->toString(); + $database = EnvironmentVariable::from(name: 'DATABASE_NAME')->toString(); + $username = EnvironmentVariable::from(name: 'DATABASE_USER')->toString(); + $password = EnvironmentVariable::from(name: 'DATABASE_PASSWORD')->toString(); + + return new Database( + host: $host, + port: $port, + database: $database, + username: $username, + password: $password + ); + } + + public function start(): void + { + try { + $this->connection()->executeQuery('SELECT 1'); + return; + } catch (Exception) { + } + + new Process(['docker', 'rm', '-f', sprintf('tiny-blocks-reaper-%s', $this->host)])->run(); + new Process(['docker', 'rm', '-f', $this->host])->run(); + + MySQLDockerContainer::from(image: 'mysql:8.4', name: $this->host) + ->pullImage() + ->withTimezone(timezone: 'UTC') + ->withUsername(user: $this->username) + ->withPassword(password: $this->password) + ->withDatabase(database: $this->database) + ->withPortMapping(portOnHost: (int)$this->port, portOnContainer: 3306) + ->withRootPassword(rootPassword: 'root') + ->withGrantedHosts() + ->withoutAutoRemove() + ->withReadinessTimeout(timeoutInSeconds: 60) + ->run(); + } + + public function connection(): Connection + { + return DriverManager::getConnection(params: [ + 'host' => '127.0.0.1', + 'port' => (int)$this->port, + 'user' => $this->username, + 'driver' => 'pdo_mysql', + 'dbname' => $this->database, + 'password' => $this->password + ]); + } +} diff --git a/tests/DoctrineOutboxRepositoryTest.php b/tests/DoctrineOutboxRepositoryTest.php new file mode 100644 index 0000000..4ed8ce7 --- /dev/null +++ b/tests/DoctrineOutboxRepositoryTest.php @@ -0,0 +1,1239 @@ +executeStatement('DROP TABLE IF EXISTS outbox_events'); + OutboxTableFactory::createWithBinaryIdentities( + connection: self::$connection, + tableLayout: TableLayout::default() + ); + } + + public function testPushWhenNoTransactionThenOutboxRequiresActiveTransaction(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And a record to push */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @Then an exception requiring an active transaction is thrown */ + $this->expectException(OutboxRequiresActiveTransaction::class); + $this->expectExceptionMessage('push() must be called within an active transaction.'); + + /** @When pushing without an active transaction */ + $repository->push(records: $records); + } + + public function testPushWhenMultipleSerializersAndFirstMatchesThenFirstIsUsed(): void + { + /** @Given a repository with two serializers supporting the same event */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [ + new OrderPlacedSerializer(), + new FallbackOrderPlacedSerializer() + ]) + ); + + /** @When pushing an order placed event inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then the payload is from the first serializer, not the fallback */ + self::assertSame('{}', self::$connection->fetchOne('SELECT payload FROM outbox_events LIMIT 1')); + } + + public function testPushWhenReflectionPayloadSerializerThenEventPropertiesAreEncoded(): void + { + /** @Given a repository using ReflectionPayloadSerializer as the only serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new PayloadSerializerReflection()]) + ); + + /** @When pushing a record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then the payload reflects the event's public properties */ + self::assertSame('[]', self::$connection->fetchOne('SELECT payload FROM outbox_events LIMIT 1')); + } + + public function testPushWhenCallerRollsBackThenNoRecordPersisted(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And the caller opens a transaction */ + self::$connection->beginTransaction(); + + /** @And a record is pushed inside the caller transaction */ + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + + /** @When the caller rolls back the transaction */ + self::$connection->rollBack(); + + /** @Then no records are persisted */ + self::assertSame(0, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM outbox_events')); + } + + public function testPushWhenTwoRecordsThenBothPersisted(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @When pushing two records inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ), + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then exactly two records are stored */ + self::assertSame(2, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM outbox_events')); + } + + public function testPushWhenUuidWithNullBytesThenBytesPreservedInStorage(): void + { + /** @Given a UUID whose bytes include a leading null byte */ + $recordId = Uuid::fromBytes(bytes: "\x00\x11\x22\x33\x44\x55\x67\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff"); + + /** @And a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @When pushing a record with the null-byte UUID inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + id: $recordId + ) + ])); + self::$connection->commit(); + + /** @Then the retrieved bytes are identical to the original UUID bytes */ + $row = self::$connection->fetchAssociative('SELECT id FROM outbox_events LIMIT 1'); + self::assertSame($recordId->getBytes(), $row['id']); + } + + public function testPushWhenSerializerReturnsInvalidJsonThenInvalidPayloadJson(): void + { + /** @Given a repository with a serializer that produces invalid JSON */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new InvalidPayloadSerializer()]) + ); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a record to push */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @Then an exception indicating invalid JSON payload is thrown */ + $this->expectException(InvalidPayloadJson::class); + $this->expectExceptionMessage('Payload is not valid JSON: not json'); + + /** @When pushing the record */ + $repository->push(records: $records); + } + + public function testPushWhenMultipleSerializersAndSecondMatchesThenCorrectSerializerIsUsed(): void + { + /** @Given a repository with an order serializer followed by a refund serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [ + new OrderPlacedSerializer(), + new RefundIssuedSerializer() + ]) + ); + + /** @When pushing a refund event inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new RefundIssued(), + aggregateType: 'Refund', + eventTypeName: 'RefundIssued' + ) + ])); + self::$connection->commit(); + + /** @Then the payload is from the refund serializer */ + self::assertSame( + '{"type": "refund"}', + self::$connection->fetchOne('SELECT payload FROM outbox_events LIMIT 1') + ); + } + + public function testPushWhenNoSerializerSupportsThenPayloadSerializerNotConfigured(): void + { + /** @Given a repository with no serializers */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFromEmpty() + ); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a record to push */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @Then an exception indicating no serializer is configured is thrown */ + $this->expectException(PayloadSerializerNotConfigured::class); + $this->expectExceptionMessage( + 'No payload serializer configured for event class .' + ); + + /** @When pushing the record */ + $repository->push(records: $records); + } + + public function testPushWhenSerializerDoesNotSupportEventThenPayloadSerializerNotConfigured(): void + { + /** @Given a repository with only an order serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a refund event that the order serializer does not support */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new RefundIssued(), + aggregateType: 'Refund', + eventTypeName: 'RefundIssued' + ) + ]); + + /** @Then an exception indicating no serializer is configured is thrown */ + $this->expectException(PayloadSerializerNotConfigured::class); + $this->expectExceptionMessage( + 'No payload serializer configured for event class .' + ); + + /** @When pushing the unsupported event */ + $repository->push(records: $records); + } + + public function testPushWhenNoSnapshotSerializerSupportsThenSnapshotSerializerNotConfigured(): void + { + /** @Given a repository with empty snapshot serializers */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFromEmpty() + ); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a record to push */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @Then an exception indicating no snapshot serializer is configured is thrown */ + $this->expectException(SnapshotSerializerNotConfigured::class); + $this->expectExceptionMessage('No snapshot serializer configured for aggregate type "Order".'); + + /** @When pushing the record */ + $repository->push(records: $records); + } + + public function testPushWhenDuplicateEventIdThenDuplicateOutboxEvent(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And a record with a fixed id */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And the record is pushed once */ + $repository->push(records: $records); + + /** @Then an exception indicating a duplicate event is thrown */ + $this->expectException(DuplicateOutboxEvent::class); + + /** @When pushing the same record again */ + $repository->push(records: $records); + } + + public function testPushWhenDuplicateAggregateSequenceThenDuplicateAggregateSequence(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And a fixed aggregate identity and sequence number */ + $aggregateId = Uuid::uuid4()->toString(); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a first record is pushed with that aggregate and sequence number */ + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + + /** @Then an exception indicating a duplicate aggregate sequence is thrown */ + $this->expectException(DuplicateAggregateSequence::class); + $this->expectExceptionMessage( + sprintf('Duplicate aggregate sequence for at sequence number <1>.', $aggregateId) + ); + + /** @When pushing a second record with the same aggregate and sequence number but a different id */ + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + } + + public function testPushWhenNoSnapshotSerializerThenReflectionSnapshotSerializerIsUsed(): void + { + /** @Given a repository without an explicit snapshot serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @When pushing a record with snapshot data inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + snapshot: ['status' => 'placed'] + ) + ])); + self::$connection->commit(); + + /** @Then the snapshot is stored using the default reflection serializer */ + self::assertSame( + '{"status": "placed"}', + self::$connection->fetchOne('SELECT snapshot FROM outbox_events LIMIT 1') + ); + } + + public function testPushWhenCustomSnapshotSerializerThenCustomSnapshotIsUsed(): void + { + /** @Given a repository with a custom snapshot serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new CustomOrderSnapshotSerializer()]) + ); + + /** @When pushing a record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then the snapshot uses the custom serializer output */ + self::assertSame('{"custom": true}', self::$connection->fetchOne('SELECT snapshot FROM outbox_events LIMIT 1')); + } + + public function testPushWhenExplicitReflectionSnapshotSerializerThenBehaviorMatchesDefault(): void + { + /** @Given a repository with ReflectionSnapshotSerializer explicitly provided */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @When pushing a record with snapshot data inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + snapshot: ['status' => 'placed'] + ) + ])); + self::$connection->commit(); + + /** @Then the snapshot is identical to the default behavior */ + self::assertSame( + '{"status": "placed"}', + self::$connection->fetchOne('SELECT snapshot FROM outbox_events LIMIT 1') + ); + } + + public function testPushWhenCustomTableNameThenRecordStoredInCustomTable(): void + { + /** @Given a custom table layout with a different table name */ + $tableLayout = TableLayout::builder() + ->withTableName(tableName: 'custom_outbox') + ->build(); + + /** @And any pre-existing custom table is dropped */ + self::$connection->executeStatement('DROP TABLE IF EXISTS custom_outbox'); + + /** @And the custom table is registered for cleanup */ + self::registerTableForCleanup(tableName: 'custom_outbox'); + + /** @And the custom table is created */ + OutboxTableFactory::createWithBinaryIdentities( + connection: self::$connection, + tableLayout: $tableLayout + ); + + /** @And a repository using the custom layout */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + ); + + /** @When pushing a record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then the record is in the custom table */ + self::assertSame(1, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM custom_outbox')); + + /** @And the default table remains empty */ + self::assertSame(0, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM outbox_events')); + } + + public function testPushWhenStringIdentityTypeStoredThenIdIsUuidString(): void + { + /** @Given a layout with STRING identity columns */ + $tableLayout = TableLayout::builder() + ->withTableName(tableName: 'string_outbox') + ->withColumns( + columns: Columns::builder() + ->withId(name: 'id', type: IdentityColumnType::STRING) + ->withAggregateId(name: 'aggregate_id', type: IdentityColumnType::STRING) + ->build() + ) + ->build(); + + /** @And any pre-existing string outbox table is dropped */ + self::$connection->executeStatement('DROP TABLE IF EXISTS string_outbox'); + + /** @And the string outbox table is registered for cleanup */ + self::registerTableForCleanup(tableName: 'string_outbox'); + + /** @And the string outbox table is created */ + OutboxTableFactory::createWithStringIdentities( + connection: self::$connection, + tableLayout: $tableLayout + ); + + /** @And a repository using the string layout */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + ); + + /** @When pushing a record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + self::$connection->commit(); + + /** @Then the id is stored as a 36-character UUID string */ + $row = self::$connection->fetchAssociative('SELECT id FROM string_outbox LIMIT 1'); + self::assertSame(36, strlen($row['id'])); + } + + public function testPushWhenNonUuidAggregateIdWithStringTypeThenStoredAsOriginalString(): void + { + /** @Given a layout with STRING identity columns */ + $tableLayout = TableLayout::builder() + ->withTableName(tableName: 'string_outbox') + ->withColumns( + columns: Columns::builder() + ->withId(name: 'id', type: IdentityColumnType::STRING) + ->withAggregateId(name: 'aggregate_id', type: IdentityColumnType::STRING) + ->build() + ) + ->build(); + + /** @And any pre-existing string outbox table is dropped */ + self::$connection->executeStatement('DROP TABLE IF EXISTS string_outbox'); + + /** @And the string outbox table is registered for cleanup */ + self::registerTableForCleanup(tableName: 'string_outbox'); + + /** @And the string outbox table is created */ + OutboxTableFactory::createWithStringIdentities( + connection: self::$connection, + tableLayout: $tableLayout + ); + + /** @And a repository using the string layout */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + ); + + /** @And a non-UUID aggregate identity */ + $aggregateId = 'ord-1'; + + /** @When pushing a record with a non-UUID aggregate id inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId + ) + ])); + self::$connection->commit(); + + /** @Then the aggregate_id is stored as the original string */ + self::assertSame($aggregateId, self::$connection->fetchOne('SELECT aggregate_id FROM string_outbox LIMIT 1')); + } + + public function testPushWhenSingleRecordThenAllFieldsPersistedCorrectly(): void + { + /** @Given a repository with an order placed serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @When pushing the record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + revision: Revision::of(value: 2), + snapshot: ['status' => 'placed'], + occurredOn: Instant::fromString(value: '2024-06-01 12:00:00.000000'), + sequenceNumber: SequenceNumber::of(value: 3) + ) + ])); + self::$connection->commit(); + + /** @Then the row is retrievable from the database */ + $row = self::$connection->fetchAssociative('SELECT * FROM outbox_events LIMIT 1'); + + /** @And the id is stored as 16-byte binary */ + self::assertSame(16, strlen($row['id'])); + + /** @And the aggregate_id is stored as 16-byte binary */ + self::assertSame(16, strlen($row['aggregate_id'])); + + /** @And the event_type is correct */ + self::assertSame('OrderPlaced', $row['event_type']); + + /** @And the revision is correct */ + self::assertSame(2, (int)$row['revision']); + + /** @And the sequence_number is correct */ + self::assertSame(3, (int)$row['sequence_number']); + + /** @And the payload matches the serializer output */ + self::assertSame('{}', $row['payload']); + + /** @And the snapshot contains the aggregate state */ + self::assertSame('{"status": "placed"}', $row['snapshot']); + + /** @And the payload and snapshot store distinct JSON objects */ + self::assertNotSame($row['payload'], $row['snapshot']); + + /** @And the aggregate_type is correct */ + self::assertSame('Order', $row['aggregate_type']); + + /** @And the occurred_at is stored */ + self::assertStringStartsWith('2024-06-01', $row['occurred_at']); + } + + public function testPushWhenKnownIdThenPersistedIdMatchesOriginal(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @And a record with a known id */ + $recordId = Uuid::uuid4(); + + /** @When pushing the record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + id: $recordId + ) + ])); + self::$connection->commit(); + + /** @Then the persisted id bytes match the original record id */ + $row = self::$connection->fetchAssociative( + 'SELECT id FROM outbox_events WHERE id = UUID_TO_BIN(?)', + [$recordId->toString()] + ); + self::assertSame($recordId->getBytes(), $row['id']); + } + + public function testPushWhenAllColumnNamesAreCustomThenRecordStoredInCustomColumns(): void + { + /** @Given a layout with all column names customized */ + $tableLayout = TableLayout::builder() + ->withTableName(tableName: 'custom_columns_outbox') + ->withColumns( + columns: Columns::builder() + ->withId(name: 'event_id', type: IdentityColumnType::BINARY) + ->withPayload(name: 'event_payload') + ->withRevision(name: 'event_revision') + ->withSnapshot(name: 'event_snapshot') + ->withCreatedAt(name: 'event_created_at') + ->withEventType(name: 'event_event_type') + ->withOccurredAt(name: 'event_occurred_at') + ->withAggregateId(name: 'event_aggregate_id', type: IdentityColumnType::BINARY) + ->withAggregateType(name: 'event_aggregate_type') + ->withSequenceNumber(name: 'event_sequence_number') + ->build() + ) + ->build(); + + /** @And any pre-existing table is dropped */ + self::$connection->executeStatement('DROP TABLE IF EXISTS custom_columns_outbox'); + + /** @And the custom columns table is registered for cleanup */ + self::registerTableForCleanup(tableName: 'custom_columns_outbox'); + + /** @And the table is created with custom column names */ + OutboxTableFactory::createWithBinaryIdentities( + connection: self::$connection, + tableLayout: $tableLayout + ); + + /** @And a repository using this layout */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + ); + + /** @When pushing a record inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + sequenceNumber: SequenceNumber::of(value: 5) + ) + ])); + self::$connection->commit(); + + /** @Then the record is retrievable from the custom table */ + $row = self::$connection->fetchAssociative('SELECT * FROM custom_columns_outbox LIMIT 1'); + + /** @And event_id is stored as 16-byte binary */ + self::assertSame(16, strlen($row['event_id'])); + + /** @And event_aggregate_id is stored as 16-byte binary */ + self::assertSame(16, strlen($row['event_aggregate_id'])); + + /** @And event_aggregate_type is correct */ + self::assertSame('Order', $row['event_aggregate_type']); + + /** @And event_event_type is correct */ + self::assertSame('OrderPlaced', $row['event_event_type']); + + /** @And event_revision is correct */ + self::assertSame(1, (int)$row['event_revision']); + + /** @And event_sequence_number is correct */ + self::assertSame(5, (int)$row['event_sequence_number']); + + /** @And event_payload matches the serializer output */ + self::assertSame('{}', $row['event_payload']); + + /** @And event_snapshot is stored */ + self::assertNotNull($row['event_snapshot']); + + /** @And event_occurred_at is stored */ + self::assertNotNull($row['event_occurred_at']); + + /** @And event_created_at is auto-populated by the database */ + self::assertNotNull($row['event_created_at']); + } + + public function testPushWhenSnapshotSerializerReturnsInvalidJsonThenInvalidSnapshotJson(): void + { + /** @Given a repository with a snapshot serializer that produces invalid JSON */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new InvalidSnapshotSerializer()]) + ); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a record to push */ + $records = EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]); + + /** @Then an exception indicating invalid JSON snapshot is thrown */ + $this->expectException(InvalidSnapshotJson::class); + $this->expectExceptionMessage('Snapshot is not valid JSON: not json'); + + /** @When pushing the record */ + $repository->push(records: $records); + } + + public function testPushWhenEventRecordsIsEmptyThenNoInsertIsExecuted(): void + { + /** @Given a repository */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + ); + + /** @When pushing an empty EventRecords collection inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: EventRecords::createFromEmpty()); + self::$connection->commit(); + + /** @Then no records are persisted */ + self::assertSame(0, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM outbox_events')); + } + + public function testPushWhenRealEventualAggregateRootThenEventRecordIsPersistedCorrectly(): void + { + /** @Given a repository with a reflection payload serializer */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new PayloadSerializerReflection()]) + ); + + /** @When pushing the aggregate's recorded events inside a transaction */ + self::$connection->beginTransaction(); + $repository->push(records: Order::place(orderId: Uuid::uuid4()->toString())->recordedEvents()); + self::$connection->commit(); + + /** @Then one outbox record is persisted */ + self::assertSame(1, (int)self::$connection->fetchOne('SELECT COUNT(*) FROM outbox_events')); + + /** @And the aggregate type is Order */ + self::assertSame('Order', self::$connection->fetchOne('SELECT aggregate_type FROM outbox_events LIMIT 1')); + } + + public function testPushWhenNullTableLayoutThenSqlUsesDefaultTableName(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = $this->createMock(Connection::class); + + /** @And the connection reports an active transaction */ + $connection->method('isTransactionActive')->willReturn(true); + + /** @Then the SQL statement targets the default table name */ + $connection->expects(self::once()) + ->method('executeStatement') + ->with(self::stringContains('outbox_events')) + ->willReturn(1); + + /** @When pushing a record using the default table layout */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenCustomTableLayoutThenSqlUsesCustomTableName(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = $this->createMock(Connection::class); + + /** @And the connection reports an active transaction */ + $connection->method('isTransactionActive')->willReturn(true); + + /** @And a custom table layout with a distinct table name */ + $tableLayout = TableLayout::builder()->withTableName(tableName: 'custom_outbox')->build(); + + /** @Then the SQL statement targets the custom table name */ + $connection->expects(self::once()) + ->method('executeStatement') + ->with(self::stringContains('custom_outbox')) + ->willReturn(1); + + /** @When pushing a record using the custom table layout */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenNullSnapshotSerializersThenExecuteStatementIsCalled(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = $this->createMock(Connection::class); + + /** @And the connection reports an active transaction */ + $connection->method('isTransactionActive')->willReturn(true); + + /** @Then the statement is executed exactly once */ + $connection->expects(self::once())->method('executeStatement')->willReturn(1); + + /** @When pushing a record using the default snapshot serializers */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenCustomSnapshotSerializersThenExecuteStatementUsesCustomSnapshot(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = $this->createMock(Connection::class); + + /** @And the connection reports an active transaction */ + $connection->method('isTransactionActive')->willReturn(true); + + /** @Then the statement is executed with the custom snapshot value */ + $connection->expects(self::once()) + ->method('executeStatement') + ->with( + self::anything(), + self::callback(static fn(array $params): bool => $params['snapshot'] === '{"custom":true}') + ) + ->willReturn(1); + + /** @When pushing a record using custom snapshot serializers */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new CustomOrderSnapshotSerializer()]) + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenRecordHasAllFieldsThenExecuteStatementReceivesAllBindings(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = $this->createMock(Connection::class); + + /** @And the connection reports an active transaction */ + $connection->method('isTransactionActive')->willReturn(true); + + /** @And a table layout with string identity columns */ + $tableLayout = TableLayout::builder() + ->withColumns( + columns: Columns::builder() + ->withId(name: 'id', type: IdentityColumnType::STRING) + ->withAggregateId(name: 'aggregate_id', type: IdentityColumnType::STRING) + ->build() + ) + ->build(); + + /** @And a variable to capture the parameters passed to executeStatement */ + $capturedParameters = null; + + /** @And the connection captures all parameters on executeStatement */ + $connection->expects(self::once()) + ->method('executeStatement') + ->willReturnCallback( + function (string $sql, array $params) use (&$capturedParameters): int { + $capturedParameters = $params; + return 1; + } + ); + + /** @When pushing a record with all deterministic fields */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + id: Uuid::fromString('550e8400-e29b-41d4-a716-446655440000'), + occurredOn: Instant::fromString(value: '2021-01-01T00:00:00+00:00'), + aggregateId: '6ba7b810-9dad-11d1-80b4-00c04fd430c8' + ) + ]) + ); + + /** @Then executeStatement receives all nine expected parameter bindings */ + self::assertSame( + [ + 'id' => '550e8400-e29b-41d4-a716-446655440000', + 'aggregateId' => '6ba7b810-9dad-11d1-80b4-00c04fd430c8', + 'aggregateType' => 'Order', + 'eventType' => 'OrderPlaced', + 'revision' => 1, + 'sequenceNumber' => 1, + 'payload' => '{}', + 'snapshot' => '[]', + 'occurredAt' => '2021-01-01T00:00:00+00:00' + ], + $capturedParameters + ); + } + + public function testPushWhenUniqueConstraintOnSequenceNumberThenDuplicateAggregateSequenceIsThrown(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = self::createConfiguredStub(Connection::class, ['isTransactionActive' => true]); + + /** @And the connection raises a sequence number constraint violation */ + $connection->method('executeStatement')->willThrowException( + new UniqueConstraintViolationException( + new DriverExceptionStub('Duplicate entry for key uniq_aggregate_sequence'), + null + ) + ); + + /** @Then a duplicate aggregate sequence exception is thrown */ + $this->expectException(DuplicateAggregateSequence::class); + $this->expectExceptionMessage( + 'Duplicate aggregate sequence for at sequence number <1>.' + ); + + /** @When pushing the record */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: '6ba7b810-9dad-11d1-80b4-00c04fd430c8' + ) + ]) + ); + } + + public function testPushWhenUniqueConstraintOnEventIdThenDuplicateOutboxEventIsThrown(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = self::createConfiguredStub(Connection::class, ['isTransactionActive' => true]); + + /** @And the connection raises a duplicate event id violation */ + $connection->method('executeStatement')->willThrowException( + new UniqueConstraintViolationException( + new DriverExceptionStub('Duplicate entry for key PRIMARY'), + null + ) + ); + + /** @Then a duplicate outbox event exception is thrown */ + $this->expectException(DuplicateOutboxEvent::class); + $this->expectExceptionMessageMatches('/Event with id ".+" already exists in outbox\./'); + + /** @When pushing the record */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]) + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenUniqueConstraintWithCustomNameThenDuplicateAggregateSequenceIsThrown(): void + { + /** @Given a mocked connection with an active transaction */ + $connection = self::createConfiguredStub(Connection::class, ['isTransactionActive' => true]); + + /** @And a custom table layout with a distinct unique constraint name */ + $tableLayout = TableLayout::builder() + ->withUniqueConstraint(name: 'uniq_custom_outbox_sequence') + ->build(); + + /** @And the connection raises a violation on the custom constraint name */ + $connection->method('executeStatement')->willThrowException( + new UniqueConstraintViolationException( + new DriverExceptionStub('Duplicate entry for key uniq_custom_outbox_sequence'), + null + ) + ); + + /** @Then a duplicate aggregate sequence exception is thrown */ + $this->expectException(DuplicateAggregateSequence::class); + + /** @When pushing a record with the custom table layout */ + new DoctrineOutboxRepository( + connection: $connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + )->push( + records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ]) + ); + } + + public function testPushWhenConstraintNameIsCustomThenDuplicateAggregateSequence(): void + { + /** @Given a custom table layout with a distinct unique constraint name */ + $tableLayout = TableLayout::builder() + ->withTableName(tableName: 'custom_constraint_outbox') + ->withUniqueConstraint(name: 'uniq_custom_outbox_sequence') + ->build(); + + /** @And any pre-existing table is dropped */ + self::$connection->executeStatement('DROP TABLE IF EXISTS custom_constraint_outbox'); + + /** @And the table is registered for cleanup */ + self::registerTableForCleanup(tableName: 'custom_constraint_outbox'); + + /** @And the table is created with the custom constraint name */ + OutboxTableFactory::createWithBinaryIdentities( + connection: self::$connection, + tableLayout: $tableLayout + ); + + /** @And a repository using the custom table layout */ + $repository = new DoctrineOutboxRepository( + connection: self::$connection, + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + tableLayout: $tableLayout + ); + + /** @And a fixed aggregate identity */ + $aggregateId = Uuid::uuid4()->toString(); + + /** @And the connection has an active transaction */ + self::$connection->beginTransaction(); + + /** @And a first record is pushed with that aggregate and sequence number */ + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + + /** @Then an exception indicating a duplicate aggregate sequence is thrown */ + $this->expectException(DuplicateAggregateSequence::class); + + /** @When pushing a second record with the same aggregate and sequence number */ + $repository->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + } +} diff --git a/tests/InMemoryOutboxRepositoryTest.php b/tests/InMemoryOutboxRepositoryTest.php new file mode 100644 index 0000000..7fa94f4 --- /dev/null +++ b/tests/InMemoryOutboxRepositoryTest.php @@ -0,0 +1,330 @@ +beginTransaction(); + + /** @When a single valid event record is pushed */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + + /** @And the transaction is committed */ + $outbox->commit(); + + /** @Then one record is persisted in the repository */ + self::assertCount(1, $outbox->persistedRecords()); + } + + public function testPushWhenMultipleRecordsThenAllArePersistedInOrder(): void + { + /** @Given an in-memory repository with configured serializers */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @When two event records are pushed in a single call */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + sequenceNumber: SequenceNumber::of(value: 1) + ), + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + sequenceNumber: SequenceNumber::of(value: 2) + ) + ])); + + /** @And the transaction is committed */ + $outbox->commit(); + + /** @Then both records are persisted in the repository */ + self::assertCount(2, $outbox->persistedRecords()); + } + + public function testPushWhenNoTransactionIsActiveThenOutboxRequiresActiveTransaction(): void + { + /** @Given an in-memory repository without an active transaction */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @Then an exception requiring an active transaction is thrown */ + $this->expectException(OutboxRequiresActiveTransaction::class); + + /** @When pushing a record without starting a transaction */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + } + + public function testPushWhenNoPayloadSerializerMatchesThenPayloadSerializerNotConfigured(): void + { + /** @Given an in-memory repository with no payload serializers configured */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFromEmpty(), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @Then an exception indicating no configured payload serializer is thrown */ + $this->expectException(PayloadSerializerNotConfigured::class); + + /** @When pushing a record whose event type has no matching serializer */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + } + + public function testPushWhenNoSnapshotSerializerMatchesThenSnapshotSerializerNotConfigured(): void + { + /** @Given an in-memory repository with no snapshot serializers configured */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFromEmpty() + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @Then an exception indicating no configured snapshot serializer is thrown */ + $this->expectException(SnapshotSerializerNotConfigured::class); + + /** @When pushing a record whose aggregate type has no matching snapshot serializer */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + } + + public function testPushWhenPayloadSerializerReturnsInvalidJsonThenInvalidPayloadJson(): void + { + /** @Given an in-memory repository with a serializer that produces invalid JSON */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new InvalidPayloadSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @Then an exception indicating invalid payload JSON is thrown */ + $this->expectException(InvalidPayloadJson::class); + + /** @When pushing a record whose serializer produces malformed JSON */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + } + + public function testPushWhenSnapshotSerializerReturnsInvalidJsonThenInvalidSnapshotJson(): void + { + /** @Given an in-memory repository with a snapshot serializer that produces invalid JSON */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new InvalidSnapshotSerializer()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @Then an exception indicating invalid snapshot JSON is thrown */ + $this->expectException(InvalidSnapshotJson::class); + + /** @When pushing a record whose snapshot serializer produces malformed JSON */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced' + ) + ])); + } + + public function testPushWhenTwoRecordsShareTheSameIdThenDuplicateOutboxEvent(): void + { + /** @Given an in-memory repository with configured serializers */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @And a fixed event id shared by both records */ + $eventId = Uuid::uuid4(); + + /** @And a first record with the fixed id is pushed */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + id: $eventId + ) + ])); + + /** @Then a duplicate outbox event exception is thrown */ + $this->expectException(DuplicateOutboxEvent::class); + + /** @When a second record with the same id is pushed */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + id: $eventId, + sequenceNumber: SequenceNumber::of(value: 2) + ) + ])); + } + + public function testPushWhenTwoRecordsShareTheSameAggregateSequenceThenDuplicateAggregateSequence(): void + { + /** @Given an in-memory repository with configured serializers */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @And a fixed aggregate identity shared by both records */ + $aggregateId = Uuid::uuid4()->toString(); + + /** @And a first record with the fixed aggregate and sequence number 1 is pushed */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + + /** @Then a duplicate aggregate sequence exception is thrown */ + $this->expectException(DuplicateAggregateSequence::class); + + /** @When a second record with the same aggregate and sequence number is pushed */ + $outbox->push(records: EventRecords::createFrom(elements: [ + EventRecordFactory::create( + event: new OrderPlaced(), + aggregateType: 'Order', + eventTypeName: 'OrderPlaced', + aggregateId: $aggregateId, + sequenceNumber: SequenceNumber::of(value: 1) + ) + ])); + } + + public function testPushWhenEventRecordsIsEmptyThenNoRecordIsPersisted(): void + { + /** @Given an in-memory repository with configured serializers */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new OrderPlacedSerializer()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @When an empty EventRecords collection is pushed */ + $outbox->push(records: EventRecords::createFromEmpty()); + + /** @And the transaction is committed */ + $outbox->commit(); + + /** @Then no records are persisted in the repository */ + self::assertCount(0, $outbox->persistedRecords()); + } + + public function testPushWhenRealEventualAggregateRootThenEventRecordIsPersisted(): void + { + /** @Given an in-memory repository with a reflection payload serializer */ + $outbox = new InMemoryOutboxRepositoryMock( + payloadSerializers: PayloadSerializers::createFrom(elements: [new PayloadSerializerReflection()]), + snapshotSerializers: SnapshotSerializers::createFrom(elements: [new SnapshotSerializerReflection()]) + ); + + /** @And a transaction is started */ + $outbox->beginTransaction(); + + /** @When the aggregate's recorded events are pushed */ + $outbox->push(records: Order::place(orderId: Uuid::uuid4()->toString())->recordedEvents()); + + /** @And the transaction is committed */ + $outbox->commit(); + + /** @Then one event record is persisted in the repository */ + self::assertCount(1, $outbox->persistedRecords()); + } +} diff --git a/tests/IntegrationTestCase.php b/tests/IntegrationTestCase.php new file mode 100644 index 0000000..cf1a8ab --- /dev/null +++ b/tests/IntegrationTestCase.php @@ -0,0 +1,37 @@ +connection(); + } + + protected static function registerTableForCleanup(string $tableName): void + { + self::$customTables[] = $tableName; + } + + protected function tearDown(): void + { + if (self::$connection->isTransactionActive()) { + self::$connection->rollBack(); + } + + foreach (self::$customTables as $tableName) { + self::$connection->executeStatement(sprintf('DROP TABLE IF EXISTS %s', $tableName)); + } + + self::$customTables = []; + } +} diff --git a/tests/Mocks/CustomOrderSnapshotSerializer.php b/tests/Mocks/CustomOrderSnapshotSerializer.php new file mode 100644 index 0000000..c76b11f --- /dev/null +++ b/tests/Mocks/CustomOrderSnapshotSerializer.php @@ -0,0 +1,22 @@ +event instanceof OrderPlaced; + } + + public function serialize(EventRecord $record): SerializedPayload + { + return SerializedPayload::from(payload: '{"fallback":true}'); + } +} diff --git a/tests/Mocks/InMemoryOutboxRepositoryMock.php b/tests/Mocks/InMemoryOutboxRepositoryMock.php new file mode 100644 index 0000000..d33f3b3 --- /dev/null +++ b/tests/Mocks/InMemoryOutboxRepositoryMock.php @@ -0,0 +1,106 @@ +transactionActive = true; + } + + public function commit(): void + { + $this->transactionActive = false; + } + + public function persistedRecords(): array + { + return $this->records; + } + + public function rollback(): void + { + $this->transactionActive = false; + $this->records = []; + $this->aggregateSequences = []; + } + + public function push(EventRecords $records): void + { + if (!$this->transactionActive) { + throw OutboxRequiresActiveTransaction::asMissing(); + } + + $records->each(actions: function (EventRecord $record): void { + $payloadSerializer = $this->payloadSerializers->findFor(record: $record); + + if (is_null($payloadSerializer)) { + throw new PayloadSerializerNotConfigured(eventClass: $record->event::class); + } + + $snapshotSerializer = $this->snapshotSerializers->findFor(record: $record); + + if (is_null($snapshotSerializer)) { + throw SnapshotSerializerNotConfigured::for(aggregateType: $record->aggregateType); + } + + $payloadSerializer->serialize(record: $record); + $snapshotSerializer->serialize(record: $record); + + $aggregateKey = sprintf( + '%s|%s|%d', + $record->aggregateType, + $record->identity->identityValue(), + $record->sequenceNumber->value + ); + + if (isset($this->aggregateSequences[$aggregateKey])) { + throw new DuplicateAggregateSequence( + aggregateId: (string)$record->identity->identityValue(), + aggregateType: $record->aggregateType, + sequenceNumber: $record->sequenceNumber->value + ); + } + + $eventId = (string)$record->id; + + if (isset($this->records[$eventId])) { + throw DuplicateOutboxEvent::forRecord( + eventId: $record->id, + previous: new UniqueConstraintViolationException( + new DriverExceptionStub('Duplicate entry for key PRIMARY'), + null + ) + ); + } + + $this->aggregateSequences[$aggregateKey] = true; + $this->records[$eventId] = $record; + }); + } +} diff --git a/tests/Mocks/InvalidPayloadSerializer.php b/tests/Mocks/InvalidPayloadSerializer.php new file mode 100644 index 0000000..14ca476 --- /dev/null +++ b/tests/Mocks/InvalidPayloadSerializer.php @@ -0,0 +1,23 @@ +event instanceof OrderPlaced; + } + + public function serialize(EventRecord $record): SerializedPayload + { + return SerializedPayload::from(payload: 'not json'); + } +} diff --git a/tests/Mocks/InvalidSnapshotSerializer.php b/tests/Mocks/InvalidSnapshotSerializer.php new file mode 100644 index 0000000..8f7315f --- /dev/null +++ b/tests/Mocks/InvalidSnapshotSerializer.php @@ -0,0 +1,22 @@ +event instanceof OrderPlaced; + } + + public function serialize(EventRecord $record): SerializedPayload + { + return SerializedPayload::from(payload: '{}'); + } +} diff --git a/tests/Mocks/RefundIssuedSerializer.php b/tests/Mocks/RefundIssuedSerializer.php new file mode 100644 index 0000000..525cef3 --- /dev/null +++ b/tests/Mocks/RefundIssuedSerializer.php @@ -0,0 +1,23 @@ +event instanceof RefundIssued; + } + + public function serialize(EventRecord $record): SerializedPayload + { + return SerializedPayload::from(payload: '{"type":"refund"}'); + } +} diff --git a/tests/Models/EventRecordFactory.php b/tests/Models/EventRecordFactory.php new file mode 100644 index 0000000..6c37b83 --- /dev/null +++ b/tests/Models/EventRecordFactory.php @@ -0,0 +1,46 @@ +toString()), + revision: $revision ?? Revision::initial(), + occurredOn: $occurredOn ?? Instant::now(), + snapshotData: new SnapshotData(payload: $snapshot ?? []), + aggregateType: $aggregateType, + sequenceNumber: $sequenceNumber ?? SequenceNumber::first() + ); + } +} diff --git a/tests/Models/Order.php b/tests/Models/Order.php new file mode 100644 index 0000000..4046e23 --- /dev/null +++ b/tests/Models/Order.php @@ -0,0 +1,29 @@ +push(event: new OrderPlaced()); + return $order; + } + + protected function snapshotState(): array + { + return ['status' => 'placed']; + } +} diff --git a/tests/Models/OrderId.php b/tests/Models/OrderId.php new file mode 100644 index 0000000..eaad4f3 --- /dev/null +++ b/tests/Models/OrderId.php @@ -0,0 +1,17 @@ +executeStatement( + sprintf( + $template, + $tableLayout->tableName, + $tableLayout->columns->id->name, + $tableLayout->columns->aggregateType, + $tableLayout->columns->aggregateId->name, + $tableLayout->columns->eventType, + $tableLayout->columns->revision, + $tableLayout->columns->sequenceNumber, + $tableLayout->columns->payload, + $tableLayout->columns->snapshot, + $tableLayout->columns->occurredAt, + $tableLayout->columns->createdAt, + $tableLayout->columns->aggregateId->name, + $tableLayout->uniqueConstraint->name, + $tableLayout->columns->aggregateType, + $tableLayout->columns->aggregateId->name, + $tableLayout->columns->sequenceNumber + ) + ); + } + + public static function createWithStringIdentities(Connection $connection, TableLayout $tableLayout): void + { + $template = <<executeStatement( + sprintf( + $template, + $tableLayout->tableName, + $tableLayout->columns->id->name, + $tableLayout->columns->aggregateType, + $tableLayout->columns->aggregateId->name, + $tableLayout->columns->eventType, + $tableLayout->columns->revision, + $tableLayout->columns->sequenceNumber, + $tableLayout->columns->payload, + $tableLayout->columns->snapshot, + $tableLayout->columns->occurredAt, + $tableLayout->columns->createdAt, + $tableLayout->columns->aggregateId->name, + $tableLayout->uniqueConstraint->name, + $tableLayout->columns->aggregateType, + $tableLayout->columns->aggregateId->name, + $tableLayout->columns->sequenceNumber + ) + ); + } +} diff --git a/tests/bootstrap.php b/tests/bootstrap.php new file mode 100644 index 0000000..8a12f71 --- /dev/null +++ b/tests/bootstrap.php @@ -0,0 +1,9 @@ +start(); From 702a85de8f6257a5a53338168ca71ffe62e5d3e2 Mon Sep 17 00:00:00 2001 From: Gustavo Freze Date: Thu, 14 May 2026 01:05:21 -0300 Subject: [PATCH 2/3] chore: remove .idea directory from version control Co-Authored-By: Claude Sonnet 4.6 --- .idea/.gitignore | 5 ----- .idea/misc.xml | 6 ------ 2 files changed, 11 deletions(-) delete mode 100644 .idea/.gitignore delete mode 100644 .idea/misc.xml diff --git a/.idea/.gitignore b/.idea/.gitignore deleted file mode 100644 index b58b603..0000000 --- a/.idea/.gitignore +++ /dev/null @@ -1,5 +0,0 @@ -# Default ignored files -/shelf/ -/workspace.xml -# Editor-based HTTP Client requests -/httpRequests/ diff --git a/.idea/misc.xml b/.idea/misc.xml deleted file mode 100644 index 13515d1..0000000 --- a/.idea/misc.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - \ No newline at end of file From ccce5b55717e8a196cf426881f033de2ac77b7fd Mon Sep 17 00:00:00 2001 From: Gustavo Freze Date: Thu, 14 May 2026 01:10:55 -0300 Subject: [PATCH 3/3] chore: Update CI configuration with database environment variables for testing. --- .github/workflows/ci.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 81df43a..20d4018 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -85,6 +85,12 @@ jobs: needs: [load-config, auto-review] runs-on: ubuntu-latest timeout-minutes: 15 + env: + DATABASE_HOST: outbox-test-db + DATABASE_NAME: outbox_test + DATABASE_USER: outbox + DATABASE_PASSWORD: outbox + TEST_DB_HOST_PORT: '33306' steps: - name: Checkout uses: actions/checkout@v4