Skip to content
This repository was archived by the owner on Oct 7, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions .changeset/aggregate-carries-version.md
Original file line number Diff line number Diff line change
@@ -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.
12 changes: 12 additions & 0 deletions .changeset/aggregate-requires-identity.md
Original file line number Diff line number Diff line change
@@ -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.
12 changes: 11 additions & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
82 changes: 56 additions & 26 deletions docs/how-to/model-an-event-driven-aggregate.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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<number, ConcurrentModification> {
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<number, ConcurrentModification> {
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<number, ConcurrentModification> {
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
Expand Down
6 changes: 6 additions & 0 deletions docs/reference/declaration.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
16 changes: 12 additions & 4 deletions docs/reference/entry-points.md
Original file line number Diff line number Diff line change
Expand Up @@ -156,24 +156,32 @@ 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
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, InvalidEntity>` {#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, InvalidEntity>` {#someaggregate-replay}

A stored stream → the aggregate, emitting nothing. The stream is untrusted, so
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<SomeEntity, InvalidEntity>` {#entity-update-patch-result-someentity-invalidentity}

Expand Down
1 change: 1 addition & 0 deletions docs/typedoc.json
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
"Flags",
"GeneratedKeys",
"Generators",
"IdentityKeys",
"ImmutableKeys",
"InputOf",
"InspectionSrc",
Expand Down
72 changes: 50 additions & 22 deletions examples/billing-persistence/src/subscriptions.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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([
Expand All @@ -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");
});
Loading
Loading