From 2370c4df05b34e2312a64aca16800f59e8937106 Mon Sep 17 00:00:00 2001 From: Benoit Travers Date: Tue, 6 Oct 2026 14:52:31 +0200 Subject: [PATCH 1/2] feat!: require an identity field on every aggregate An aggregate root is what other aggregates reference and what a repository loads, so `Entity.aggregate` now rejects a field map with no `identity` field: a compile error (an inline rejection literal) plus a declaration-time defect, the redeclaration precedent. `sameIdentityAs` is therefore always available on an aggregate. `Entity` keeps identity optional, since it also models aggregate parts and value-like records. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/aggregate-requires-identity.md | 12 +++++++++ CLAUDE.md | 5 +++- .../how-to/model-an-event-driven-aggregate.md | 4 ++- docs/reference/declaration.md | 6 +++++ docs/typedoc.json | 1 + packages/entity/src/aggregate.spec.ts | 13 +++++++++ packages/entity/src/aggregate.test-d.ts | 7 +++++ packages/entity/src/aggregate.ts | 27 ++++++++++++++++++- 8 files changed, 72 insertions(+), 3 deletions(-) create mode 100644 .changeset/aggregate-requires-identity.md diff --git a/.changeset/aggregate-requires-identity.md b/.changeset/aggregate-requires-identity.md new file mode 100644 index 0000000..afe40eb --- /dev/null +++ b/.changeset/aggregate-requires-identity.md @@ -0,0 +1,12 @@ +--- +"@btravstack/entity": minor +--- + +**Breaking:** `Entity.aggregate` now requires at least one field flagged +`identity: true`. An aggregate root is what other aggregates reference and what +a repository loads, so a field map without one is a compile error, and a +declaration that gets past the types throws while it runs. `sameIdentityAs` is +therefore always available on an aggregate. + +To migrate, flag the aggregate's id: `id: Entity.field(SomeId, { identity: true })`. +`Entity` is unchanged: identity stays optional there. diff --git a/CLAUDE.md b/CLAUDE.md index 3c58ca4..a6a4f35 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -169,7 +169,10 @@ what they own: map still being inferred and every fresh literal widened (measured: `NoInfer`, a constrained return parameter and an intersection-free signature all widened). `AggregateStatic` has no `_zod` in its type, so an aggregate cannot - be nested as a field. + be nested as a field. An aggregate **must** flag an `identity` field: a + compile error on the field map (an inline rejection literal, not a named + alias, for TypeDoc) plus a declaration-time defect, the redeclaration + precedent. `Entity` keeps identity optional on purpose. - **`freeze.ts`** — `deepFreeze`, the runtime half of immutability. Freezes and recurses into arrays and plain objects, freezes `Date` as a leaf, and deliberately leaves `Map`/`Set`/class instances alone. Which _fields_ to skip diff --git a/docs/how-to/model-an-event-driven-aggregate.md b/docs/how-to/model-an-event-driven-aggregate.md index 2b282d3..7748941 100644 --- a/docs/how-to/model-an-event-driven-aggregate.md +++ b/docs/how-to/model-an-event-driven-aggregate.md @@ -105,7 +105,9 @@ handler returning `status: "ACTIVE"` is checked against the enum, and a missing field is a compile error rather than a surprise at runtime. `invariants`, `computed` and the `identity` and `immutable` flags work as on any -entity. What an aggregate does not have is `update`, the factories, +entity, with one difference: an aggregate must flag an `identity` field, since +a root is what others reference and what a repository loads. Leaving it out +does not compile, and `sameIdentityAs` is therefore always available. What an aggregate does not have is `update`, the factories, `createInput` and `updateInput`. ## Decide with emit diff --git a/docs/reference/declaration.md b/docs/reference/declaration.md index e9a7c55..10611ff 100644 --- a/docs/reference/declaration.md +++ b/docs/reference/declaration.md @@ -445,6 +445,12 @@ Fields and options are separate calls so the fields are fixed before the handlers are checked: a handler's returned literals and fields are checked exactly, where with both in one call they widened. +The fields must flag at least one `identity` field. An aggregate root is what +other aggregates reference and what a repository loads, so a field map without +one is a compile error, and a declaration that gets past the types throws +while it runs. (An `Entity` keeps identity optional: it also models parts of an +aggregate and value-like records that have none.) + The class has `make`, `inspect`, `start` and `replay` as statics, and `toJSON`, `sameIdentityAs` and `emit` on its instances. It has no `update`, no factories, no `createInput` and no `updateInput`, and it cannot be nested as another diff --git a/docs/typedoc.json b/docs/typedoc.json index 428fdb3..81b8b9c 100644 --- a/docs/typedoc.json +++ b/docs/typedoc.json @@ -32,6 +32,7 @@ "Flags", "GeneratedKeys", "Generators", + "IdentityKeys", "ImmutableKeys", "InputOf", "InspectionSrc", diff --git a/packages/entity/src/aggregate.spec.ts b/packages/entity/src/aggregate.spec.ts index ec12949..790f903 100644 --- a/packages/entity/src/aggregate.spec.ts +++ b/packages/entity/src/aggregate.spec.ts @@ -204,3 +204,16 @@ test("a stream that breaks today's invariant is an InvalidEntity, like make", () ]); expect(issues).toEqual([[[], "TOO_MANY_ITEMS"]]); }); + +test("an aggregate without an identity field is refused while the declaration runs", () => { + const declare = Entity.aggregate("Anonymous") as unknown as ( + fields: object, + ) => (options: object) => unknown; + expect(() => + declare({ status: z.enum(["open"]) })({ + events: CartEvent, + opens: { CartOpened: () => ({ status: "open" }) }, + evolve: { ItemAdded: (r: object) => r, CartCheckedOut: (r: object) => r }, + }), + ).toThrow(/Anonymous: an aggregate root needs an identity/u); +}); diff --git a/packages/entity/src/aggregate.test-d.ts b/packages/entity/src/aggregate.test-d.ts index 60fa192..4463a3e 100644 --- a/packages/entity/src/aggregate.test-d.ts +++ b/packages/entity/src/aggregate.test-d.ts @@ -94,3 +94,10 @@ Entity.aggregate("Incomplete")({ id: Entity.field(Id, { identity: true }), statu opens: { Opened: (e) => ({ id: e.id }) }, evolve: { Renamed: (r) => r, Closed: (r) => r }, }); + +// @ts-expect-error an aggregate root has an identity: it is how others reference it and how it is loaded +Entity.aggregate("Anonymous")({ label: Label })({ + events: Event, + opens: { Opened: () => ({ label: "x" }) }, + evolve: { Renamed: (r) => r, Closed: (r) => r }, +}); diff --git a/packages/entity/src/aggregate.ts b/packages/entity/src/aggregate.ts index 6a61328..75b3415 100644 --- a/packages/entity/src/aggregate.ts +++ b/packages/entity/src/aggregate.ts @@ -4,6 +4,7 @@ import type { z } from "zod"; import type { ComputedField } from "./computed.js"; import { InvalidEntity } from "./errors.js"; +import { isFieldSpec } from "./field.js"; import type { Invariant } from "./invariant.js"; import { keysOf, renderIssue } from "./issues.js"; import type { OnlyNominal } from "./shape.js"; @@ -12,6 +13,7 @@ import type { EventNamed, Events, Fields, + IdentityKeys, InputOf, RecordOf, Schemas, @@ -47,7 +49,17 @@ type Maker = { make: (state: unknown) => Result }; export const createAggregate = (buildEntity: BuildEntityClass) => (tag: Tag) => - (fields: S & OnlyNominal) => + ( + // An aggregate root has an identity: it is what other aggregates reference + // and what a repository loads by. Without one, the field map is rejected + // here — an inline literal, so TypeDoc has no internal name to report — + // and `sameIdentityAs` is always callable on an aggregate. + fields: S & + OnlyNominal & + ([IdentityKeys] extends [never] + ? { readonly __anAggregateRootNeedsAnIdentityField: never } + : unknown), + ) => < Ev extends Events, O extends z.output["type"], @@ -67,6 +79,19 @@ export const createAggregate = readonly computed?: { [K in keyof A]: ComputedField> }; readonly invariants?: readonly Invariant>[]; }): AggregateStatic => { + // The same rule at runtime, for a caller the types did not reach — the + // precedent is a root's redeclared field: a compile error, and a defect + // while the declaration is on the stack. + if ( + !Object.values(fields as Fields).some( + (v) => isFieldSpec(v) && (v.flags as { identity?: true }).identity === true, + ) + ) { + // oxlint-disable-next-line unthrown/no-throw + throw new Error( + `${tag}: an aggregate root needs an identity — flag at least one field \`identity: true\`.`, + ); + } const { events, opens, evolve, ...entityOptions } = options; const Base = buildEntity(tag)(fields as Fields, entityOptions) as Record & { readonly prototype: Record; From 1af25eee57e905aab052c99c89da8885ccddae01 Mon Sep 17 00:00:00 2001 From: Benoit Travers Date: Tue, 6 Oct 2026 15:01:11 +0200 Subject: [PATCH 2/2] feat!: let an aggregate's decision carry its expected version A repository's `save` needed the decision plus a version threaded by hand from the load, and chaining two commands then saving the last decision silently dropped the first one's events. Each aggregate instance now remembers, beside it rather than on it, the version it was loaded at and every event decided since. `make` requires `{ version }`, `replay` uses the stream length, `start` is 0, and `emit` returns all pending events plus `expectedVersion`. `save(decision)` needs nothing else; a chain saves as one decision; reusing an already-saved state is a conflict, never an overwrite. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/aggregate-carries-version.md | 20 +++++ CLAUDE.md | 9 +- .../how-to/model-an-event-driven-aggregate.md | 78 +++++++++++------ docs/reference/entry-points.md | 16 +++- .../src/subscriptions.spec.ts | 72 +++++++++++----- .../billing-persistence/src/subscriptions.ts | 47 +++++------ packages/entity/src/aggregate.spec.ts | 45 +++++++++- packages/entity/src/aggregate.test-d.ts | 8 ++ packages/entity/src/aggregate.ts | 84 ++++++++++++++++--- packages/entity/src/types.ts | 29 +++++-- 10 files changed, 308 insertions(+), 100 deletions(-) create mode 100644 .changeset/aggregate-carries-version.md diff --git a/.changeset/aggregate-carries-version.md b/.changeset/aggregate-carries-version.md new file mode 100644 index 0000000..96ce345 --- /dev/null +++ b/.changeset/aggregate-carries-version.md @@ -0,0 +1,20 @@ +--- +"@btravstack/entity": minor +--- + +**Breaking:** an aggregate's decision now carries everything a repository +needs to save it. + +- `Entity.Decision` gains `expectedVersion`, the version the store must still + be at, and its `events` now hold **every** event decided since the aggregate + was loaded. A chain of commands saves as one decision without losing the + earlier events; reusing an already-saved state yields a conflict, never an + overwrite. +- An aggregate's `make` requires the version: `make(row, { version })`. + `replay` takes the stream's length, and `start` is version `0`. +- The version is kept beside the instance, not on it: no reserved field name, + nothing in `toJSON()`, and the package never interprets the number. + +To migrate, pass `{ version }` to an aggregate's `make`, and drop the separate +version argument from a repository's `save`: read `decision.expectedVersion` +instead. `Entity`'s `make` is unchanged. diff --git a/CLAUDE.md b/CLAUDE.md index a6a4f35..13662ec 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -172,7 +172,14 @@ what they own: be nested as a field. An aggregate **must** flag an `identity` field: a compile error on the field map (an inline rejection literal, not a named alias, for TypeDoc) plus a declaration-time defect, the redeclaration - precedent. `Entity` keeps identity optional on purpose. + precedent. `Entity` keeps identity optional on purpose. **A decision is + self-sufficient for a save**: each instance's unsaved version and pending + events live in a module `WeakMap` (`unsaved`) — never on the instance, so no + field name is reserved and nothing reaches `toJSON()`. `make(row, { version +})` requires the version, `replay` uses the stream length, `start` is `0`, + and `emit` returns _all_ pending events plus `expectedVersion`. Chained + commands therefore save as one decision; reusing an already-saved state is + a conflict, never an overwrite. - **`freeze.ts`** — `deepFreeze`, the runtime half of immutability. Freezes and recurses into arrays and plain objects, freezes `Date` as a leaf, and deliberately leaves `Map`/`Set`/class instances alone. Which _fields_ to skip diff --git a/docs/how-to/model-an-event-driven-aggregate.md b/docs/how-to/model-an-event-driven-aggregate.md index 7748941..cf6eeeb 100644 --- a/docs/how-to/model-an-event-driven-aggregate.md +++ b/docs/how-to/model-an-event-driven-aggregate.md @@ -167,59 +167,87 @@ to it. ## Persist the state, or the events -A decision holds both halves, so the repository picks one. The domain code +A decision holds both halves, and the version the store must still be at, so +a repository's `save` takes the decision and nothing else. The domain code above does not change. -**State-based.** Store `decision.state.toJSON()` with a version, and write -`decision.events` to an outbox in the same transaction. Load with `make`. +**State-based.** Load with `make(row, { version })`. Save `decision.state.toJSON()` +if the row is still at `decision.expectedVersion`, and write `decision.events` +to an outbox in the same transaction. ```ts -save(decision: Decision, expected: number): Result { - const { id } = decision.state; - const current = this.#rows.get(id)?.version ?? 0; - if (current !== expected) return Err(new ConcurrentModification({ id, expected })); - this.#rows.set(id, { state: decision.state.toJSON(), version: current + 1 }); +load(id: string) { + const row = this.#rows.get(id); + if (row === undefined) return Err(new SubscriptionNotFound({ id })); + return Subscription.make(row.state, { version: row.version }); +} + +save(decision: Decision): Result { + const { state, expectedVersion: expected } = decision; + const current = this.#rows.get(state.id)?.version ?? 0; + if (current !== expected) return Err(new ConcurrentModification({ id: state.id, expected })); + this.#rows.set(state.id, { state: state.toJSON(), version: current + 1 }); this.outbox.push(...decision.events); return Ok(current + 1); } ``` -**Event-sourced.** Append `decision.events` to the stream if it is still at the -version the command read. Load with `replay`. +**Event-sourced.** Load with `replay`, which takes the stream's length as the +version. Append `decision.events` if the stream is still that long. ```ts -load(id: string) { - const stream = this.#streams.get(id); - if (stream === undefined) return Err(new SubscriptionNotFound({ id })); - return Subscription.replay(stream).map((subscription) => ({ - subscription, - version: stream.length, - })); +save(decision: Decision): Result { + const { state, expectedVersion: expected } = decision; + const stream = this.#streams.get(state.id) ?? []; + if (stream.length !== expected) return Err(new ConcurrentModification({ id: state.id, expected })); + this.#streams.set(state.id, [...stream, ...decision.events]); + return Ok(stream.length + decision.events.length); } ``` -A use case written against the port runs on either: +A use case written against the port runs on either, and has no version to +thread from the load to the save: ```ts export const changeSeats = (repository: SubscriptionRepository) => (id: string, seats: number) => repository .load(id) - .flatMap(({ subscription, version }) => - subscription - .changeSeats(seats) - .flatMap((decision) => repository.save(decision, version)), - ); + .flatMap((subscription) => subscription.changeSeats(seats)) + .flatMap((decision) => repository.save(decision)); ``` +## What a decision remembers + +An aggregate remembers two things it has not saved: the version it was loaded +at, and every event decided since. `make` and `replay` start with nothing +pending; `start` starts at version `0` with its opening event. Each `emit` +returns a decision holding **all** the pending events, and its state carries +them forward. + +That makes a chain of commands one save: the last decision of +`start(…).state.changeSeats(5).state.cancel(at)` holds all three events, at +version `0`. Saving only the last decision of a chain loses nothing. + +It also makes reusing a saved state safe. A state whose decision was already +saved still says it was loaded at the old version, so its next decision is a +`ConcurrentModification`, never an overwrite. Reload after saving. + +The version lives beside the instance, not on it: it is not a field, not in +`toJSON()`, and reserves no name. The package carries the number and never +interprets it, so it can be a row counter or a stream position. + Version checks, the outbox and the event store all stay in the adapter: the aggregate does no I/O. [Persist an aggregate relationally](/how-to/persist-relationally) does the state-based half against a real database. ## Rehydrate without emitting -`make` (a snapshot or a state row) and `replay` (a stream) both return an -aggregate and emit nothing. +`make` (a snapshot or a state row, with its version) and `replay` (a stream) +both return an aggregate with nothing pending, and emit nothing. A `make` +without a version is a compile error, and a defect if forced past the types: +an aggregate that does not know its version could not tell a repository what +to expect. `replay` treats a stored stream as untrusted input, the way `make` treats a row: every event is parsed against the declared union, and a bad one is an diff --git a/docs/reference/entry-points.md b/docs/reference/entry-points.md index 6c77557..a7a0076 100644 --- a/docs/reference/entry-points.md +++ b/docs/reference/entry-points.md @@ -156,8 +156,9 @@ that breaks the aggregate is a bug in it. What an aggregate's command returns after its business checks. Parses each event against the declared union, folds the events onto the current state with the `evolve` handlers, and verifies the result with `make`. The `Decision` -holds the parsed events and the verified state; the source aggregate is -unchanged. A creation event does not compile here: an aggregate that exists +holds the verified state, **every** event decided since the aggregate was +loaded (so a chain of commands saves as one decision), and `expectedVersion`, +the version the store must still be at. The source aggregate is unchanged. A creation event does not compile here: an aggregate that exists cannot be created again, which mirrors `start` accepting nothing else. An event that breaks an invariant, fails its schema, or reaches a handler that @@ -165,6 +166,13 @@ throws is a **defect**, never an `Err`: a decision that cannot hold is a bug in the command, and nothing about it should be persisted. Only `emit` and `start` build a `Decision`; a hand-written `{ state, events }` does not compile. +## `SomeAggregate.make(row, { version })` → `Result` {#someaggregate-make} + +An aggregate's `make` takes the version the row or snapshot was loaded at, and +requires it: the aggregate's next decision tells the repository to expect that +version. Otherwise it is the entity's `make`. A missing version does not +compile, and is a defect if forced past the types. + ## `SomeAggregate.replay(events)` → `Result` {#someaggregate-replay} A stored stream → the aggregate, emitting nothing. The stream is untrusted, so @@ -172,8 +180,8 @@ every event is parsed; a bad one is an `InvalidEntity` whose issue path starts with its index (`[3, "seats"]`). The first event must be a creation event, and a creation event later in the stream is refused at its index. The fold ends in `make`, which is strict: a stream breaking a rule added since is an -`InvalidEntity`, as a row would be. Upcast old event versions before calling -it. +`InvalidEntity`, as a row would be. The stream's length becomes the +aggregate's version. Upcast old event versions before calling it. ## `entity.update(patch)` → `Result` {#entity-update-patch-result-someentity-invalidentity} diff --git a/examples/billing-persistence/src/subscriptions.spec.ts b/examples/billing-persistence/src/subscriptions.spec.ts index 0de01be..eb6b9da 100644 --- a/examples/billing-persistence/src/subscriptions.spec.ts +++ b/examples/billing-persistence/src/subscriptions.spec.ts @@ -20,44 +20,72 @@ for (const [style, repository] of styles) { test(`${style}: a started, changed subscription loads back in the state it was decided in`, () => { const repo = repository(); const started = startSubscription(organizationId, 3).get(); - expect(repo.save(started, 0)).toBeOkWith(1); + expect(repo.save(started)).toBeOk(); expect(changeSeats(repo)(started.state.id, 5)).toBeOk(); - const { subscription, version } = repo.load(started.state.id).getOrThrow(); + const subscription = repo.load(started.state.id).getOrThrow(); expect(subscription.seats).toBe(5); expect(subscription.sameIdentityAs(started.state)).toBe(true); - expect(version).toBe(2); }); - test(`${style}: two saves from the same version, one wins and one is a conflict`, () => { + test(`${style}: two saves from the same load, one wins and one is a conflict`, () => { const repo = repository(); const started = startSubscription(organizationId, 3).get(); - repo.save(started, 0).getOrThrow(); + repo.save(started).getOrThrow(); const mine = repo.load(started.state.id).getOrThrow(); const theirs = repo.load(started.state.id).getOrThrow(); - const a = mine.subscription.changeSeats(4).getOrThrow(); - const b = theirs.subscription.changeSeats(6).getOrThrow(); - expect(repo.save(a, mine.version)).toBeOkWith(2); - expect(repo.save(b, theirs.version)).toBeErrTagged("ConcurrentModification"); + expect(repo.save(mine.changeSeats(4).getOrThrow())).toBeOk(); + expect(repo.save(theirs.changeSeats(6).getOrThrow())).toBeErrTagged("ConcurrentModification"); }); test(`${style}: a refused command saves nothing`, () => { const repo = repository(); const started = startSubscription(organizationId, 3).get(); - repo.save(started, 0).getOrThrow(); + repo.save(started).getOrThrow(); expect(changeSeats(repo)(started.state.id, 3)).toBeErrTagged("SeatsUnchanged"); - expect(repo.load(started.state.id).getOrThrow().version).toBe(1); + expect(repo.load(started.state.id).getOrThrow().seats).toBe(3); + }); + + test(`${style}: chained commands save as one decision, losing none of their events`, () => { + const repo = repository(); + // start, change and cancel before anything is stored: the last decision holds all three + const decision = startSubscription(organizationId, 3) + .get() + .state.changeSeats(5) + .getOrThrow() + .state.cancel("2026-10-06T09:00:00.000Z") + .getOrThrow(); + expect(decision.events.map((e) => e.type)).toEqual([ + "SubscriptionStarted", + "SeatsChanged", + "SubscriptionCancelled", + ]); + repo.save(decision).getOrThrow(); + + const reloaded = repo.load(decision.state.id).getOrThrow(); + expect(reloaded.seats).toBe(5); + expect(reloaded.status).toBe("CANCELLED"); + }); + + test(`${style}: deciding again from an already-saved state is a conflict, never an overwrite`, () => { + const repo = repository(); + const started = startSubscription(organizationId, 3).get(); + repo.save(started).getOrThrow(); + // `started.state` still says "nothing stored yet": a reload is the way forward + expect(repo.save(started.state.changeSeats(4).getOrThrow())).toBeErrTagged( + "ConcurrentModification", + ); }); } test("state-based: the decision's events reach the outbox with the state", () => { const repo = new StateBasedSubscriptions(); const started = startSubscription(organizationId, 3).get(); - repo.save(started, 0).getOrThrow(); + repo.save(started).getOrThrow(); changeSeats(repo)(started.state.id, 5).getOrThrow(); expect(repo.outbox.map((row) => (row as { event: { type: string } }).event.type)).toEqual([ @@ -66,18 +94,18 @@ test("state-based: the decision's events reach the outbox with the state", () => ]); }); -test("event-sourced and state-based stores agree on the state, from the same decisions", () => { +test("event-sourced and state-based stores agree on the state, from the same decision", () => { const states = new StateBasedSubscriptions(); const events = new EventSourcedSubscriptions(); - const started = startSubscription(organizationId, 3).get(); - const cancelled = started.state.cancel("2026-10-06T09:00:00.000Z").getOrThrow(); - for (const repo of [states, events]) { - repo.save(started, 0).getOrThrow(); - repo.save(cancelled, 1).getOrThrow(); - } - - const fromRow = states.load(started.state.id).getOrThrow().subscription; - const fromStream = events.load(started.state.id).getOrThrow().subscription; + const cancelled = startSubscription(organizationId, 3) + .get() + .state.cancel("2026-10-06T09:00:00.000Z") + .getOrThrow(); + states.save(cancelled).getOrThrow(); + events.save(cancelled).getOrThrow(); + + const fromRow = states.load(cancelled.state.id).getOrThrow(); + const fromStream = events.load(cancelled.state.id).getOrThrow(); expect(fromStream.toJSON()).toEqual(fromRow.toJSON()); expect(fromStream.status).toBe("CANCELLED"); }); diff --git a/examples/billing-persistence/src/subscriptions.ts b/examples/billing-persistence/src/subscriptions.ts index 042adcd..449edac 100644 --- a/examples/billing-persistence/src/subscriptions.ts +++ b/examples/billing-persistence/src/subscriptions.ts @@ -7,9 +7,14 @@ * * - **state-based**: the state's `toJSON()` and a version, with the events * written to an outbox in the same atomic step (#37 does this against a - * real database). Loading is `make`. + * real database). Loading is `make(row, { version })`. * - **event-sourced**: the events, appended to the subscription's stream if it - * is still at the version the command read. Loading is `replay`. + * is still at the version the command read. Loading is `replay`, which + * takes the stream's length as the version. + * + * The decision carries the version its aggregate was loaded at, so `save` + * takes the decision and nothing else: there is no version to forget or mix + * up between the load and the save. * * Both implement one port, and `changeSeats` below runs against either: the * switch is infrastructure, not a domain rewrite. Both stores are in memory @@ -38,13 +43,10 @@ export class ConcurrentModification extends TaggedError("ConcurrentModification" override message = `subscription ${this.id} changed since version ${this.expected}`; } -/** A loaded aggregate and the version a save must still find. */ -export type Loaded = { readonly subscription: Subscription; readonly version: number }; - export type SubscriptionRepository = { - load(id: string): Result; - /** `expected` is the version the command read; `0` for a new subscription. Returns the new version. */ - save(decision: Decision, expected: number): Result; + load(id: string): Result; + /** Saves everything the decision holds, if the store is still at `decision.expectedVersion`. */ + save(decision: Decision): Result; }; const stored = (value: T): unknown => JSON.parse(JSON.stringify(value)); @@ -54,17 +56,15 @@ export class StateBasedSubscriptions implements SubscriptionRepository { readonly #rows = new Map(); readonly outbox: unknown[] = []; - load(id: string): Result { + load(id: string): Result { const row = this.#rows.get(id); if (row === undefined) return Err(new SubscriptionNotFound({ id })); - return Subscription.make(row.state).map((subscription) => ({ - subscription, - version: row.version, - })); + return Subscription.make(row.state, { version: row.version }); } - save(decision: Decision, expected: number): Result { - const { id } = decision.state; + save(decision: Decision): Result { + const { state, expectedVersion: expected } = decision; + const { id } = state; const current = this.#rows.get(id)?.version ?? 0; if (current !== expected) return Err(new ConcurrentModification({ id, expected })); // One synchronous block stands in for one transaction: the row and the @@ -80,17 +80,15 @@ export class StateBasedSubscriptions implements SubscriptionRepository { export class EventSourcedSubscriptions implements SubscriptionRepository { readonly #streams = new Map(); - load(id: string): Result { + load(id: string): Result { const stream = this.#streams.get(id); if (stream === undefined) return Err(new SubscriptionNotFound({ id })); - return Subscription.replay(stream).map((subscription) => ({ - subscription, - version: stream.length, - })); + return Subscription.replay(stream); } - save(decision: Decision, expected: number): Result { - const { id } = decision.state; + save(decision: Decision): Result { + const { state, expectedVersion: expected } = decision; + const { id } = state; const stream = this.#streams.get(id) ?? []; if (stream.length !== expected) return Err(new ConcurrentModification({ id, expected })); const next = [...stream, ...decision.events.map(stored)]; @@ -115,6 +113,5 @@ export const changeSeats = > => repository .load(id) - .flatMap(({ subscription, version }) => - subscription.changeSeats(seats).flatMap((decision) => repository.save(decision, version)), - ); + .flatMap((subscription) => subscription.changeSeats(seats)) + .flatMap((decision) => repository.save(decision)); diff --git a/packages/entity/src/aggregate.spec.ts b/packages/entity/src/aggregate.spec.ts index 790f903..bf49121 100644 --- a/packages/entity/src/aggregate.spec.ts +++ b/packages/entity/src/aggregate.spec.ts @@ -57,6 +57,8 @@ class Cart extends Entity.aggregate("Cart")({ const id = "0199b1f4-1b1e-7000-8000-000000000000"; const opened = (): Cart => Cart.start({ type: "CartOpened", cartId: id }).get().state; +/** An open cart as a repository hands it over: stored at version 1, nothing pending. */ +const loaded = (): Cart => Cart.make(opened().toJSON(), { version: 1 }).getOrThrow(); /** The outcome of a decision as one comparable value. */ const channel = (r: Result) => @@ -73,10 +75,12 @@ test("start opens an aggregate from a creation event and returns a sealed decisi expect(decision.state).toBeInstanceOf(Cart); expect(decision.state.toJSON()).toEqual({ id, status: "open", items: [] }); expect(decision.events).toEqual([{ type: "CartOpened", cartId: id }]); + // nothing is stored yet + expect(decision.expectedVersion).toBe(0); }); test("a method folds its events onto the current state and verifies the result once", () => { - const cart = opened(); + const cart = loaded(); const decision = cart.addItem("apple", 2).getOrThrow(); expect(decision.events).toEqual([{ type: "ItemAdded", productId: "apple", quantity: 2 }]); expect(decision.state.items).toEqual([{ productId: "apple", quantity: 2 }]); @@ -144,7 +148,9 @@ test("start refuses a non-opening event as a defect", () => { test("replay folds a stored stream into the same state, and emits nothing", () => { const first = Cart.start({ type: "CartOpened", cartId: id }).get(); const second = first.state.addItem("apple", 2).getOrThrow(); - const stream = [...first.events, ...second.events]; + // the last decision already holds every event since creation + const stream = second.events; + expect(stream.map((e) => e.type)).toEqual(["CartOpened", "ItemAdded"]); const replayed = Cart.replay(stream).getOrThrow(); expect(replayed).toBeInstanceOf(Cart); @@ -154,7 +160,7 @@ test("replay folds a stored stream into the same state, and emits nothing", () = test("make rehydrates a snapshot, and an aggregate has no update and no factory", () => { const snapshot = opened().addItem("apple", 1).getOrThrow().state.toJSON(); - const cart = Cart.make(snapshot).getOrThrow(); + const cart = Cart.make(snapshot, { version: 2 }).getOrThrow(); expect(cart.items).toHaveLength(1); expect("update" in cart).toBe(false); expect("factory" in Cart).toBe(false); @@ -217,3 +223,36 @@ test("an aggregate without an identity field is refused while the declaration ru }), ).toThrow(/Anonymous: an aggregate root needs an identity/u); }); + +/* ── The version a decision must still find ────────────────────────── */ + +test("a decision carries the version its state was loaded at", () => { + const row = opened().toJSON(); + expect( + Cart.make(row, { version: 7 }).getOrThrow().addItem("a", 1).getOrThrow().expectedVersion, + ).toBe(7); + + const stream = Cart.start({ type: "CartOpened", cartId: id }).get().events; + const replayed = Cart.replay([...stream, { type: "ItemAdded", productId: "a", quantity: 1 }]); + // a stream's version is its length + expect(replayed.getOrThrow().checkOut().get().expectedVersion).toBe(2); +}); + +test("chained commands accumulate every event since the load, so saving one decision loses none", () => { + const first = loaded().addItem("a", 1).getOrThrow(); + const second = first.state.addItem("b", 1).getOrThrow(); + expect(second.events.map((e) => e.type)).toEqual(["ItemAdded", "ItemAdded"]); + expect(second.expectedVersion).toBe(1); + + // a brand-new aggregate's decisions start from its opening event + const fresh = opened().addItem("a", 1).getOrThrow(); + expect(fresh.events.map((e) => e.type)).toEqual(["CartOpened", "ItemAdded"]); + expect(fresh.expectedVersion).toBe(0); +}); + +test("make without a version is a defect: a loaded aggregate must say what it was loaded at", () => { + const row = opened().toJSON(); + expect(channel((Cart.make as (s: unknown) => Result)(row))).toBe( + "defect", + ); +}); diff --git a/packages/entity/src/aggregate.test-d.ts b/packages/entity/src/aggregate.test-d.ts index 4463a3e..ead6107 100644 --- a/packages/entity/src/aggregate.test-d.ts +++ b/packages/entity/src/aggregate.test-d.ts @@ -101,3 +101,11 @@ Entity.aggregate("Anonymous")({ label: Label })({ opens: { Opened: () => ({ label: "x" }) }, evolve: { Renamed: (r) => r, Closed: (r) => r }, }); + +// @ts-expect-error a loaded aggregate says which version it was loaded at +Doc.make({ id: "0199b1f4-1b1e-7000-8000-000000000000" }); +Doc.make({ id: "0199b1f4-1b1e-7000-8000-000000000000" }, { version: 3 }); + +// the decision names the version a repository must still find +const expected: number = decision.expectedVersion; +void expected; diff --git a/packages/entity/src/aggregate.ts b/packages/entity/src/aggregate.ts index 75b3415..3f8c35e 100644 --- a/packages/entity/src/aggregate.ts +++ b/packages/entity/src/aggregate.ts @@ -25,8 +25,19 @@ export type BuildEntityClass = ( ) => (fields: Fields, options?: Record) => object; type Event = { readonly type: string }; + +/** + * What an aggregate instance has not saved yet: the version its state was + * loaded at, and every event decided since. Kept beside the instance rather + * than on it, so no field name is reserved and nothing reaches `toJSON()`. + * Every instance an aggregate hands out has an entry: `make` and `replay` set + * one with nothing pending, `start` and `emit` carry one forward. + */ +const unsaved = new WeakMap< + object, + { readonly version: number; readonly pending: readonly Event[] } +>(); type Handlers = Record unknown>; -type Maker = { make: (state: unknown) => Result }; /** * `class X extends Entity.aggregate("X")(fields)({ events, opens, evolve })` @@ -98,6 +109,10 @@ export const createAggregate = readonly input: z.ZodObject; }; + // The entity's own `make`, kept before the aggregate's replaces it: the + // verification step of every fold, which records no version. + const verify = Base["make"] as (this: object, state: unknown) => Result; + // State changes only through events, and `start` is creation. Reflect.deleteProperty(Base.prototype, "update"); Reflect.deleteProperty(Base, "factory"); @@ -123,15 +138,29 @@ export const createAggregate = const bug = (detail: string) => new Error(`${tag}: ${detail}`); - /** The decision: the verified state, and the parsed events that produced it. */ - const decide = (Ctor: Maker, record: unknown, decided: readonly Event[]) => - Ctor.make(record) + /** + * The decision: the verified state, every event since the load, and the + * version the store must still be at. The new state inherits both, so a + * chained command's decision still saves the earlier events. + */ + const decide = ( + Ctor: object, + record: unknown, + decided: readonly Event[], + from: { readonly version: number; readonly pending: readonly Event[] }, + ) => + verify + .call(Ctor, record) .mapErrCases((m, defect) => m.with(P.tag("InvalidEntity"), (invalid) => defect(bug(`the decision breaks the aggregate: ${invalid.message}`)), ), ) - .map((state) => Object.freeze({ state, events: Object.freeze([...decided]) })); + .map((state) => { + const pending = Object.freeze([...from.pending, ...decided]); + unsaved.set(state, { version: from.version, pending }); + return Object.freeze({ state, events: pending, expectedVersion: from.version }); + }); /** Events from domain code: a schema failure is a bug in the command, so a defect. */ const parseDecided = (raw: readonly unknown[]) => @@ -159,16 +188,22 @@ export const createAggregate = }, record); function emit(this: object, ...raw: unknown[]) { - const Ctor = this.constructor as unknown as Maker; + const from = unsaved.get(this); return parseDecided(raw).flatMap((decided) => fromThrowable( - () => fold(recordOf(this), decided), + () => { + if (from === undefined) { + // oxlint-disable-next-line unthrown/no-throw + throw bug("an aggregate built outside make, replay, start and emit cannot decide"); + } + return fold(recordOf(this), decided); + }, (cause, defect) => defect(cause), - )().flatMap((record) => decide(Ctor, record, decided)), + )().flatMap((record) => decide(this.constructor, record, decided, from!)), ); } - function start(this: Maker, raw: unknown) { + function start(this: object, raw: unknown) { return parseDecided([raw]).flatMap((decided) => { const [event] = decided as [Event]; return fromThrowable( @@ -181,7 +216,24 @@ export const createAggregate = return (opener as (e: Event) => unknown)(event); }, (cause, defect) => defect(cause), - )().flatMap((record) => decide(this, record, decided)); + )().flatMap((record) => decide(this, record, decided, { version: 0, pending: [] })); + }); + } + + /** + * A snapshot or a state row → the aggregate, with the version it was loaded + * at. A missing or malformed version is a defect: an aggregate that does + * not know its version would hand a repository a decision it cannot check. + */ + function make(this: object, state: unknown, loaded?: { readonly version?: unknown }) { + return verify.call(this, state).map((instance) => { + const version = loaded?.version; + if (typeof version !== "number" || !Number.isInteger(version) || version < 0) { + // oxlint-disable-next-line unthrown/no-throw + throw bug("make needs the version the row was loaded at: make(row, { version })"); + } + unsaved.set(instance, { version, pending: [] }); + return instance; }); } @@ -194,7 +246,7 @@ export const createAggregate = * adapter's job, before this. The final `make` is strict: a stream that * breaks a rule added since is refused like a row would be (#71). */ - function replay(this: Maker, raw: unknown): Result { + function replay(this: object, raw: unknown): Result { if (!Array.isArray(raw) || raw.length === 0) { return invalid([{ message: "a stream is a non-empty array of events" }]); } @@ -231,14 +283,20 @@ export const createAggregate = return fromThrowable( () => fold((openers[opening.type] as (e: Event) => unknown)(opening), rest), (cause, defect) => defect(cause), - )().flatMap((record) => this.make(record)); + )() + .flatMap((record) => verify.call(this, record)) + .map((state) => { + // a stream's version is its length + unsaved.set(state, { version: raw.length, pending: [] }); + return state; + }); }) as Result; } for (const [key, value] of Object.entries({ emit })) { Object.defineProperty(Base.prototype, key, { value, writable: true, configurable: true }); } - for (const [key, value] of Object.entries({ start, replay, events })) { + for (const [key, value] of Object.entries({ make, start, replay, events })) { Object.defineProperty(Base, key, { value, writable: true, configurable: true }); } diff --git a/packages/entity/src/types.ts b/packages/entity/src/types.ts index 86fe862..c4f125b 100644 --- a/packages/entity/src/types.ts +++ b/packages/entity/src/types.ts @@ -623,15 +623,22 @@ export declare class DecisionKey { } /** - * What a command returns: the events it decided, and the state those events - * produce, already verified by `make`. Only `emit` and `start` build one, so a - * repository that takes a `Decision` can only be handed events that were - * folded and checked against every invariant. Persist `state.toJSON()`, - * `events`, or both — the decision is the same either way. + * What a command returns: the state, already verified by `make`; every event + * decided since the aggregate was loaded; and the version it was loaded at. + * Only `emit` and `start` build one, so a repository that takes a `Decision` + * can only be handed events that were folded and checked against every + * invariant — and it needs nothing else to save one. + * + * `events` accumulates across chained commands, so saving the last decision + * of a chain loses none of the earlier ones. `expectedVersion` is the one the + * store must still be at: `0` for a new aggregate, the stream length after + * `replay`, the row's version after `make`. The package carries it and never + * interprets it. */ export type Decision = { readonly state: A; readonly events: readonly E[]; + readonly expectedVersion: number; // The property name is the diagnostic, as with `Sealed`. readonly __onlyEmitOrStartMakeADecision: DecisionKey; }; @@ -705,8 +712,16 @@ export type AggregateStatic< /** the event union, read by `Entity.Event` */ readonly __event: z.output; readonly __instance: ConstructedAggregate; - /** a snapshot or a state-based row → aggregate; emits nothing */ - make(this: new (d: Sealed>) => T, state: unknown): Result; + /** + * a snapshot or a state-based row → aggregate; emits nothing. The version is + * required: it is what the aggregate's next decision tells the store to + * expect, so an aggregate can never be saved without one. + */ + make( + this: new (d: Sealed>) => T, + state: unknown, + loaded: { readonly version: number }, + ): Result; inspect(state: unknown): Result>, InvalidEntity>; /** an opening event → the decision that creates the aggregate */ start(