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/.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..13662ec 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -169,7 +169,17 @@ 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. **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 2b282d3..cf6eeeb 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 @@ -165,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/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/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/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/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 ec12949..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); @@ -204,3 +210,49 @@ 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); +}); + +/* ── 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 60fa192..ead6107 100644 --- a/packages/entity/src/aggregate.test-d.ts +++ b/packages/entity/src/aggregate.test-d.ts @@ -94,3 +94,18 @@ 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 }, +}); + +// @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 6a61328..3f8c35e 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, @@ -23,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 })` @@ -47,7 +60,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,12 +90,29 @@ 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; 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"); @@ -98,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[]) => @@ -134,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( @@ -156,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; }); } @@ -169,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" }]); } @@ -206,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(