From 701f93219f91c74cc04b573866d0faa41ea33e8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sat, 31 Oct 2026 12:36:09 +0100 Subject: [PATCH] feat(core): seal personal data in scheduled commands and dead letters, cancelled by forget A command's payload takes personal() and person() as an event's does. Wherever the command is stored, scheduled with delay, retried or dead-lettered, its marked fields are sealed with the key of the person it names, and the handler still receives them as the schema gives them. forget cancels the person's scheduled commands and removes their dead letters in its transaction, and PersonalDataForgotten names them. --- .changeset/personal-data.md | 2 + .../forgetting-in-an-immutable-log.md | 8 +- docs/src/content/docs/guides/dead-letters.md | 6 +- docs/src/content/docs/guides/personal-data.md | 55 +- docs/src/content/docs/reference/errors.md | 3 +- .../src/content/docs/reference/limitations.md | 4 - packages/cli/src/generate/emit/plus-types.ts | 2 +- packages/core/src/adapter/deferred-writes.ts | 7 +- packages/core/src/adapter/sql/dead-letters.ts | 13 + packages/core/src/adapter/sql/index.ts | 1 + .../src/adapter/sqlite/dead-letter-store.ts | 18 +- packages/core/src/adapter/sqlite/scheduler.ts | 48 +- packages/core/src/adapter/sqlite/schema.ts | 6 +- .../src/adapter/storage/dead-letter-store.ts | 12 + .../src/adapter/storage/personal-key-store.ts | 8 + .../core/src/adapter/storage/scheduler.ts | 16 +- .../testing/dead-letter-store.contract.ts | 35 ++ .../src/adapter/testing/scheduler.contract.ts | 58 ++- .../testing/storage-transaction.contract.ts | 63 +++ .../src/kernel/aggregate/build-aggregates.ts | 33 +- packages/core/src/kernel/aggregate/runtime.ts | 4 + packages/core/src/kernel/command/pipeline.ts | 30 +- .../src/kernel/dead-letters/dead-letters.ts | 36 +- .../src/kernel/personal-data/commands.test.ts | 474 ++++++++++++++++++ .../core/src/kernel/personal-data/commands.ts | 121 +++++ .../src/kernel/personal-data/forget.test.ts | 60 ++- .../core/src/kernel/personal-data/forget.ts | 33 +- .../src/kernel/personal-data/marks.test.ts | 2 +- .../core/src/kernel/personal-data/marks.ts | 8 +- packages/core/src/kernel/scheduler/worker.ts | 18 +- packages/core/src/memory/dead-letter-store.ts | 18 +- packages/core/src/memory/scheduler.ts | 23 +- packages/core/src/memory/transaction.ts | 57 ++- packages/core/src/modules/bounda-events.ts | 21 +- packages/core/src/modules/command.ts | 4 +- packages/core/src/modules/index.ts | 5 + packages/core/src/modules/payload.ts | 15 +- packages/core/src/modules/personal.ts | 30 +- .../domain/order/commands/+types/pay-order.ts | 2 +- .../order/commands/+types/place-order.ts | 2 +- .../commands/+types/register-customer.ts | 2 +- .../customer/commands/register-customer.ts | 4 +- .../order/commands/+types/cancel-order.ts | 2 +- .../domain/order/commands/+types/pay-order.ts | 2 +- .../order/commands/+types/place-order.ts | 2 +- packages/core/test-types/inference.test-d.ts | 37 +- packages/core/test-types/negatives.test-d.ts | 9 +- packages/postgresql/src/dead-letter-store.ts | 14 +- packages/postgresql/src/scheduler.ts | 41 +- packages/postgresql/src/schema.ts | 10 +- 50 files changed, 1331 insertions(+), 153 deletions(-) create mode 100644 packages/core/src/adapter/sql/dead-letters.ts create mode 100644 packages/core/src/kernel/personal-data/commands.test.ts create mode 100644 packages/core/src/kernel/personal-data/commands.ts diff --git a/.changeset/personal-data.md b/.changeset/personal-data.md index 7caf1f68..95a9233b 100644 --- a/.changeset/personal-data.md +++ b/.changeset/personal-data.md @@ -8,3 +8,5 @@ --- Personal data in events, forgotten by crypto-shredding. An event's `payload` receives `personal` and `person` next to `z`: `personal(schema)` marks a field as personal data, stored encrypted (AES-256-GCM through WebCrypto, on Node and in `workerd`) with a key of the person the event names, `person(schema)` marks the field that names them, and without it the aggregate's id does. `app.forget(personId)`, on every `BoundaClient`, deletes the key and appends `PersonalDataForgotten` to the stream `bounda:`, in one transaction: from then on every personal field of theirs reads as `null`, typed `T | null` in every handler. A read model that projects an event with personal fields must handle it from `projections/bounda/personal-data-forgotten.ts`, which the generator checks in `.bounda/types.ts`; a policy may react to it from `policies/bounda/`. The adapters keep the keys, with their fingerprints, in a new `personal_keys` table, `Storage` and `StorageTransaction` gain `personalKeys`, and the storage contracts cover it. The global stream shows the fields as stored, encrypted. A command that loaded a person's data before `forget` and appends after it fails with `ConcurrencyError` and is decided again, instead of writing their data back under the new key another command may have given them meanwhile: the state remembers the key each person's data was read under, and `AppendArgs` gains `persons` (`{ person, fingerprint }`) for this, which a store ignores. Snapshots and remembered rejections name the same and are written only while that key is still the person's; in PostgreSQL a GIN index on their `persons` lets `forget` find them without reading every row. + +A command's `payload` receives `personal` and `person` too (`CommandPayloadArgs`), for the commands that carry personal data: wherever the command is stored, scheduled with `delay`, retried or dead-lettered, its marked fields are encrypted with the key of the person it names, and its handler still receives them as the schema gives them, never `null`, since a command whose person is forgotten never runs. In the same transaction, `forget` cancels the person's scheduled commands and removes their dead letters, and `PersonalDataForgotten` names them in `cancelledCommands` and `removedDeadLetters`. A run already under way when `forget` comes writes nothing, dead letter included, and one sealed under a key its person no longer holds is dropped unrun. `app.deadLetters` shows a scheduled command's payload opened and refuses to retry one whose person was forgotten. `ScheduledCommand`, `ScheduleArgs` and `DeadLetter` gain `persons`, `Scheduler` and `DeadLetterStore` gain `forget(person)`, both tables a `persons` column (with a GIN index in PostgreSQL), and the storage contracts cover them. diff --git a/docs/src/content/docs/concepts/forgetting-in-an-immutable-log.md b/docs/src/content/docs/concepts/forgetting-in-an-immutable-log.md index 5c2e4dcd..44441b80 100644 --- a/docs/src/content/docs/concepts/forgetting-in-an-immutable-log.md +++ b/docs/src/content/docs/concepts/forgetting-in-an-immutable-log.md @@ -130,11 +130,9 @@ have work to do when it happens. systems a policy wrote to. The key store is a table like the others: a backup of the whole database restores the keys too, so a backup kept past a person's erasure is itself a copy of their data. -- **Scheduled commands and dead letters.** A command scheduled with `delay` is stored with its - payload, in the clear, until it runs, and a dead-lettered scheduled command keeps it. Keep - personal data out of delayed commands, or carry an id and resolve it when the command runs. - A dead letter also keeps the error a handler threw, message and stack: one that quotes a - decrypted field keeps it in the clear. +- **Errors.** A dead letter keeps the error a handler threw, message and stack: one that quotes a + decrypted field keeps it in the clear. A command's payload marked `personal` is encrypted where + it is stored, scheduled or dead-lettered, and `forget` cancels it; a payload left unmarked is not. - **Encrypted data under the GDPR.** Whether a ciphertext without a key still counts as personal data is a legal question Verraes raises and this page does not answer. diff --git a/docs/src/content/docs/guides/dead-letters.md b/docs/src/content/docs/guides/dead-letters.md index ab6548c3..e5a55e79 100644 --- a/docs/src/content/docs/guides/dead-letters.md +++ b/docs/src/content/docs/guides/dead-letters.md @@ -34,7 +34,7 @@ What a retry runs, and when it marks the letter, depends on the kind: | Policy | The handler again, for the stored event, with the event's correlation. The inbox is bypassed on purpose: it already says the handler ran, and you are asking for another run. | In the same transaction as what the run writes | | Process | The handler again for the stored event, with the instance's current state, or for a deadline's letter (`deadline:`) the handler of that deadline, with a new `idempotencyKey`; an event that completes the process completes it. Then the events parked behind the failure, in order, and the process is `started` again, its deadlines scheduled at their moments (one already past runs at once); see [a failed process](#a-failed-process). | Once the instance has drained what was parked; a retry cut short is taken up by retrying the same letter | | Follow-up of a timed-out process | Its handler and nothing else: the process stays timed out, and a handler a deploy removed lets the event through. | In the same transaction as what the run writes | -| Scheduled command | The dropped command again, with the payload the letter recorded. A rejection settles the letter too: it is the aggregate's answer, logged as `command rejected`. | In the same transaction as what the run writes | +| Scheduled command | The dropped command again, with the payload the letter recorded, its [personal fields](/guides/personal-data/#in-a-commands-payload) opened. A rejection settles the letter too: it is the aggregate's answer, logged as `command rejected`. | In the same transaction as what the run writes | Two retries of one letter, or a retry and a discard, never both settle it, even when they run at once: whichever gets there second is refused with `DeadLetterSettledError` @@ -43,8 +43,8 @@ is refused, so a double click does not store a command's decision twice, though have run in both; a process's retry has already handled its events by then. A letter the app as it now is cannot retry is refused with `DeadLetterNotRetriableError` (`DEAD_LETTER_NOT_RETRIABLE`) without running anything: its policy, process or scheduled command is gone from the registry, its -policy or process no longer handles the event, or its instance failed on another step whose letter -comes first. +policy or process no longer handles the event, its instance failed on another step whose letter +comes first, or the person whose data its scheduled command holds was forgotten since. The same operations are on the app as `app.deadLetters` — `list`, `count`, `get`, `retry` and `discard` — for a script or an admin route. diff --git a/docs/src/content/docs/guides/personal-data.md b/docs/src/content/docs/guides/personal-data.md index 533aa08f..38c283b1 100644 --- a/docs/src/content/docs/guides/personal-data.md +++ b/docs/src/content/docs/guides/personal-data.md @@ -45,9 +45,30 @@ export const payload = ({ z, personal, person }: Event.PayloadArgs) => }); ``` -An event names at most one person. Only events have the marks: a command's or a query's -`PayloadArgs` has no `personal`, since commands are not stored and nothing of theirs has to be -forgotten. +An event names at most one person. + +## In a command's payload + +A command is stored too when it waits: dispatched with `delay`, retried by the scheduler after a +failure, or dead-lettered once it gives up. A command that carries personal data marks its payload +the same way, with `personal` and `person` from its own `PayloadArgs`: + +```ts +// app/domain/customer/commands/remind-customer.ts +import type { Command } from "./+types/remind-customer"; + +export const payload = ({ z, personal }: Command.PayloadArgs) => + z.object({ customerId: z.string(), email: personal(z.email()) }); +``` + +Wherever the command is stored, the marked fields are encrypted with the key of the person it +names, the aggregate's id or the field marked `person`, and [`forget`](#forgetting-a-person) +cancels it. A command dispatched at once is never stored, so the marks change nothing for it. The +handler reads `command.payload.email` as a `string`, never `null`: a command whose person is +forgotten never runs. A query's `PayloadArgs` has no `personal`, since a query is never stored. + +A command scheduled before its fields were marked keeps its payload as it was written, in the +clear, and runs with it; `forget` cannot find it. ## What the types say @@ -69,9 +90,12 @@ await app.catchUpReadModels({ through: forgotten }); ``` `forget` is on every `BoundaClient`: the app, the Durable Object through `connect`, the `bounda` -context of React Router. In one transaction it deletes the person's key and appends -`PersonalDataForgotten`, with `{ personId }` as payload, to the stream `bounda:`: the -runtime's own aggregate, `bounda`, has one stream per person it forgot. It resolves like a +context of React Router. In one transaction it deletes the person's key, cancels the scheduled +commands that hold their data, removes the dead letters of those that gave up, and appends +`PersonalDataForgotten` to the stream `bounda:`: the runtime's own aggregate, `bounda`, +has one stream per person it forgot. Its payload is `{ personId, cancelledCommands, +removedDeadLetters }`, the ids and types of what went, the only record of commands that never +became events. It resolves like a command, with the event's position, so `catchUpReadModels({ through })` waits for the read models that handle it, and the Durable Object's `forget` waits for them itself unless `consistency` is `"eventual"`. A person without a key is forgotten all the same, since a read @@ -86,6 +110,13 @@ person's data was read under, and an append that would seal their fields under a with [`ConcurrencyError`](/reference/errors/), as on a stream that moved. The retry decides on the state without their data. +A scheduled command already running when `forget` comes writes nothing: its claim is gone, so +neither its events nor a dead letter are stored, though its handler may have run. One that a +`forget` misses, stored in the instant between reading the key and the key going, is sealed under +a key no one holds: it is dropped when it comes due, unrun, and the worker logs `scheduled command +dropped: its person was forgotten`. The [dead letters](/guides/dead-letters/) list shows a +scheduled command's payload with its fields opened. + A [snapshot](/concepts/long-streams/#snapshots) holds the state in the clear, so the same transaction removes every snapshot whose state carried a field of theirs; each one names the people it holds and the key it read each one under, and none is written for a person who no @@ -98,14 +129,14 @@ what it answered, and a rejection's message may quote the state it was decided o rejection names the people whose fields were in that state, with their keys as a snapshot does, and `forget` removes it too: the same request sent again is decided again, on the state without them. What it cannot see is personal -data that came in the command's payload, which is never marked: keep that out of a rejection's +data that came in the command's payload, marked or not: keep that out of a rejection's message and say what went wrong with its code and the ids involved. -Command payloads are never sealed anywhere, so `forget` does not reach what holds one: a command -scheduled with `delay` keeps its payload until it runs, a -[dead letter](/guides/dead-letters/) of a scheduled command keeps it until it is retried or -discarded, and your logs keep whatever you wrote to them. Pass ids in those payloads, not the -data itself: the handler reads the email from the state, where it is sealed and forgotten. +What `forget` does not reach is text your code writes and data left unmarked: your logs keep +whatever you wrote to them, the message of an error a handler throws stays in its dead letter and, +for a scheduled command, in the `ScheduledCommandFailed` event on the aggregate's stream, and a +command's payload left unmarked is stored as it came. Log and throw with ids, not the data itself, +and mark every field of a command's payload that holds it. ## Handling it in a read model diff --git a/docs/src/content/docs/reference/errors.md b/docs/src/content/docs/reference/errors.md index b456064d..c207d551 100644 --- a/docs/src/content/docs/reference/errors.md +++ b/docs/src/content/docs/reference/errors.md @@ -79,7 +79,8 @@ dead letter with `errorType: "retriable_exhausted"`. retried or discarded after another retry or discard had settled it. - **`DEAD_LETTER_NOT_RETRIABLE`** (`DeadLetterNotRetriableError`; an operator's answer). A dead letter cannot be retried in the app as it is now: its handler is gone or no longer handles the - event, or its process failed on another step first. The letter stays `failed`. + event, its process failed on another step first, or the person whose data its scheduled command + holds was forgotten since. The letter stays `failed`. - **`REBUILD_SUPERSEDED`** (`RebuildSupersededError`; an operator's answer). Another rebuild of the same read model took over; this one stopped without writing. `readModel`. - **`SNAPSHOT_FAILED`** (`SnapshotError`; only in an app from `createTestApp`). A diff --git a/docs/src/content/docs/reference/limitations.md b/docs/src/content/docs/reference/limitations.md index 9ce89ee6..f877505d 100644 --- a/docs/src/content/docs/reference/limitations.md +++ b/docs/src/content/docs/reference/limitations.md @@ -21,10 +21,6 @@ get today. Each item says why, so nobody discovers it the hard way: - **Notifications for scheduled commands.** The worker that runs due commands polls at `pollInterval`; only the event dispatcher is woken by `NOTIFY`. See [Tuning](/guides/deployment/#tuning). -- **Personal data outside events.** A field marked `personal` is encrypted in the event store and - nowhere else: a command scheduled with `delay` keeps its payload in the clear until it runs, - and so does the dead letter of one that gave up. Keep personal data out of delayed commands. - See [Forgetting in an immutable log](/concepts/forgetting-in-an-immutable-log/#what-stays-outside). What a production app does get, and where it is explained: [rebuilding a read model](/guides/deployment/#rebuilding-a-read-model) without taking it offline, diff --git a/packages/cli/src/generate/emit/plus-types.ts b/packages/cli/src/generate/emit/plus-types.ts index 410cf5bf..79be2457 100644 --- a/packages/cli/src/generate/emit/plus-types.ts +++ b/packages/cli/src/generate/emit/plus-types.ts @@ -101,7 +101,7 @@ const commandFiles = (aggregate: AggregateModel, typesPath: string): GeneratedFi imports: { generated: true, module: command.path }, namespace: "Command", members: [ - ["PayloadArgs", "core.PayloadArgs"], + ["PayloadArgs", "core.CommandPayloadArgs"], [ "RejectionsArgs", generic("core.CommandRejectionsArgs", [ diff --git a/packages/core/src/adapter/deferred-writes.ts b/packages/core/src/adapter/deferred-writes.ts index f59f1eee..46754f4e 100644 --- a/packages/core/src/adapter/deferred-writes.ts +++ b/packages/core/src/adapter/deferred-writes.ts @@ -27,8 +27,9 @@ export interface DeferredWriteTarget { * every write in order, and `flush` runs the writes recorded so far against another set of the * same stores, a transaction's or the live ones. `add` answers the letter as filed, whatever the * store holds under that id already; `get` on the idempotency store answers a record `put` - * recorded before anything `live` holds. `tryClaim`, `claimDue`, both `renew` and `purge` act at - * once, on `live`, since their answer decides what happens next. + * recorded before anything `live` holds. `tryClaim`, `claimDue`, both `renew`, `purge` and the + * `forget` of the scheduler and the dead letter store act at once, on `live`, since their answer + * decides what happens next. */ export interface DeferredWrites { readonly stores: DeferredStores; @@ -81,6 +82,7 @@ export const deferWrites: DeferWritesFunction = ({ target.deadLetterStore.updateStatus(id, status), ), remove: later((target, id) => target.deadLetterStore.remove(id)), + forget: deadLetterStore.forget, }, scheduler: { claimDue: scheduler.claimDue, @@ -92,6 +94,7 @@ export const deferWrites: DeferWritesFunction = ({ fail: later((target, args) => target.scheduler.fail(args)), defer: later((target, args) => target.scheduler.defer(args)), renew: scheduler.renew, + forget: scheduler.forget, }, idempotencyStore: { get: async (args) => records.get(args.key) ?? idempotencyStore.get(args), diff --git a/packages/core/src/adapter/sql/dead-letters.ts b/packages/core/src/adapter/sql/dead-letters.ts new file mode 100644 index 00000000..18e1d959 --- /dev/null +++ b/packages/core/src/adapter/sql/dead-letters.ts @@ -0,0 +1,13 @@ +import type { DeadLetter } from "../storage/dead-letter-store.ts"; +import { codePointOrder } from "./order.ts"; + +export interface ByFirstFailureFunction { + (a: DeadLetter, b: DeadLetter): number; +} + +/** + * The order of `DeadLetterStore.list` and `DeadLetterStore.forget`: `firstFailedAt`, then `id` by + * code point. For a store whose engine hands rows back unordered, as `RETURNING` does. + */ +export const byFirstFailure: ByFirstFailureFunction = (a, b) => + codePointOrder(a.firstFailedAt, b.firstFailedAt) || codePointOrder(a.id, b.id); diff --git a/packages/core/src/adapter/sql/index.ts b/packages/core/src/adapter/sql/index.ts index 085e8b22..f735dcde 100644 --- a/packages/core/src/adapter/sql/index.ts +++ b/packages/core/src/adapter/sql/index.ts @@ -1,4 +1,5 @@ export type { SqlDatabase, SqlTransaction } from "./database.ts"; +export { byFirstFailure } from "./dead-letters.ts"; export { postgresqlDialect } from "./dialect.ts"; export { quoteIdentifier, storageTableNameFor, tableNameFor } from "./identifiers.ts"; export type { ExistingColumn } from "./read-model-schema.ts"; diff --git a/packages/core/src/adapter/sqlite/dead-letter-store.ts b/packages/core/src/adapter/sqlite/dead-letter-store.ts index ba4fc0cd..b2ad4780 100644 --- a/packages/core/src/adapter/sqlite/dead-letter-store.ts +++ b/packages/core/src/adapter/sqlite/dead-letter-store.ts @@ -5,9 +5,12 @@ import type { DeadLetterKind, DeadLetterStatus, DeadLetterStore, + KeyedPerson, ListDeadLettersArgs, } from "../index.ts"; import type { SqlDatabase } from "../sql/database.ts"; +import { byFirstFailure } from "../sql/dead-letters.ts"; +import { naming } from "./held-keys.ts"; export interface CreateSqliteDeadLetterStoreArgs { readonly db: SqlDatabase; @@ -19,7 +22,12 @@ export interface CreateSqliteDeadLetterStoreFunction { } const COLUMNS = - '"id", "kind", "handler", "event_id", "event_type", "aggregate_type", "aggregate_id", "error_type", "error_message", "error_stack", "attempts", "first_failed_at", "last_failed_at", "status", "payload"'; + '"id", "kind", "handler", "event_id", "event_type", "aggregate_type", "aggregate_id", "error_type", "error_message", "error_stack", "attempts", "first_failed_at", "last_failed_at", "status", "payload", "persons"'; + +const personsOf = (row: Record): { persons?: KeyedPerson[] } => { + const persons = JSON.parse(String(row.persons)) as KeyedPerson[]; + return persons.length === 0 ? {} : { persons }; +}; const toLetter = (row: Record): DeadLetter => ({ id: String(row.id), @@ -41,6 +49,7 @@ const toLetter = (row: Record): DeadLetter => ({ ...(row.payload === null || row.payload === undefined ? {} : { payload: JSON.parse(String(row.payload)) as unknown }), + ...personsOf(row), }); const filters = ( @@ -70,7 +79,7 @@ export const createSqliteDeadLetterStore: CreateSqliteDeadLetterStoreFunction = return { add: async (letter) => { await db.run( - `INSERT INTO ${table} (${COLUMNS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'failed', ?) ON CONFLICT ("id") DO NOTHING`, + `INSERT INTO ${table} (${COLUMNS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'failed', ?, ?) ON CONFLICT ("id") DO NOTHING`, [ letter.id, letter.kind, @@ -86,6 +95,7 @@ export const createSqliteDeadLetterStore: CreateSqliteDeadLetterStoreFunction = letter.firstFailedAt, letter.lastFailedAt, letter.payload === undefined ? null : JSON.stringify(letter.payload), + JSON.stringify(letter.persons ?? []), ], ); return (await get(letter.id)) as DeadLetter; @@ -115,5 +125,9 @@ export const createSqliteDeadLetterStore: CreateSqliteDeadLetterStoreFunction = if (updated.length === 0) throw new DeadLetterSettledError({ id }); }, remove: (id) => db.run(`DELETE FROM ${table} WHERE "id" = ?`, [id]), + forget: async (person) => + (await db.all(`DELETE FROM ${table} WHERE ${naming(table)} RETURNING ${COLUMNS}`, [person])) + .map(toLetter) + .sort(byFirstFailure), }; }; diff --git a/packages/core/src/adapter/sqlite/scheduler.ts b/packages/core/src/adapter/sqlite/scheduler.ts index 466c0dfb..d7f20743 100644 --- a/packages/core/src/adapter/sqlite/scheduler.ts +++ b/packages/core/src/adapter/sqlite/scheduler.ts @@ -1,8 +1,15 @@ import { ScheduledClaimLostError } from "../../contracts/errors.ts"; import type { CausationContext } from "../../contracts/metadata.ts"; -import type { ClaimedCommand, ScheduledClaim, ScheduledCommand, Scheduler } from "../index.ts"; +import type { + ClaimedCommand, + KeyedPerson, + ScheduledClaim, + ScheduledCommand, + Scheduler, +} from "../index.ts"; import type { SqlDatabase } from "../sql/database.ts"; import { byExecuteAt, earliestDue } from "../sql/index.ts"; +import { naming } from "./held-keys.ts"; export interface CreateSqliteSchedulerArgs { readonly db: SqlDatabase; @@ -14,19 +21,23 @@ export interface CreateSqliteSchedulerFunction { } const COLUMNS = - '"dedupe_key", "command_type", "aggregate_id", "payload", "execute_at", "context", "attempts"'; + '"dedupe_key", "command_type", "aggregate_id", "payload", "execute_at", "context", "attempts", "persons"'; -const toScheduled = (row: Record): ScheduledCommand => ({ - dedupeKey: String(row.dedupe_key), - command: { - type: String(row.command_type), - aggregateId: String(row.aggregate_id), - payload: JSON.parse(String(row.payload)) as unknown, - }, - executeAt: String(row.execute_at), - context: JSON.parse(String(row.context)) as CausationContext, - attempts: Number(row.attempts), -}); +const toScheduled = (row: Record): ScheduledCommand => { + const persons = JSON.parse(String(row.persons)) as KeyedPerson[]; + return { + dedupeKey: String(row.dedupe_key), + command: { + type: String(row.command_type), + aggregateId: String(row.aggregate_id), + payload: JSON.parse(String(row.payload)) as unknown, + }, + executeAt: String(row.execute_at), + context: JSON.parse(String(row.context)) as CausationContext, + attempts: Number(row.attempts), + ...(persons.length === 0 ? {} : { persons }), + }; +}; const toClaimed = (row: Record): ClaimedCommand => ({ ...toScheduled(row), @@ -59,15 +70,16 @@ export const createSqliteScheduler: CreateSqliteSchedulerFunction = ({ db, table ); return { - schedule: ({ dedupeKey, command, executeAt, context, keepTimingOfSameCommand }) => + schedule: ({ dedupeKey, command, executeAt, context, keepTimingOfSameCommand, persons = [] }) => db.run( - `INSERT INTO ${table} (${COLUMNS}, "claimed_at", "last_error", "revision") VALUES (?, ?, ?, ?, ?, ?, 0, NULL, NULL, 0) + `INSERT INTO ${table} (${COLUMNS}, "claimed_at", "last_error", "revision") VALUES (?, ?, ?, ?, ?, ?, 0, ?, NULL, NULL, 0) ON CONFLICT ("dedupe_key") DO UPDATE SET "command_type" = excluded."command_type", "aggregate_id" = excluded."aggregate_id", "payload" = excluded."payload", "execute_at" = excluded."execute_at", "context" = excluded."context", + "persons" = excluded."persons", "attempts" = 0, "last_error" = NULL, "revision" = ${table}."revision" + 1 @@ -77,6 +89,7 @@ export const createSqliteScheduler: CreateSqliteSchedulerFunction = ({ db, table AND ${table}."payload" = excluded."payload" ${keepTimingOfSameCommand === true ? "" : `AND ${table}."execute_at" = excluded."execute_at"`} AND ${table}."context" = excluded."context" + AND ${table}."persons" = excluded."persons" )`, [ dedupeKey, @@ -85,6 +98,7 @@ export const createSqliteScheduler: CreateSqliteSchedulerFunction = ({ db, table JSON.stringify(command.payload), executeAt.toISOString(), JSON.stringify(context), + JSON.stringify(persons), ], ), cancel: (dedupeKey) => db.run(`DELETE FROM ${table} WHERE "dedupe_key" = ?`, [dedupeKey]), @@ -153,5 +167,9 @@ export const createSqliteScheduler: CreateSqliteSchedulerFunction = ({ db, table [limit ?? -1, offset], ) ).map(toScheduled), + forget: async (person) => + (await db.all(`DELETE FROM ${table} WHERE ${naming(table)} RETURNING ${COLUMNS}`, [person])) + .map(toScheduled) + .sort(byExecuteAt), }; }; diff --git a/packages/core/src/adapter/sqlite/schema.ts b/packages/core/src/adapter/sqlite/schema.ts index 64e90bf0..642abab8 100644 --- a/packages/core/src/adapter/sqlite/schema.ts +++ b/packages/core/src/adapter/sqlite/schema.ts @@ -89,7 +89,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "first_failed_at" TEXT NOT NULL, "last_failed_at" TEXT NOT NULL, "status" TEXT NOT NULL, - "payload" TEXT + "payload" TEXT, + "persons" TEXT NOT NULL DEFAULT '[]' )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.deadLetters, "status")} ON ${tables.deadLetters} ("status")`, `CREATE TABLE IF NOT EXISTS ${tables.scheduledCommands} ( @@ -103,7 +104,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "claimed_at" TEXT, "last_error" TEXT, "revision" INTEGER NOT NULL DEFAULT 0, - "claim_id" TEXT + "claim_id" TEXT, + "persons" TEXT NOT NULL DEFAULT '[]' )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.scheduledCommands, "execute_at")} ON ${tables.scheduledCommands} ("execute_at")`, `CREATE TABLE IF NOT EXISTS ${tables.idempotencyKeys} ( diff --git a/packages/core/src/adapter/storage/dead-letter-store.ts b/packages/core/src/adapter/storage/dead-letter-store.ts index ad19159e..309f10a8 100644 --- a/packages/core/src/adapter/storage/dead-letter-store.ts +++ b/packages/core/src/adapter/storage/dead-letter-store.ts @@ -1,3 +1,5 @@ +import type { KeyedPerson } from "./personal-key-store.ts"; + /** * What gave up: a policy or process handler (a delayed policy run included), or a scheduled * command, one dispatched with `delay`. @@ -35,6 +37,11 @@ export interface DeadLetter { * Policy and process letters point at a stored event instead. */ readonly payload?: unknown; + /** + * Only on `scheduled` letters whose payload holds personal fields sealed: whose they are, with + * the key each is sealed under, as the scheduled command named them. + */ + readonly persons?: readonly KeyedPerson[]; /** * Process letters only: how many steps wait on the instance. While the letter is `failed`, the * events parked behind it for its retry to handle; on the letter a retry returns, the steps @@ -75,4 +82,9 @@ export interface DeadLetterStore { */ updateStatus(id: string, status: DeadLetterStatus): Promise; remove(id: string): Promise; + /** + * Removes every letter whose `persons` names `person`, whatever its status, and resolves to them + * in the order of `list`. + */ + forget(person: string): Promise; } diff --git a/packages/core/src/adapter/storage/personal-key-store.ts b/packages/core/src/adapter/storage/personal-key-store.ts index c3830809..7bec2179 100644 --- a/packages/core/src/adapter/storage/personal-key-store.ts +++ b/packages/core/src/adapter/storage/personal-key-store.ts @@ -26,6 +26,14 @@ export interface KeyedPersonIdFunction { export const keyedPersonId: KeyedPersonIdFunction = ({ person, fingerprint }) => JSON.stringify([person, fingerprint]); +export interface NamesPersonFunction { + (row: { readonly persons?: readonly KeyedPerson[] }, person: string): boolean; +} + +// Under any key: what `forget` removes for a store that filters its rows itself. +export const namesPerson: NamesPersonFunction = (row, person) => + row.persons?.some((held) => held.person === person) ?? false; + /** * The keys the personal fields of events are encrypted with, one per person. `get` is `null` for * a person without one, as for one forgotten. `add` stores `key` for `person` unless one is in diff --git a/packages/core/src/adapter/storage/scheduler.ts b/packages/core/src/adapter/storage/scheduler.ts index 118d8d99..f9e2b2db 100644 --- a/packages/core/src/adapter/storage/scheduler.ts +++ b/packages/core/src/adapter/storage/scheduler.ts @@ -1,9 +1,11 @@ import type { NewCommand } from "../../contracts/command.ts"; import type { CausationContext } from "../../contracts/metadata.ts"; +import type { KeyedPerson } from "./personal-key-store.ts"; /** * A command waiting for its time. `dedupeKey` identifies the entry: scheduling the same key - * again replaces it. + * again replaces it. `persons` names the people whose personal fields the payload holds sealed, + * with the key each is sealed under; absent when it holds none. */ export interface ScheduledCommand { readonly dedupeKey: string; @@ -11,6 +13,7 @@ export interface ScheduledCommand { readonly executeAt: string; readonly context: CausationContext; readonly attempts: number; + readonly persons?: readonly KeyedPerson[]; } /** @@ -43,6 +46,11 @@ export interface ScheduleArgs { * off stays backed off. */ readonly keepTimingOfSameCommand?: boolean; + /** + * The people whose personal fields the payload holds sealed, for `forget` to find the command + * by; none when left out. + */ + readonly persons?: readonly KeyedPerson[]; } export interface ClaimDueArgs { @@ -140,4 +148,10 @@ export interface Scheduler { * Every command, earliest `executeAt` first, then by `dedupeKey`. */ list(args?: ListScheduledArgs): Promise; + /** + * Removes every command whose `persons` names `person`, claimed or not, and resolves to them in + * the order of `list`: a worker running one then finds its claim lost, so what it writes, a dead + * letter included, is rolled back. + */ + forget(person: string): Promise; } diff --git a/packages/core/src/adapter/testing/dead-letter-store.contract.ts b/packages/core/src/adapter/testing/dead-letter-store.contract.ts index 3d91a8d5..2f425934 100644 --- a/packages/core/src/adapter/testing/dead-letter-store.contract.ts +++ b/packages/core/src/adapter/testing/dead-letter-store.contract.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it } from "vitest"; import { DeadLetterSettledError } from "../../contracts/errors.ts"; import type { DeadLetterStore, NewDeadLetter } from "../storage/dead-letter-store.ts"; +import { testKeyedPerson as held } from "./fixtures.ts"; export interface DeadLetterStoreContractArgs { readonly create: () => Promise; @@ -163,6 +164,40 @@ export const deadLetterStoreContract: DeadLetterStoreContractFunction = ({ creat expect(rejected?.reason).toBeInstanceOf(DeadLetterSettledError); }); + it("keeps whose personal data a letter holds, and forgets the ones that name a person, whatever their status", async () => { + const scheduled = (id: string, at: string, persons?: NewDeadLetter["persons"]) => + letter(id, { + kind: "scheduled", + handler: "PlaceOrder", + eventType: "PlaceOrder", + firstFailedAt: at, + payload: { orderId: id }, + ...(persons === undefined ? {} : { persons }), + }); + await store.add(scheduled("a", "2026-01-01T00:02:00.000Z", [held("c-1", "1")])); + await store.add( + scheduled("b", "2026-01-01T00:01:00.000Z", [held("c-2", "2"), held("c-1", "1")]), + ); + await store.add(scheduled("c", "2026-01-01T00:00:00.000Z", [held("c-2", "2")])); + await store.add(scheduled("d", "2026-01-01T00:00:00.000Z")); + await store.add(scheduled("e", "2026-01-01T00:00:00.000Z", [])); + await store.updateStatus("b", "discarded"); + expect((await store.get("a"))?.persons).toEqual([held("c-1", "1")]); + expect(await store.get("d")).not.toHaveProperty("persons"); + expect(await store.get("e")).not.toHaveProperty("persons"); + + const forgotten = await store.forget("c-1"); + expect(forgotten).toEqual([ + { + ...scheduled("b", "2026-01-01T00:01:00.000Z", [held("c-2", "2"), held("c-1", "1")]), + status: "discarded", + }, + { ...scheduled("a", "2026-01-01T00:02:00.000Z", [held("c-1", "1")]), status: "failed" }, + ]); + expect((await store.list()).map((kept) => kept.id)).toEqual(["c", "d", "e"]); + expect(await store.forget("c-1")).toEqual([]); + }); + it("updates status and removes letters", async () => { await store.add(letter("a")); await store.updateStatus("a", "discarded"); diff --git a/packages/core/src/adapter/testing/scheduler.contract.ts b/packages/core/src/adapter/testing/scheduler.contract.ts index b2d26829..49068fc8 100644 --- a/packages/core/src/adapter/testing/scheduler.contract.ts +++ b/packages/core/src/adapter/testing/scheduler.contract.ts @@ -1,7 +1,7 @@ import { beforeEach, describe, expect, it } from "vitest"; import { ScheduledClaimLostError } from "../../contracts/errors.ts"; import type { ClaimedCommand, ScheduleArgs, Scheduler } from "../storage/scheduler.ts"; -import { testCommand, testContext } from "./fixtures.ts"; +import { testKeyedPerson as held, testCommand, testContext } from "./fixtures.ts"; export interface SchedulerContractArgs { readonly create: () => Promise; @@ -541,6 +541,62 @@ export const schedulerContract: SchedulerContractFunction = ({ create }) => { expect(await scheduler.claimDue({ now: at(1), limit: 10, leaseMs: 60_000 })).toHaveLength(1); }); + it("keeps whose personal data a command holds, and forgets the ones that name a person, claimed or not", async () => { + const schedule = (dedupeKey: string, ms: number, persons?: ScheduleArgs["persons"]) => + scheduler.schedule({ + dedupeKey, + command: testCommand(dedupeKey), + executeAt: at(ms), + context: testContext, + ...(persons === undefined ? {} : { persons }), + }); + await schedule("a", 1_000, [held("c-1", "1")]); + await schedule("b", 1_000, [held("c-2", "2")]); + await schedule("c", 1_000); + await schedule("d", 500, [held("c-1", "1"), held("c-3", "3")]); + await schedule("e", 2_000, []); + const claimed = claimOf( + await scheduler.claimDue({ now: at(1_000), limit: 10, leaseMs: 60_000 }), + "a", + ); + expect(claimed.persons).toEqual([held("c-1", "1")]); + const listed = await scheduler.list(); + expect(listed.find((entry) => entry.dedupeKey === "c")).not.toHaveProperty("persons"); + expect(listed.find((entry) => entry.dedupeKey === "e")).not.toHaveProperty("persons"); + + const forgotten = await scheduler.forget("c-1"); + expect(forgotten.map((entry) => entry.dedupeKey)).toEqual(["d", "a"]); + expect(forgotten[0]).toEqual({ + dedupeKey: "d", + command: testCommand("d"), + executeAt: at(500).toISOString(), + context: testContext, + attempts: 0, + persons: [held("c-1", "1"), held("c-3", "3")], + }); + expect((await scheduler.list()).map((entry) => entry.dedupeKey)).toEqual(["b", "c", "e"]); + await expect(scheduler.complete(claimed)).rejects.toBeInstanceOf(ScheduledClaimLostError); + expect(await scheduler.forget("c-1")).toEqual([]); + expect(await scheduler.forget("c")).toEqual([]); + }); + + it("replaces whose personal data a command holds when its key is scheduled again", async () => { + const schedule = (persons?: ScheduleArgs["persons"]) => + scheduler.schedule({ + dedupeKey: "a", + command: testCommand("1"), + executeAt: at(1_000), + context: testContext, + ...(persons === undefined ? {} : { persons }), + }); + await schedule([held("c-1", "1")]); + await schedule([held("c-1", "2")]); + expect((await scheduler.list())[0]?.persons).toEqual([held("c-1", "2")]); + await schedule(); + expect((await scheduler.list())[0]).not.toHaveProperty("persons"); + expect(await scheduler.forget("c-1")).toEqual([]); + }); + it("lists pending commands ordered by execution time with paging", async () => { await scheduler.schedule({ dedupeKey: "b", diff --git a/packages/core/src/adapter/testing/storage-transaction.contract.ts b/packages/core/src/adapter/testing/storage-transaction.contract.ts index 50bde347..323bf15c 100644 --- a/packages/core/src/adapter/testing/storage-transaction.contract.ts +++ b/packages/core/src/adapter/testing/storage-transaction.contract.ts @@ -338,6 +338,69 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ ); }); + it("cancels a person's scheduled commands and removes their dead letters with the rest, or not at all", async () => { + const schedule = (dedupeKey: string, person: string) => + storage.scheduler.schedule({ + dedupeKey, + command: testCommand(person), + executeAt: new Date(now.getTime() + 60_000), + context: testContext, + persons: [held(person, "1")], + }); + await schedule("command:c1", "c-1"); + await schedule("command:c2", "c-2"); + await storage.deadLetterStore.add({ ...testDeadLetter("d1"), persons: [held("c-1", "1")] }); + await storage.deadLetterStore.add({ ...testDeadLetter("d2"), persons: [held("c-2", "1")] }); + const forget = (tx: StorageTransaction) => + Promise.all([tx.scheduler.forget("c-1"), tx.deadLetterStore.forget("c-1")]); + await expect( + storage.transact(async (tx) => { + await forget(tx); + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + expect(await storage.scheduler.list()).toHaveLength(2); + expect(await storage.deadLetterStore.count()).toBe(2); + const [cancelled, removed] = await storage.transact(forget); + expect(cancelled.map((entry) => entry.dedupeKey)).toEqual(["command:c1"]); + expect(removed.map((letter) => letter.id)).toEqual(["d1"]); + expect((await storage.scheduler.list()).map((entry) => entry.dedupeKey)).toEqual([ + "command:c2", + ]); + expect((await storage.deadLetterStore.list()).map((letter) => letter.id)).toEqual(["d2"]); + }); + + it("leaves no dead letter of a person behind when a forget races the worker giving up their command", async () => { + await storage.scheduler.schedule({ + dedupeKey: "command:c1", + command: testCommand("c-1"), + executeAt: now, + context: testContext, + persons: [held("c-1", "1")], + }); + const [claim] = await storage.scheduler.claimDue({ now, limit: 1, leaseMs: 60_000 }); + if (claim === undefined) throw new Error("nothing claimed"); + const [givenUp, forgotten] = await Promise.allSettled([ + storage.transact(async (tx) => { + await tx.deadLetterStore.add({ ...testDeadLetter("d1"), persons: [held("c-1", "1")] }); + await tx.scheduler.fail({ claim, error: "boom" }); + }), + storage.transact(async (tx) => ({ + cancelled: await tx.scheduler.forget("c-1"), + removed: await tx.deadLetterStore.forget("c-1"), + })), + ]); + if (forgotten.status === "rejected") throw forgotten.reason; + if (givenUp.status === "fulfilled") { + expect(forgotten.value.removed.map((letter) => letter.id)).toEqual(["d1"]); + } else { + expect(givenUp.reason).toBeInstanceOf(ScheduledClaimLostError); + expect(forgotten.value.cancelled.map((entry) => entry.dedupeKey)).toEqual(["command:c1"]); + } + expect(await storage.deadLetterStore.count()).toBe(0); + expect(await storage.scheduler.list()).toEqual([]); + }); + it("changes and removes dead letters with the rest, or not at all", async () => { await storage.deadLetterStore.add(testDeadLetter("d1")); await storage.deadLetterStore.add(testDeadLetter("d2")); diff --git a/packages/core/src/kernel/aggregate/build-aggregates.ts b/packages/core/src/kernel/aggregate/build-aggregates.ts index c845402a..bb0a850a 100644 --- a/packages/core/src/kernel/aggregate/build-aggregates.ts +++ b/packages/core/src/kernel/aggregate/build-aggregates.ts @@ -3,7 +3,7 @@ import type { ResolvedConfig } from "../../config/types.ts"; import { ConfigurationError } from "../../contracts/errors.ts"; import { createEventBuilders } from "../../modules/event.ts"; import { capitalize } from "../../modules/naming.ts"; -import type { EventPayloadArgs } from "../../modules/payload.ts"; +import type { CommandPayloadArgs, EventPayloadArgs } from "../../modules/payload.ts"; import { person, personal } from "../../modules/personal.ts"; import type { AggregateEntry, Registry } from "../../modules/registry.ts"; import { personalMarksOf } from "../personal-data/marks.ts"; @@ -16,12 +16,16 @@ import type { EventRuntime, } from "./runtime.ts"; +// One `personal` marks both: only its type tells an event's field, read as `null` once its person +// is forgotten, from a command's, which never runs then. +const schemaArgs = { z, personal, person } as EventPayloadArgs & CommandPayloadArgs; + const compileSchema = ( - payload: ((args: EventPayloadArgs) => unknown) | undefined, + payload: ((args: EventPayloadArgs & CommandPayloadArgs) => unknown) | undefined, path: string, ): z.ZodType | null => { if (payload === undefined) return null; - const schema = payload({ z, personal, person }); + const schema = payload(schemaArgs); if (!(schema instanceof z.ZodType)) { throw new ConfigurationError(`${path}: payload must return a Zod schema`); } @@ -52,16 +56,21 @@ const buildEvents = (name: string, entry: AggregateEntry): Record => Object.fromEntries( - Object.entries(entry.commands).map(([key, command]) => [ - key, - { + Object.entries(entry.commands).map(([key, command]) => { + const path = `aggregates.${name}.commands.${key}`; + const schema = compileSchema(command.module.payload, path); + return [ key, - type: capitalize(key), - schema: compileSchema(command.module.payload, `aggregates.${name}.commands.${key}`), - handler: command.module.handler as CommandRuntime["handler"], - rejections: (command.module.rejections ?? null) as CommandRuntime["rejections"], - }, - ]), + { + key, + type: capitalize(key), + schema, + handler: command.module.handler as CommandRuntime["handler"], + rejections: (command.module.rejections ?? null) as CommandRuntime["rejections"], + personal: personalMarksOf({ schema, path }), + }, + ]; + }), ); const buildSnapshot = ( diff --git a/packages/core/src/kernel/aggregate/runtime.ts b/packages/core/src/kernel/aggregate/runtime.ts index 8cb79963..cd871998 100644 --- a/packages/core/src/kernel/aggregate/runtime.ts +++ b/packages/core/src/kernel/aggregate/runtime.ts @@ -33,6 +33,10 @@ export interface CommandRuntime { readonly rejections: | ((args: Record) => Readonly>) | null; + /** + * `null` for a command without personal fields. + */ + readonly personal: PersonalMarks | null; } /** diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index 5e04e782..f8485f89 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -38,6 +38,7 @@ import type { Logger } from "../../contracts/logger.ts"; import type { CausationContext } from "../../contracts/metadata.ts"; import type { AggregateLoader } from "../aggregate/load-aggregate.ts"; import type { AggregateRuntime, AggregatesRuntime, CommandRuntime } from "../aggregate/runtime.ts"; +import { type SealedPayload, sealCommandPayload } from "../personal-data/commands.ts"; import { errorDetails } from "../shared/retry.ts"; import { withTimeout } from "../shared/timeout.ts"; import { withPorts } from "../shared/with-ports.ts"; @@ -306,9 +307,9 @@ const rejectOf = ( /** * Commands with `delay` are scheduled with the payload as the caller passed it, in the JSON form - * every scheduler stores. Validating that form here only rejects early what would fail when the - * command runs, such as a date JSON turns into a string; the validation whose result the handler - * sees happens then, so a transform does not apply twice. + * every scheduler stores, its personal fields sealed. Validating that form here only rejects early + * what would fail when the command runs, such as a date JSON turns into a string; the validation + * whose result the handler sees happens then, so a transform does not apply twice. */ export const createCommandPipeline: CreateCommandPipelineFunction = ({ aggregates, @@ -476,17 +477,19 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ const schedule = async ( scheduler: Scheduler, delay: DurationInput, + { payload: stored, persons }: SealedPayload, ): Promise => { const executeAt = new Date(clock.now().getTime() + parseDuration(delay)); await scheduler.schedule({ dedupeKey: scheduledCommandKey(commandId), - command: { type, payload: input, aggregateId }, + command: { type, payload: stored, aggregateId }, executeAt, context: { correlationId: command.metadata.correlationId, causationId: command.metadata.causationId, depth, }, + persons, }); return { scheduled: true, @@ -531,20 +534,27 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ // of work, or in the transaction of the append that stores them, so a duplicate running // beside it loses on the key and stores nothing. const perform = async (): Promise => { - if (idempotency === undefined) { - return options.delay === undefined - ? execute(aggregate, runtime, command, stores.eventStore, signals) - : schedule(stores.scheduler, options.delay); - } if (options.delay !== undefined) { const delay = options.delay; + // Before any transaction opens, since the key store it may write to is the storage's own. + const sealed = await sealCommandPayload({ + type, + aggregateId, + payload: input, + marks: runtime.personal, + keys: storage.personalKeys, + }); + if (idempotency === undefined) return schedule(stores.scheduler, delay, sealed); const scheduling = async (transaction: DispatchStores): Promise => { - const answer = await schedule(transaction.scheduler, delay); + const answer = await schedule(transaction.scheduler, delay, sealed); await transaction.idempotencyStore.put(recordOf(idempotency, answer)); return answer; }; return within === undefined ? storage.transact(scheduling) : scheduling(within); } + if (idempotency === undefined) { + return execute(aggregate, runtime, command, stores.eventStore, signals); + } if (within !== undefined) { const answer = await execute(aggregate, runtime, command, within.eventStore, signals); await within.idempotencyStore.put( diff --git a/packages/core/src/kernel/dead-letters/dead-letters.ts b/packages/core/src/kernel/dead-letters/dead-letters.ts index b3be3347..2568b70c 100644 --- a/packages/core/src/kernel/dead-letters/dead-letters.ts +++ b/packages/core/src/kernel/dead-letters/dead-letters.ts @@ -13,6 +13,7 @@ import type { Logger } from "../../contracts/logger.ts"; import type { CausationContext } from "../../contracts/metadata.ts"; import type { AggregatesRuntime } from "../aggregate/runtime.ts"; import type { CommandPipeline } from "../command/pipeline.ts"; +import { openCommandPayload } from "../personal-data/commands.ts"; import type { PoliciesRuntime } from "../policy/build-policies.ts"; import type { PolicyExecutor } from "../policy/executor.ts"; import type { ProcessDeadLetters } from "../process/dead-letter-retry.ts"; @@ -25,12 +26,13 @@ import { commitWork, type UnitOfWork } from "../unit-of-work/unit-of-work.ts"; */ export interface DeadLetters { /** - * The letters of the store, process ones with how many events are `parked` behind them. + * The letters of the store, process ones with how many events are `parked` behind them, and a + * scheduled command's payload with its personal fields opened. */ list(args?: ListDeadLettersArgs): Promise; count(args?: ListDeadLettersArgs): Promise; /** - * One letter of the store, a process one with how many events are `parked` behind it. + * One letter of the store, as `list` gives it. */ get(id: string): Promise; /** @@ -43,7 +45,8 @@ export interface DeadLetters { * counts as retried, as the scheduler would have settled it. Rejects with the handler's error * when it fails again, and the letter stays `failed`. Rejects without running anything with * `NotFoundError` when the letter, or the event or process instance it names, is missing, and - * with `DeadLetterNotRetriableError` when the app as it now is cannot retry it. Rejects with + * with `DeadLetterNotRetriableError` when the app as it now is cannot retry it, or the person + * whose personal fields a scheduled command holds was forgotten since. Rejects with * `DeadLetterSettledError` a letter that is no longer `failed`, or that another retry or a * discard settled while this one ran: a retry marked in its own transaction then writes * nothing, any other process's has already handled its events. @@ -152,6 +155,15 @@ export const createDeadLetters: CreateDeadLettersFunction = ({ `Command "${letter.eventType}" is no longer in the registry`, ); } + const { payload, forgotten } = await openCommandPayload({ + payload: letter.payload, + keys: storage.personalKeys, + }); + if (forgotten) { + throw new DeadLetterNotRetriableError( + `Command "${letter.eventType}" holds personal data of a person forgotten since`, + ); + } const context: CausationContext = { correlationId: ids.next(), causationId: letter.id, @@ -160,7 +172,7 @@ export const createDeadLetters: CreateDeadLettersFunction = ({ await settled(letter, async (unit) => { await pipeline.dispatchUnattended({ type: letter.eventType, - payload: letter.payload, + payload, context, within: unit, }); @@ -202,8 +214,18 @@ export const createDeadLetters: CreateDeadLettersFunction = ({ } }; + // A forgotten person's fields read as `null`, as an event's do. + const opened = async (letter: DeadLetter): Promise => { + if (letter.payload === undefined) return letter; + const { payload } = await openCommandPayload({ + payload: letter.payload, + keys: storage.personalKeys, + }); + return { ...letter, payload }; + }; + const withParked = async (letter: DeadLetter): Promise => { - if (letter.kind !== "process") return letter; + if (letter.kind !== "process") return opened(letter); return { ...letter, parked: await processes.parkedBehind(letter) }; }; @@ -223,7 +245,7 @@ export const createDeadLetters: CreateDeadLettersFunction = ({ handler: letter.handler, at: clock.now().toISOString(), }); - const retried: DeadLetter = { ...letter, status: "retried" }; + const retried: DeadLetter = { ...(await opened(letter)), status: "retried" }; if (letter.kind !== "process") return retried; return { ...retried, parked: await processes.stillParked(letter) }; }, @@ -235,7 +257,7 @@ export const createDeadLetters: CreateDeadLettersFunction = ({ kind: letter.kind, handler: letter.handler, }); - return { ...letter, status: "discarded" }; + return { ...(await opened(letter)), status: "discarded" }; }, }; }; diff --git a/packages/core/src/kernel/personal-data/commands.test.ts b/packages/core/src/kernel/personal-data/commands.test.ts new file mode 100644 index 00000000..b6cce37c --- /dev/null +++ b/packages/core/src/kernel/personal-data/commands.test.ts @@ -0,0 +1,474 @@ +import { describe, expect, it } from "vitest"; +import { z } from "zod"; +import { DeadLetterNotRetriableError, ValidationError } from "../../contracts/errors.ts"; +import { silentLogger } from "../../contracts/logger.ts"; +import { memory } from "../../memory/index.ts"; +import type { PersonalDataForgottenEvent } from "../../modules/bounda-events.ts"; +import type { CommandPayloadArgs, EventPayloadArgs } from "../../modules/payload.ts"; +import { person, personal } from "../../modules/personal.ts"; +import type { Registry } from "../../modules/registry.ts"; +import { createTestApp } from "../../testing/index.ts"; +import { createRecordingLogger } from "../test-support.ts"; +import { openCommandPayload, sealCommandPayload } from "./commands.ts"; +import { generatePersonalKey, isSealed } from "./crypto.ts"; +import { personalMarksOf } from "./marks.ts"; + +const received: unknown[] = []; +// Runs inside `remindCustomer`'s handler once, before it answers. +let whileRunning: (() => Promise) | null = null; +let failing = false; + +type Builders = Record unknown>; + +const registry = { + aggregates: { + customer: { + events: { + customerInvited: { + payload: ({ z, personal }: EventPayloadArgs) => z.object({ email: personal(z.email()) }), + evolve: () => ({}), + }, + customerReminded: { + payload: ({ z, personal }: EventPayloadArgs) => z.object({ to: personal(z.email()) }), + evolve: () => ({}), + }, + }, + commands: { + remindCustomer: { + module: { + payload: ({ z, personal }: CommandPayloadArgs) => + z.object({ customerId: z.string(), email: personal(z.email()) }), + handler: async ({ + command, + events, + }: { + command: { payload: { email: string } }; + events: Builders; + }) => { + received.push(command.payload); + const run = whileRunning; + whileRunning = null; + await run?.(); + if (failing) throw new ValidationError("the mail server refused it", []); + return [events.customerReminded?.({ to: command.payload.email })]; + }, + }, + }, + inviteCustomer: { + module: { + payload: ({ z, personal }: CommandPayloadArgs) => + z.object({ customerId: z.string(), email: personal(z.email()) }), + handler: ({ + command, + events, + }: { + command: { payload: { email: string } }; + events: Builders; + }) => [events.customerInvited?.({ email: command.payload.email })], + }, + }, + introduceFriend: { + module: { + payload: ({ z, personal, person }: CommandPayloadArgs) => + z.object({ + customerId: z.string(), + friendId: person(z.string()), + note: personal(z.string()).optional(), + }), + handler: ({ command }: { command: { payload: unknown } }) => { + received.push(command.payload); + return []; + }, + }, + }, + }, + policies: { + remindOnCustomerInvited: { + module: { + on: "CustomerInvited", + handler: async ({ + event, + commands, + }: { + event: { aggregateId: string; payload: { email: string | null } }; + commands: Record Promise>; + }) => { + if (event.payload.email === null) return; + await commands.remindCustomer?.( + { customerId: event.aggregateId, email: event.payload.email }, + { delay: "1m" }, + ); + }, + }, + source: "customer", + }, + }, + processes: {}, + }, + }, + readModels: {}, +} as const satisfies Registry; + +const ana = { customerId: "c-1", email: "ana@example.com" }; + +const setup = async () => { + received.length = 0; + whileRunning = null; + failing = false; + const adapter = memory(); + const storage = await adapter.createStorage({ logger: silentLogger }); + const { logger, entries: logs } = createRecordingLogger(); + const { app, clock } = await createTestApp({ registry, adapter, logger }); + const runDue = async () => { + clock.advance(60_000); + await app.runUntilIdle(); + }; + const forgottenPayload = async () => { + const events = await app.events.read(); + const forgotten = events.filter((event) => event.type === "PersonalDataForgotten"); + return (forgotten.at(-1) as PersonalDataForgottenEvent | undefined)?.payload; + }; + return { app, storage, logs, runDue, forgottenPayload }; +}; + +describe("sealCommandPayload and openCommandPayload", () => { + const marks = (shape: z.ZodRawShape) => + personalMarksOf({ schema: z.object(shape), path: "aggregates.customer.commands.c" }); + + it("seals the marked fields with the key of the aggregate's person, creating it", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const sealed = await sealCommandPayload({ + type: "RemindCustomer", + aggregateId: "c-1", + payload: { customerId: "c-1", email: "ana@example.com" }, + marks: marks({ customerId: z.string(), email: personal(z.email()) }), + keys: storage.personalKeys, + }); + const key = await storage.personalKeys.get("c-1"); + expect(sealed.persons).toEqual([{ person: "c-1", fingerprint: key?.fingerprint }]); + const payload = sealed.payload as { customerId: unknown; email: unknown }; + expect(payload.customerId).toBe("c-1"); + expect(isSealed(payload.email)).toBe(true); + expect(await openCommandPayload({ payload, keys: storage.personalKeys })).toEqual({ + payload: { customerId: "c-1", email: "ana@example.com" }, + forgotten: false, + }); + }); + + it("seals under the person a field names, and keeps the key in force", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const kept = await storage.personalKeys.add("f-1", await generatePersonalKey()); + const sealed = await sealCommandPayload({ + type: "IntroduceFriend", + aggregateId: "c-1", + payload: { customerId: "c-1", friendId: "f-1", note: "old friend" }, + marks: marks({ friendId: person(z.string()), note: personal(z.string()) }), + keys: storage.personalKeys, + }); + expect(sealed.persons).toEqual([{ person: "f-1", fingerprint: kept.fingerprint }]); + expect(await storage.personalKeys.get("c-1")).toBeNull(); + }); + + it("leaves a payload with nothing to seal as it is, naming no one and keying no one", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const shape = { friendId: person(z.string()), note: personal(z.string()).optional() }; + for (const [payload, payloadMarks] of [ + [{ friendId: "f-1" }, marks(shape)], + [{ friendId: "f-1", note: "hi" }, null], + ["not an object", marks(shape)], + ] as const) { + expect( + await sealCommandPayload({ + type: "IntroduceFriend", + aggregateId: "c-1", + payload, + marks: payloadMarks, + keys: storage.personalKeys, + }), + ).toEqual({ payload, persons: [] }); + } + expect(await storage.personalKeys.get("f-1")).toBeNull(); + }); + + it.each([undefined, "", 42n])( + "refuses a person field that names no one: %s", + async (friendId) => { + const storage = await memory().createStorage({ logger: silentLogger }); + await expect( + sealCommandPayload({ + type: "IntroduceFriend", + aggregateId: "c-1", + payload: { friendId, note: "hi" }, + marks: marks({ friendId: person(z.string()), note: personal(z.string()) }), + keys: storage.personalKeys, + }), + ).rejects.toThrow( + new ValidationError('Command IntroduceFriend names no person in "friendId"', [ + { path: ["friendId"], message: "Expected a non-empty string" }, + ]), + ); + }, + ); + + it("opens to null, telling it forgotten, a field whose key is gone or was replaced", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const { payload } = await sealCommandPayload({ + type: "RemindCustomer", + aggregateId: "c-1", + payload: { email: "ana@example.com", plan: "pro" }, + marks: marks({ email: personal(z.email()), plan: z.string() }), + keys: storage.personalKeys, + }); + await storage.personalKeys.remove("c-1"); + const forgotten = { payload: { email: null, plan: "pro" }, forgotten: true }; + expect(await openCommandPayload({ payload, keys: storage.personalKeys })).toEqual(forgotten); + await storage.personalKeys.add("c-1", await generatePersonalKey()); + expect(await openCommandPayload({ payload, keys: storage.personalKeys })).toEqual(forgotten); + }); + + it("names a person by a numeric id, as text", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const sealed = await sealCommandPayload({ + type: "IntroduceFriend", + aggregateId: "c-1", + payload: { friendId: 7, note: "hi" }, + marks: marks({ friendId: person(z.number()), note: personal(z.string()) }), + keys: storage.personalKeys, + }); + expect(sealed.persons.map((held) => held.person)).toEqual(["7"]); + }); + + it("reads each person's key once, and opens each field under its own person's key", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const shape = { + friendId: person(z.string()), + a: personal(z.string()), + b: personal(z.string()), + }; + const sealedFor = async (friendId: string, a: string, b: string) => + ( + await sealCommandPayload({ + type: "IntroduceFriend", + aggregateId: "c-1", + payload: { friendId, a, b }, + marks: marks(shape), + keys: storage.personalKeys, + }) + ).payload as Record; + const ana = await sealedFor("f-1", "a1", "b1"); + const bo = await sealedFor("f-2", "a2", "b2"); + const reads: string[] = []; + const keys = { + ...storage.personalKeys, + get: (person: string) => { + reads.push(person); + return storage.personalKeys.get(person); + }, + }; + expect(await openCommandPayload({ payload: ana, keys })).toEqual({ + payload: { friendId: "f-1", a: "a1", b: "b1" }, + forgotten: false, + }); + expect(reads).toEqual(["f-1"]); + expect(await openCommandPayload({ payload: { a: ana.a, b: bo.b }, keys })).toEqual({ + payload: { a: "a1", b: "b2" }, + forgotten: false, + }); + }); + + it("opens a payload that holds nothing sealed as it is", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + for (const payload of [{ email: "ana@example.com" }, "text", null]) { + expect(await openCommandPayload({ payload, keys: storage.personalKeys })).toEqual({ + payload, + forgotten: false, + }); + } + }); +}); + +describe("personal data in delayed commands", () => { + it("stores a delayed command's personal fields sealed and runs it with them in the clear", async () => { + const { app, storage, runDue } = await setup(); + await app.commands.remindCustomer(ana, { delay: "1m" }); + const [entry] = await storage.scheduler.list(); + const key = await storage.personalKeys.get("c-1"); + expect(entry?.persons).toEqual([{ person: "c-1", fingerprint: key?.fingerprint }]); + expect(isSealed((entry?.command.payload as { email?: unknown } | undefined)?.email)).toBe(true); + expect(JSON.stringify(entry)).not.toContain("ana@example.com"); + await runDue(); + expect(received).toEqual([ana]); + expect(await storage.scheduler.list()).toEqual([]); + await app.stop(); + }); + + it("seals what a policy schedules from its unit of work", async () => { + const { app, storage, runDue } = await setup(); + await app.commands.inviteCustomer(ana); + await app.runUntilIdle(); + const [entry] = await storage.scheduler.list(); + expect(entry?.command.type).toBe("RemindCustomer"); + expect(entry?.persons?.map((held) => held.person)).toEqual(["c-1"]); + expect(JSON.stringify(entry)).not.toContain("ana@example.com"); + await runDue(); + expect(received).toEqual([ana]); + await app.stop(); + }); + + it("seals what it stores with an idempotency key too", async () => { + const { app, storage } = await setup(); + await app.commands.remindCustomer(ana, { delay: "1m", idempotencyKey: "remind-1" }); + expect(JSON.stringify(await storage.scheduler.list())).not.toContain("ana@example.com"); + await app.stop(); + }); + + it("cancels a person's delayed commands on forget, and names them in PersonalDataForgotten", async () => { + const { app, storage, runDue, forgottenPayload } = await setup(); + await app.commands.remindCustomer(ana, { delay: "1m" }); + await app.commands.remindCustomer( + { customerId: "c-2", email: "bo@example.com" }, + { delay: "1m" }, + ); + await app.commands.introduceFriend( + { customerId: "c-2", friendId: "c-1", note: "met at school" }, + { delay: "2m" }, + ); + const named = (await storage.scheduler.list()).filter((entry) => + entry.persons?.some((held) => held.person === "c-1"), + ); + expect(named).toHaveLength(2); + + await app.forget("c-1"); + expect(await forgottenPayload()).toEqual({ + personId: "c-1", + cancelledCommands: named.map((entry) => ({ + commandId: entry.dedupeKey.replace(/^command:/, ""), + type: entry.command.type, + })), + removedDeadLetters: [], + }); + expect((await storage.scheduler.list()).map((entry) => entry.command.aggregateId)).toEqual([ + "c-2", + ]); + await runDue(); + await runDue(); + expect(received).toEqual([{ customerId: "c-2", email: "bo@example.com" }]); + await app.stop(); + }); + + it("dead-letters a delayed command sealed, shows it opened, and removes it on forget", async () => { + const { app, storage, runDue, forgottenPayload } = await setup(); + failing = true; + await app.commands.remindCustomer(ana, { delay: "1m" }); + await runDue(); + const [stored] = await storage.deadLetterStore.list(); + const key = await storage.personalKeys.get("c-1"); + expect(stored?.persons).toEqual([{ person: "c-1", fingerprint: key?.fingerprint }]); + expect(JSON.stringify(stored)).not.toContain("ana@example.com"); + expect(await app.deadLetters.list()).toEqual([{ ...stored, payload: ana }]); + expect(await app.deadLetters.get(stored?.id ?? "")).toEqual({ ...stored, payload: ana }); + expect(await app.deadLetters.discard(stored?.id ?? "")).toEqual({ + ...stored, + payload: ana, + status: "discarded", + }); + + await app.forget("c-1"); + expect(await storage.deadLetterStore.list()).toEqual([]); + expect(await forgottenPayload()).toEqual({ + personId: "c-1", + cancelledCommands: [], + removedDeadLetters: [{ id: stored?.id, commandType: "RemindCustomer" }], + }); + await app.stop(); + }); + + it("retries a dead-lettered delayed command with its payload opened", async () => { + const { app, runDue } = await setup(); + failing = true; + await app.commands.remindCustomer(ana, { delay: "1m" }); + await runDue(); + failing = false; + const [letter] = await app.deadLetters.list(); + expect(await app.deadLetters.retry(letter?.id ?? "")).toMatchObject({ + payload: ana, + status: "retried", + }); + expect(received).toEqual([ana, ana]); + await app.stop(); + }); + + it.each([ + ["answers", false], + ["fails", true], + ])( + "writes nothing, no dead letter either, for a run a forget overtakes while its handler %s", + async (_, fails) => { + const { app, storage, runDue, forgottenPayload } = await setup(); + failing = fails; + await app.commands.remindCustomer(ana, { delay: "1m" }); + whileRunning = () => app.forget("c-1"); + await runDue(); + expect(whileRunning).toBeNull(); + expect(received).toEqual([ana]); + expect((await app.events.read()).map((event) => event.type)).toEqual([ + "PersonalDataForgotten", + ]); + expect((await forgottenPayload())?.cancelledCommands).toHaveLength(1); + expect(await storage.deadLetterStore.list()).toEqual([]); + expect(await storage.scheduler.list()).toEqual([]); + await app.stop(); + }, + ); + + it("drops, never running it, a delayed command sealed under a key its person no longer holds", async () => { + const { app, storage, logs, runDue } = await setup(); + await app.commands.remindCustomer(ana, { delay: "1m" }); + await storage.personalKeys.remove("c-1"); + await storage.personalKeys.add("c-1", await generatePersonalKey()); + await runDue(); + expect(received).toEqual([]); + expect(await storage.scheduler.list()).toEqual([]); + expect(await storage.deadLetterStore.list()).toEqual([]); + expect(logs).toContainEqual({ + level: "warn", + message: "scheduled command dropped: its person was forgotten", + fields: { command: "RemindCustomer", dedupeKey: expect.stringMatching(/^command:/) }, + }); + await app.stop(); + }); + + it("refuses to retry a dead letter whose person was forgotten since, and shows its fields as null", async () => { + const { app, storage, runDue } = await setup(); + failing = true; + await app.commands.remindCustomer(ana, { delay: "1m" }); + await runDue(); + failing = false; + await storage.personalKeys.remove("c-1"); + const [letter] = await app.deadLetters.list(); + expect(letter?.payload).toEqual({ customerId: "c-1", email: null }); + await expect(app.deadLetters.retry(letter?.id ?? "")).rejects.toThrow( + new DeadLetterNotRetriableError( + 'Command "RemindCustomer" holds personal data of a person forgotten since', + ), + ); + expect(received).toEqual([ana]); + expect((await app.deadLetters.get(letter?.id ?? ""))?.status).toBe("failed"); + await app.stop(); + }); + + it("runs a delayed command stored before its fields were marked as it was written, beyond forget's reach", async () => { + const { app, storage, runDue, forgottenPayload } = await setup(); + await storage.scheduler.schedule({ + dedupeKey: "command:legacy", + command: { type: "RemindCustomer", aggregateId: "c-1", payload: ana }, + executeAt: new Date("2026-01-01T00:00:30.000Z"), + context: { correlationId: "legacy", causationId: "legacy", depth: 0 }, + }); + await app.forget("c-1"); + expect((await forgottenPayload())?.cancelledCommands).toEqual([]); + await runDue(); + expect(received).toEqual([ana]); + await app.stop(); + }); +}); diff --git a/packages/core/src/kernel/personal-data/commands.ts b/packages/core/src/kernel/personal-data/commands.ts new file mode 100644 index 00000000..4df2e8f5 --- /dev/null +++ b/packages/core/src/kernel/personal-data/commands.ts @@ -0,0 +1,121 @@ +import type { KeyedPerson, PersonalKeyStore } from "../../adapter/storage/personal-key-store.ts"; +import { ValidationError } from "../../contracts/errors.ts"; +import { isRecord } from "../shared/merge-fields.ts"; +import { + generatePersonalKey, + importPersonalKey, + isSealed, + type PersonalKey, + seal, + unseal, +} from "./crypto.ts"; +import type { PersonalMarks } from "./marks.ts"; + +export interface SealCommandPayloadArgs { + readonly type: string; + readonly aggregateId: string; + readonly payload: unknown; + readonly marks: PersonalMarks | null; + readonly keys: PersonalKeyStore; +} + +/** + * A payload as it is stored, with the people whose fields it holds sealed, each with the key. + */ +export interface SealedPayload { + readonly payload: unknown; + readonly persons: readonly KeyedPerson[]; +} + +export interface SealCommandPayloadFunction { + (args: SealCommandPayloadArgs): Promise; +} + +const personOf = ( + { type, aggregateId }: SealCommandPayloadArgs, + payload: Record, + field: string | null, +): string => { + if (field === null) return aggregateId; + const value = payload[field]; + if ((typeof value !== "string" && typeof value !== "number") || value === "") { + throw new ValidationError(`Command ${type} names no person in "${field}"`, [ + { path: [field], message: "Expected a non-empty string" }, + ]); + } + return String(value); +}; + +const keyInForce = async (keys: PersonalKeyStore, person: string): Promise => + importPersonalKey( + (await keys.get(person)) ?? (await keys.add(person, await generatePersonalKey())), + ); + +/** + * Seals the payload's personal fields with the key of the person the command names, created on + * their first sealed field. A key removed before the payload is stored leaves it sealed under a key + * no one holds: the command then never runs, as `openCommandPayload` tells. + */ +export const sealCommandPayload: SealCommandPayloadFunction = async (args) => { + const { payload, marks, keys } = args; + if (marks === null || !isRecord(payload)) return { payload, persons: [] }; + const present = marks.fields.filter((field) => payload[field] !== undefined); + if (present.length === 0) return { payload, persons: [] }; + const person = personOf(args, payload, marks.person); + const key = await keyInForce(keys, person); + const sealed: Record = { ...payload }; + for (const field of present) { + sealed[field] = await seal({ key, person, value: payload[field] }); + } + return { payload: sealed, persons: [{ person, fingerprint: key.fingerprint }] }; +}; + +export interface OpenCommandPayloadArgs { + readonly payload: unknown; + readonly keys: PersonalKeyStore; +} + +/** + * A stored payload with its sealed fields opened. `forgotten` when one of them no longer opens, + * its person forgotten since: the field reads as `null`, and the command must not run. + */ +export interface OpenedPayload { + readonly payload: unknown; + readonly forgotten: boolean; +} + +export interface OpenCommandPayloadFunction { + (args: OpenCommandPayloadArgs): Promise; +} + +/** + * Opens whatever the payload holds sealed, whether or not the command still marks the field, so a + * payload stored before its marks changed reads as it was written; one stored before they existed + * holds nothing sealed and comes back as it is. + */ +export const openCommandPayload: OpenCommandPayloadFunction = async ({ payload, keys }) => { + if (!isRecord(payload)) return { payload, forgotten: false }; + const opened: Record = { ...payload }; + // Each person's key read and imported once, whatever number of their fields the payload seals. + const imported = new Map>(); + const keyOf = (person: string, fingerprint: string): Promise => { + const id = `${person}\u0000${fingerprint}`; + const known = imported.get(id); + if (known !== undefined) return known; + const key = keys + .get(person) + .then((stored) => + stored === null || stored.fingerprint !== fingerprint ? null : importPersonalKey(stored), + ); + imported.set(id, key); + return key; + }; + let forgotten = false; + for (const [field, value] of Object.entries(payload)) { + if (!isSealed(value)) continue; + const key = await keyOf(value.$personal.person, value.$personal.key); + if (key === null) forgotten = true; + opened[field] = await unseal({ key, sealed: value }); + } + return { payload: opened, forgotten }; +}; diff --git a/packages/core/src/kernel/personal-data/forget.test.ts b/packages/core/src/kernel/personal-data/forget.test.ts index adb4b521..d82e5e0a 100644 --- a/packages/core/src/kernel/personal-data/forget.test.ts +++ b/packages/core/src/kernel/personal-data/forget.test.ts @@ -1,6 +1,13 @@ import { describe, expect, it } from "vitest"; import type { Storage } from "../../adapter/adapter.ts"; -import { testPersonalKey } from "../../adapter/testing/fixtures.ts"; +import type { KeyedPerson } from "../../adapter/storage/personal-key-store.ts"; +import { + testCommand, + testContext, + testDeadLetter, + testKeyedPerson, + testPersonalKey, +} from "../../adapter/testing/fixtures.ts"; import { createFixedClock } from "../../contracts/clock.ts"; import { ConcurrencyError, ValidationError } from "../../contracts/errors.ts"; import { createSequentialIdGenerator } from "../../contracts/ids.ts"; @@ -46,7 +53,7 @@ describe("forgetPersonalData", () => { version: 1, position: 1, type: "PersonalDataForgotten", - payload: { personId: "c-1" }, + payload: { personId: "c-1", cancelledCommands: [], removedDeadLetters: [] }, timestamp: "2026-01-01T00:00:00.000Z", metadata: { correlationId: "id-1", @@ -60,7 +67,7 @@ describe("forgetPersonalData", () => { expect(logs).toContainEqual({ level: "info", message: "personal data forgotten", - fields: { personId: "c-1", hadKey: true }, + fields: { personId: "c-1", hadKey: true, cancelledCommands: 0, removedDeadLetters: 0 }, }); }); @@ -70,11 +77,54 @@ describe("forgetPersonalData", () => { expect((await forget("c-2")).version).toBe(2); expect(await storage.eventStore.lastPosition()).toBe(2); expect(logs.map((entry) => entry.fields)).toEqual([ - { personId: "c-2", hadKey: false }, - { personId: "c-2", hadKey: false }, + { personId: "c-2", hadKey: false, cancelledCommands: 0, removedDeadLetters: 0 }, + { personId: "c-2", hadKey: false, cancelledCommands: 0, removedDeadLetters: 0 }, ]); }); + it("cancels the scheduled commands and removes the dead letters that name the person, naming them", async () => { + const { storage, forget, logs } = await setup(); + const persons = [testKeyedPerson("c-1", "1")]; + const schedule = (dedupeKey: string, aggregateId: string, named: readonly KeyedPerson[]) => + storage.scheduler.schedule({ + dedupeKey, + command: testCommand(aggregateId), + executeAt: new Date("2026-01-01T00:01:00.000Z"), + context: testContext, + persons: named, + }); + await schedule("command:cmd-1", "c-1", persons); + await schedule("deadline:x", "c-1", persons); + await schedule("command:cmd-2", "c-2", [testKeyedPerson("c-2", "2")]); + await storage.deadLetterStore.add({ + ...testDeadLetter("dl-1"), + kind: "scheduled", + handler: "RemindCustomer", + persons, + }); + await storage.deadLetterStore.add(testDeadLetter("dl-2")); + await forget("c-1"); + const [event] = await storage.eventStore.readAll({ afterPosition: 0, limit: 10 }); + expect(event?.payload).toEqual({ + personId: "c-1", + cancelledCommands: [ + { commandId: "cmd-1", type: "RemindCustomer" }, + { commandId: "deadline:x", type: "RemindCustomer" }, + ], + removedDeadLetters: [{ id: "dl-1", commandType: "RemindCustomer" }], + }); + expect((await storage.scheduler.list()).map((entry) => entry.dedupeKey)).toEqual([ + "command:cmd-2", + ]); + expect((await storage.deadLetterStore.list()).map((letter) => letter.id)).toEqual(["dl-2"]); + expect(logs.at(-1)?.fields).toEqual({ + personId: "c-1", + hadKey: false, + cancelledCommands: 2, + removedDeadLetters: 1, + }); + }); + it("refuses an empty id", async () => { const { storage, forget } = await setup(); await expect(forget("")).rejects.toThrow( diff --git a/packages/core/src/kernel/personal-data/forget.ts b/packages/core/src/kernel/personal-data/forget.ts index d09ebb84..751dcb79 100644 --- a/packages/core/src/kernel/personal-data/forget.ts +++ b/packages/core/src/kernel/personal-data/forget.ts @@ -9,6 +9,7 @@ import { PERSONAL_DATA_FORGOTTEN_EVENT, type PersonalDataForgottenPayload, } from "../../modules/bounda-events.ts"; +import { scheduledCommandId } from "../command/pipeline.ts"; import { appendSystemEvent } from "../system-events.ts"; export interface ForgetPersonalDataArgs { @@ -26,8 +27,10 @@ export interface ForgetPersonalDataFunction { /** * Removes the person's key and appends `PersonalDataForgotten` to their stream, in one - * transaction: whoever reads the event finds the fields gone already. A person without a key is - * forgotten all the same, since read models may hold what their events carried in the clear. + * transaction: whoever reads the event finds the fields gone already. The scheduled commands and + * dead letters that hold the person's data go in the same transaction, and the event names them. + * A person without a key is forgotten all the same, since read models may hold what their events + * carried in the clear. */ export const forgetPersonalData: ForgetPersonalDataFunction = async ({ storage, @@ -43,17 +46,30 @@ export const forgetPersonalData: ForgetPersonalDataFunction = async ({ ]); } const correlationId = ids.next(); - const payload: PersonalDataForgottenPayload = { personId }; for (let attempt = 0; ; attempt += 1) { try { // The key goes first: a snapshot written beside the transaction then either waits for it - // and is forgotten with the rest, or finds no key and is never written. - const { removed, stored } = await storage.transact(async (transaction) => { + // and is forgotten with the rest, or finds no key and is never written. The scheduled + // commands go before the dead letters: a worker giving one up writes its letter and settles + // its claim together, so either this waits for it and then finds the letter, or the claim + // is lost and the letter rolls back. + const { removed, payload, stored } = await storage.transact(async (transaction) => { const hadKey = await transaction.personalKeys.remove(personId); await transaction.snapshotStore.forget(personId); await transaction.idempotencyStore.forget(personId); + const cancelled = await transaction.scheduler.forget(personId); + const letters = await transaction.deadLetterStore.forget(personId); + const payload: PersonalDataForgottenPayload = { + personId, + cancelledCommands: cancelled.map(({ dedupeKey, command }) => ({ + commandId: scheduledCommandId(dedupeKey) ?? dedupeKey, + type: command.type, + })), + removedDeadLetters: letters.map(({ id, handler }) => ({ id, commandType: handler })), + }; return { removed: hadKey, + payload, stored: await appendSystemEvent({ eventStore: transaction.eventStore, ids, @@ -66,7 +82,12 @@ export const forgetPersonalData: ForgetPersonalDataFunction = async ({ }), }; }); - logger.info("personal data forgotten", { personId, hadKey: removed }); + logger.info("personal data forgotten", { + personId, + hadKey: removed, + cancelledCommands: payload.cancelledCommands.length, + removedDeadLetters: payload.removedDeadLetters.length, + }); return { scheduled: false, aggregateType: BOUNDA_AGGREGATE, diff --git a/packages/core/src/kernel/personal-data/marks.test.ts b/packages/core/src/kernel/personal-data/marks.test.ts index d9250744..1b252476 100644 --- a/packages/core/src/kernel/personal-data/marks.test.ts +++ b/packages/core/src/kernel/personal-data/marks.test.ts @@ -53,7 +53,7 @@ describe("personalMarksOf", () => { }); expect(() => personalMarksOf({ schema, path })).toThrow( new ConfigurationError( - `${path}: marks "buyerId" and "sellerId" with person; an event names at most one person`, + `${path}: marks "buyerId" and "sellerId" with person; a payload names at most one person`, ), ); }); diff --git a/packages/core/src/kernel/personal-data/marks.ts b/packages/core/src/kernel/personal-data/marks.ts index 60a07ffc..08c54e73 100644 --- a/packages/core/src/kernel/personal-data/marks.ts +++ b/packages/core/src/kernel/personal-data/marks.ts @@ -3,8 +3,8 @@ import { ConfigurationError } from "../../contracts/errors.ts"; import { personalMarkOf } from "../../modules/personal.ts"; /** - * What an event's payload schema marks: the fields to encrypt and the field naming the person - * they belong to, `null` when the aggregate's id names them. + * What an event's or a command's payload schema marks: the fields to encrypt and the field naming + * the person they belong to, `null` when the aggregate's id names them. */ export interface PersonalMarks { readonly person: string | null; @@ -21,7 +21,7 @@ export interface PersonalMarksOfFunction { } /** - * `null` for an event without personal fields. Only the top-level fields of an object payload + * `null` for a payload without personal fields. Only the top-level fields of an object payload * are read: a mark deeper down, or on a payload that is not an object, is never found, so a * whole object or array is marked as one field. */ @@ -36,7 +36,7 @@ export const personalMarksOf: PersonalMarksOfFunction = ({ schema, path }) => { } if (persons.length > 1) { throw new ConfigurationError( - `${path}: marks ${persons.map((name) => `"${name}"`).join(" and ")} with person; an event names at most one person`, + `${path}: marks ${persons.map((name) => `"${name}"`).join(" and ")} with person; a payload names at most one person`, ); } if (fields.length === 0) return null; diff --git a/packages/core/src/kernel/scheduler/worker.ts b/packages/core/src/kernel/scheduler/worker.ts index fd28c31d..868f2ba2 100644 --- a/packages/core/src/kernel/scheduler/worker.ts +++ b/packages/core/src/kernel/scheduler/worker.ts @@ -11,6 +11,7 @@ import type { IdGenerator } from "../../contracts/ids.ts"; import type { Logger } from "../../contracts/logger.ts"; import type { AggregatesRuntime } from "../aggregate/runtime.ts"; import { type CommandPipeline, scheduledCommandId } from "../command/pipeline.ts"; +import { openCommandPayload } from "../personal-data/commands.ts"; import type { DelayedPolicies } from "../policy/delayed.ts"; import { PROCESS_DEADLINE_COMMAND, type ProcessDeadlinePayload } from "../process/deadlines.ts"; import type { ProcessDeadlines } from "../process/deliver-deadline.ts"; @@ -237,6 +238,7 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction firstFailedAt: now, lastFailedAt: now, payload: entry.command.payload, + ...(entry.persons === undefined ? {} : { persons: entry.persons }), }; }; @@ -303,9 +305,23 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction within: unit, }); } else { + const { payload, forgotten } = await openCommandPayload({ + payload: entry.command.payload, + keys: storage.personalKeys, + }); + // Its person was forgotten since the seal. A forget that cancelled the command leaves + // this claim lost, so the run writes nothing; one that missed it, stored under a key the + // forget removed after the seal read it, has it dropped here, never run. + if (forgotten) { + logger.warn("scheduled command dropped: its person was forgotten", { + command: entry.command.type, + dedupeKey: entry.dedupeKey, + }); + return; + } await pipeline.dispatchUnattended({ type: entry.command.type, - payload: entry.command.payload, + payload, context: entry.context, commandId: scheduledCommandId(entry.dedupeKey), within: unit, diff --git a/packages/core/src/memory/dead-letter-store.ts b/packages/core/src/memory/dead-letter-store.ts index 309c67b4..697a6102 100644 --- a/packages/core/src/memory/dead-letter-store.ts +++ b/packages/core/src/memory/dead-letter-store.ts @@ -1,9 +1,10 @@ -import { codePointOrder } from "../adapter/sql/order.ts"; +import { byFirstFailure } from "../adapter/sql/dead-letters.ts"; import type { DeadLetter, DeadLetterStore, ListDeadLettersArgs, } from "../adapter/storage/dead-letter-store.ts"; +import { namesPerson } from "../adapter/storage/personal-key-store.ts"; import { DeadLetterSettledError } from "../contracts/errors.ts"; import { createStoreEntries, type WithEntries } from "./entries.ts"; import { jsonCopy } from "./json-copy.ts"; @@ -21,11 +22,11 @@ const matches = (letter: DeadLetter, args: ListDeadLettersArgs): boolean => (args.handler === undefined || letter.handler === args.handler) && (args.status === undefined || letter.status === args.status); -const copy = ({ payload, ...letter }: DeadLetter): DeadLetter => - payload === undefined ? letter : { ...letter, payload: jsonCopy(payload) }; - -const byFirstFailure = (a: DeadLetter, b: DeadLetter): number => - codePointOrder(a.firstFailedAt, b.firstFailedAt) || codePointOrder(a.id, b.id); +const copy = ({ payload, persons, ...letter }: DeadLetter): DeadLetter => ({ + ...letter, + ...(payload === undefined ? {} : { payload: jsonCopy(payload) }), + ...(persons === undefined || persons.length === 0 ? {} : { persons: jsonCopy(persons) }), +}); export const createKeptDeadLetterStore: CreateKeptDeadLetterStoreFunction = () => { const letters = new Map(); @@ -61,6 +62,11 @@ export const createKeptDeadLetterStore: CreateKeptDeadLetterStoreFunction = () = remove: async (id) => { letters.delete(id); }, + forget: async (person) => { + const named = select({}).filter((letter) => namesPerson(letter, person)); + for (const { id } of named) letters.delete(id); + return named.map(copy); + }, }; return { store, entries: createStoreEntries(letters, (id: string) => id) }; }; diff --git a/packages/core/src/memory/scheduler.ts b/packages/core/src/memory/scheduler.ts index 15c2f525..fd4e163f 100644 --- a/packages/core/src/memory/scheduler.ts +++ b/packages/core/src/memory/scheduler.ts @@ -1,5 +1,6 @@ import { v4 as randomUUID } from "uuid"; import { byExecuteAt } from "../adapter/sql/scheduling.ts"; +import { namesPerson } from "../adapter/storage/personal-key-store.ts"; import type { ClaimedCommand, ScheduledClaim, @@ -31,6 +32,7 @@ const toScheduled = ({ executeAt, context, attempts, + persons, }: Entry): ScheduledCommand => ({ dedupeKey, command: { @@ -41,11 +43,13 @@ const toScheduled = ({ executeAt, context: jsonCopy(context), attempts, + ...(persons === undefined ? {} : { persons: jsonCopy(persons) }), }); const sameCommand = (a: ScheduledCommand, b: ScheduledCommand): boolean => JSON.stringify(a.command) === JSON.stringify(b.command) && - JSON.stringify(a.context) === JSON.stringify(b.context); + JSON.stringify(a.context) === JSON.stringify(b.context) && + JSON.stringify(a.persons ?? []) === JSON.stringify(b.persons ?? []); export const createKeptScheduler: CreateKeptSchedulerFunction = () => { const entries = new Map(); @@ -61,7 +65,14 @@ export const createKeptScheduler: CreateKeptSchedulerFunction = () => { }; const store: Scheduler = { - schedule: async ({ dedupeKey, command, executeAt, context, keepTimingOfSameCommand }) => { + schedule: async ({ + dedupeKey, + command, + executeAt, + context, + keepTimingOfSameCommand, + persons = [], + }) => { const scheduled: ScheduledCommand = { dedupeKey, command: { @@ -72,6 +83,7 @@ export const createKeptScheduler: CreateKeptSchedulerFunction = () => { executeAt: executeAt.toISOString(), context: jsonCopy(context), attempts: 0, + ...(persons.length === 0 ? {} : { persons: jsonCopy(persons) }), }; const existing = entries.get(dedupeKey); if ( @@ -161,6 +173,13 @@ export const createKeptScheduler: CreateKeptSchedulerFunction = () => { .sort(byExecuteAt) .slice(offset, limit === undefined ? undefined : offset + limit) .map(toScheduled), + forget: async (person) => { + const named = [...entries.values()] + .filter((entry) => namesPerson(entry, person)) + .sort(byExecuteAt); + for (const { dedupeKey } of named) entries.delete(dedupeKey); + return named.map(toScheduled); + }, }; return { store, entries: createStoreEntries(entries, (dedupeKey: string) => dedupeKey) }; }; diff --git a/packages/core/src/memory/transaction.ts b/packages/core/src/memory/transaction.ts index 997afe92..edfd0e8d 100644 --- a/packages/core/src/memory/transaction.ts +++ b/packages/core/src/memory/transaction.ts @@ -1,11 +1,15 @@ import type { Storage } from "../adapter/adapter.ts"; -import { type DeferredWriteTarget, deferWrites } from "../adapter/deferred-writes.ts"; +import { + type DeferredStores, + type DeferredWriteTarget, + deferWrites, +} from "../adapter/deferred-writes.ts"; import { createStagedEventStore } from "../adapter/staged-event-store.ts"; import type { CheckpointStore } from "../adapter/storage/checkpoint-store.ts"; import type { DeadLetterStore } from "../adapter/storage/dead-letter-store.ts"; import type { IdempotencyStore } from "../adapter/storage/idempotency-store.ts"; import type { ClaimKey, InboxLedger } from "../adapter/storage/inbox-ledger.ts"; -import type { PersonalKeyStore } from "../adapter/storage/personal-key-store.ts"; +import { namesPerson, type PersonalKeyStore } from "../adapter/storage/personal-key-store.ts"; import type { Scheduler } from "../adapter/storage/scheduler.ts"; import type { SnapshotStore } from "../adapter/storage/snapshot-store.ts"; import type { StoreEntries } from "./entries.ts"; @@ -79,6 +83,41 @@ const trackedPersonalKeys = ( }; }; +// Removed at once, as the keys are, so a forget answers what it removed: a worker that settles +// the claim of a command it cancelled finds it lost, and one that dead-lettered it first has its +// letter found. Put back the same way when the transaction fails. +const trackedForgets = ( + deferred: DeferredStores, + { scheduler, deadLetterStore, entries }: CreateUndoLogArgs, + undos: Undos, +): Pick => { + const tracked = trackedBy(undos); + return { + scheduler: { + ...deferred.scheduler, + forget: async (person) => { + const named = (await scheduler.list()).filter((entry) => namesPerson(entry, person)); + for (const { dedupeKey } of named) { + await tracked(entries.scheduler, dedupeKey, () => scheduler.cancel(dedupeKey)); + } + return named; + }, + }, + deadLetterStore: { + ...deferred.deadLetterStore, + forget: async (person) => { + const named = (await deadLetterStore.list()).filter((letter) => + namesPerson(letter, person), + ); + for (const { id } of named) { + await tracked(entries.deadLetterStore, id, () => deadLetterStore.remove(id)); + } + return named; + }, + }, + }; +}; + const createUndoLog = ({ inboxLedger, deadLetterStore, @@ -131,8 +170,9 @@ const createUndoLog = ({ * in one synchronous run, so no reader sees one stream appended without the others, nor an * event written meanwhile past a position still reserved. A failed transaction appends nothing, * gives its positions back and puts back each entry it changed, unless someone changed it since; - * its personal keys are written at once and put back the same way, and the snapshots it forgets - * are gone at once, for good, as a cache can be. + * its personal keys, and the scheduled commands and dead letters it forgets, are written at once + * and put back the same way, and the snapshots it forgets are gone at once, for good, as a cache + * can be. * `tryClaim`, `claimDue`, both `renew`, `purge` and writes made on the stores themselves do not * wait for a commit, so they can see a write it may still put back. */ @@ -153,12 +193,13 @@ export const createMemoryStorageTransaction: CreateMemoryStorageTransactionFunct }, }); const { stores, flush } = deferWrites(live); - const keyUndos: Undos = []; + const atOnce: Undos = []; try { const result = await work({ eventStore: staged, ...stores, - personalKeys: trackedPersonalKeys(personalKeys, live.entries.personalKeys, keyUndos), + ...trackedForgets(stores, live, atOnce), + personalKeys: trackedPersonalKeys(personalKeys, live.entries.personalKeys, atOnce), snapshotStore, }); // A transaction committed from inside this commit, as from a store write, waits forever. @@ -174,11 +215,11 @@ export const createMemoryStorageTransaction: CreateMemoryStorageTransactionFunct release?.(); } reserved.length = 0; - keyUndos.length = 0; + atOnce.length = 0; return result; } finally { eventStore.release(reserved); - undoAll(keyUndos); + undoAll(atOnce); } }; }; diff --git a/packages/core/src/modules/bounda-events.ts b/packages/core/src/modules/bounda-events.ts index fc96b928..3b906611 100644 --- a/packages/core/src/modules/bounda-events.ts +++ b/packages/core/src/modules/bounda-events.ts @@ -10,12 +10,31 @@ export const BOUNDA_AGGREGATE: "bounda" = "bounda"; /** * Appended by `forget(personId)` to the stream `bounda:`, once the person's key is * gone: every event's personal field of theirs reads as `null` from then on, and every read model - * that projects such fields handles it. + * that projects such fields handles it. It names the stored commands that held the person's data + * and went with the key, since no other event records them. */ export const PERSONAL_DATA_FORGOTTEN_EVENT: "PersonalDataForgotten" = "PersonalDataForgotten"; +/** + * A command scheduled with `delay` that `forget` cancelled before it ran. + */ +export interface CancelledCommand { + readonly commandId: string; + readonly type: string; +} + +/** + * The dead letter of a scheduled command that `forget` removed, whatever its status. + */ +export interface RemovedDeadLetter { + readonly id: string; + readonly commandType: string; +} + export interface PersonalDataForgottenPayload { readonly personId: string; + readonly cancelledCommands: readonly CancelledCommand[]; + readonly removedDeadLetters: readonly RemovedDeadLetter[]; } export type PersonalDataForgottenEvent = StoredEvent< diff --git a/packages/core/src/modules/command.ts b/packages/core/src/modules/command.ts index 2b91e505..6d1b7c12 100644 --- a/packages/core/src/modules/command.ts +++ b/packages/core/src/modules/command.ts @@ -2,10 +2,10 @@ import type { Command } from "../contracts/command.ts"; import type { DomainError } from "../contracts/errors.ts"; import type { HandlerState } from "./aggregate.ts"; import type { EventBuilders, EventModules } from "./event.ts"; -import type { PayloadFunction } from "./payload.ts"; +import type { CommandPayloadFunction } from "./payload.ts"; export interface CommandModule { - readonly payload?: PayloadFunction; + readonly payload?: CommandPayloadFunction; readonly rejections?: (args: never) => Readonly>; readonly handler: (args: never) => unknown; } diff --git a/packages/core/src/modules/index.ts b/packages/core/src/modules/index.ts index 85846050..519703b1 100644 --- a/packages/core/src/modules/index.ts +++ b/packages/core/src/modules/index.ts @@ -6,8 +6,10 @@ export type { UnknownState, } from "./aggregate.ts"; export type { + CancelledCommand, PersonalDataForgottenEvent, PersonalDataForgottenPayload, + RemovedDeadLetter, } from "./bounda-events.ts"; export { BOUNDA_AGGREGATE, @@ -38,6 +40,7 @@ export type { } from "./naming.ts"; export { capitalize, policyTrigger, toCamelCase } from "./naming.ts"; export type { + CommandPayloadArgs, EmptyPayload, EventPayloadArgs, PayloadArgs, @@ -46,6 +49,8 @@ export type { ZodApi, } from "./payload.ts"; export type { + CommandPersonalFunction, + CommandPersonalSchema, ForgetsPersonalData, HasPersonalFields, PersonalFunction, diff --git a/packages/core/src/modules/payload.ts b/packages/core/src/modules/payload.ts index 0d16e3dc..1fc4cb64 100644 --- a/packages/core/src/modules/payload.ts +++ b/packages/core/src/modules/payload.ts @@ -1,5 +1,5 @@ import type { z } from "zod"; -import type { PersonalFunction, PersonFunction } from "./personal.ts"; +import type { CommandPersonalFunction, PersonalFunction, PersonFunction } from "./personal.ts"; /** * The Zod namespace, injected into every `payload`, `state` and similar schema function so user @@ -29,6 +29,19 @@ export type EventPayloadFunction = ( args: EventPayloadArgs, ) => Schema; +/** + * Arguments of a command's `payload`: `z`, and `personal` and `person` to mark the fields that + * hold personal data and the one that names whose it is, for when the command is stored. + */ +export interface CommandPayloadArgs extends PayloadArgs { + readonly personal: CommandPersonalFunction; + readonly person: PersonFunction; +} + +export type CommandPayloadFunction = ( + args: CommandPayloadArgs, +) => Schema; + /** * The payload type of a module without `payload`: an object with no keys. */ diff --git a/packages/core/src/modules/personal.ts b/packages/core/src/modules/personal.ts index 7f736824..8dd4f930 100644 --- a/packages/core/src/modules/personal.ts +++ b/packages/core/src/modules/personal.ts @@ -21,20 +21,40 @@ export interface PersonSchema readonly [personBrand]: true; } +/** + * The schema `personal(schema)` returns in a command's payload: `schema`, marked as personal data. + * A command whose person is forgotten never runs, so its handler reads the field as `schema` gives + * it, never `null`. + */ +export interface CommandPersonalSchema + extends z.ZodPipe, z.output>> { + readonly [personalBrand]: true; +} + /** * Marks a field of an event's payload as personal data: it is stored encrypted with the key of * the person the event names, and reads as `null` once that person is forgotten. Any schema - * works, an object or an array included, at the top level of the payload. A command or a query - * payload cannot use it. + * works, an object or an array included, at the top level of the payload. A query payload cannot + * use it; a command's has its own, `CommandPersonalFunction`. */ export interface PersonalFunction { (schema: Schema): PersonalSchema; } /** - * Marks the field of an event's payload that identifies the person its personal fields belong - * to, such as the customer's id in an order's event. Without it, the aggregate's id is the - * person. An event has at most one. + * Marks a field of a command's payload as personal data: wherever the command is stored, scheduled + * with `delay` or dead-lettered, the field is encrypted with the key of the person the command + * names, and `forget` cancels the command or removes its dead letter. A command dispatched at + * once stores nothing, so the mark changes nothing for it. Same placement rules as an event's. + */ +export interface CommandPersonalFunction { + (schema: Schema): CommandPersonalSchema; +} + +/** + * Marks the field of an event's or a command's payload that identifies the person its personal + * fields belong to, such as the customer's id in an order's event. Without it, the aggregate's id + * is the person. A payload has at most one. */ export interface PersonFunction { (schema: Schema): PersonSchema; diff --git a/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/pay-order.ts b/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/pay-order.ts index 7481c2d5..a020199d 100644 --- a/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/pay-order.ts +++ b/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/pay-order.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../pay-order.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "PayOrder", core.PayloadOf, diff --git a/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/place-order.ts b/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/place-order.ts index b8c40ea1..82a1c787 100644 --- a/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/place-order.ts +++ b/packages/core/test-types/fixtures/order-app-inferred/app/domain/order/commands/+types/place-order.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../place-order.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "PlaceOrder", core.PayloadOf, diff --git a/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/+types/register-customer.ts b/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/+types/register-customer.ts index 1eb088bc..2758371a 100644 --- a/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/+types/register-customer.ts +++ b/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/+types/register-customer.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../register-customer.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "RegisterCustomer", core.PayloadOf, diff --git a/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/register-customer.ts b/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/register-customer.ts index a79e8c96..0edd2e72 100644 --- a/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/register-customer.ts +++ b/packages/core/test-types/fixtures/order-app/app/domain/customer/commands/register-customer.ts @@ -1,7 +1,7 @@ import type { Command } from "./+types/register-customer"; -export const payload = ({ z }: Command.PayloadArgs) => - z.object({ customerId: z.uuid(), email: z.email() }); +export const payload = ({ z, personal }: Command.PayloadArgs) => + z.object({ customerId: z.uuid(), email: personal(z.email()) }); export const handler = ({ command, events }: Command.HandlerArgs) => [ events.customerRegistered({ email: command.payload.email }), diff --git a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/cancel-order.ts b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/cancel-order.ts index c5512dcb..123ba8a1 100644 --- a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/cancel-order.ts +++ b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/cancel-order.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../cancel-order.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "CancelOrder", core.PayloadOf, diff --git a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/pay-order.ts b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/pay-order.ts index 7481c2d5..a020199d 100644 --- a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/pay-order.ts +++ b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/pay-order.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../pay-order.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "PayOrder", core.PayloadOf, diff --git a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/place-order.ts b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/place-order.ts index b8c40ea1..82a1c787 100644 --- a/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/place-order.ts +++ b/packages/core/test-types/fixtures/order-app/app/domain/order/commands/+types/place-order.ts @@ -4,7 +4,7 @@ import type * as generated from "../../../../../.bounda/types.ts"; type Module = typeof import("../place-order.ts"); export declare namespace Command { - type PayloadArgs = core.PayloadArgs; + type PayloadArgs = core.CommandPayloadArgs; type RejectionsArgs = core.CommandRejectionsArgs< "PlaceOrder", core.PayloadOf, diff --git a/packages/core/test-types/inference.test-d.ts b/packages/core/test-types/inference.test-d.ts index 6639f3b0..bbf3e9ba 100644 --- a/packages/core/test-types/inference.test-d.ts +++ b/packages/core/test-types/inference.test-d.ts @@ -2,10 +2,13 @@ import type { AppRegistry, BoundaApp, BoundaClient, + CancelledCommand, CatchUpReadModelsArgs, Clock, CommandHandlerArgs, CommandInvoker, + CommandPayloadArgs, + CommandPersonalFunction, CommandsFacade, Consistency, CreateArgs, @@ -25,6 +28,7 @@ import type { Instant, Logger, PayloadArgs, + PayloadInputOf, PersonalDataForgottenEvent, PersonalDataForgottenPayload, PersonalFunction, @@ -37,6 +41,7 @@ import type { RejectedDispatch, RejectFunction, RejectionCodeOf, + RemovedDeadLetter, ScheduledDispatch, StoredDispatch, StoredEvent, @@ -437,15 +442,30 @@ describe("personal data", () => { type OrderPlacedModule = typeof import("./fixtures/order-app/app/domain/order/order-placed.ts"); type OrderPaidModule = typeof import("./fixtures/order-app/app/domain/order/order-paid.ts"); - it("gives an event's payload the marks, and a command's or a query's none", () => { + it("gives an event's and a command's payload the marks, and a query's none", () => { expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); - expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); expectTypeOf().not.toHaveProperty("personal"); }); + it("hands a command's handler its personal fields as its schema gives them, never null", () => { + type RegisterCustomerModule = + typeof import("./fixtures/order-app/app/domain/customer/commands/register-customer.ts"); + expectTypeOf().toEqualTypeOf<{ + customerId: string; + email: string; + }>(); + expectTypeOf>().toEqualTypeOf<{ + customerId: string; + email: string; + }>(); + }); + it("writes a personal field as its schema takes it and reads it as that or null", () => { expectTypeOf() .parameter(0) @@ -501,6 +521,19 @@ describe("personal data", () => { ProjectForgotten.Args["event"]["payload"] >().toEqualTypeOf(); expectTypeOf().toEqualTypeOf<"PersonalDataForgotten">(); + expectTypeOf().toEqualTypeOf<{ + readonly personId: string; + readonly cancelledCommands: readonly CancelledCommand[]; + readonly removedDeadLetters: readonly RemovedDeadLetter[]; + }>(); + expectTypeOf().toEqualTypeOf<{ + readonly commandId: string; + readonly type: string; + }>(); + expectTypeOf().toEqualTypeOf<{ + readonly id: string; + readonly commandType: string; + }>(); expectTypeOf().toEqualTypeOf>(); expectTypeOf< AuditOnPersonalDataForgotten.HandlerArgs["event"] diff --git a/packages/core/test-types/negatives.test-d.ts b/packages/core/test-types/negatives.test-d.ts index b2ce88b1..6d4d0709 100644 --- a/packages/core/test-types/negatives.test-d.ts +++ b/packages/core/test-types/negatives.test-d.ts @@ -22,6 +22,7 @@ import type { Process as AtNextReminder } from "./fixtures/order-app/app/domain/ import type { Process as OrderPayment } from "./fixtures/order-app/app/domain/order/processes/order-payment/+types/index.ts"; import type { Process as OnOrderPaid } from "./fixtures/order-app/app/domain/order/processes/order-payment/+types/on-order-paid.ts"; import type { Projection as ProjectCustomerRegistered } from "./fixtures/order-app/app/read/customer-directory/projections/customer/+types/customer-registered.ts"; +import type { Query as GetCustomer } from "./fixtures/order-app/app/read/customer-directory/queries/+types/get-customer.ts"; import type { Projection as ProjectOrderPlaced } from "./fixtures/order-app/app/read/order-summary/projections/order/+types/order-placed.ts"; @@ -177,11 +178,11 @@ describe("what does not compile", () => { void handler; }); - it("marking personal data outside an event, or reading it as if it could not be forgotten", () => { - // @ts-expect-error a command's payload has no personal: commands are not stored - const command = ({ z, personal }: RegisterCustomer.PayloadArgs) => + it("marking personal data in a query, or reading it as if it could not be forgotten", () => { + // @ts-expect-error a query's payload has no personal: queries are never stored + const query = ({ z, personal }: GetCustomer.PayloadArgs) => z.object({ email: personal(z.email()) }); - void command; + void query; const builder = ({ events }: RegisterCustomer.HandlerArgs) => [ // @ts-expect-error a personal field is written as its schema takes it, never as null events.customerRegistered({ email: null }), diff --git a/packages/postgresql/src/dead-letter-store.ts b/packages/postgresql/src/dead-letter-store.ts index 0f42a00c..32d9b05e 100644 --- a/packages/postgresql/src/dead-letter-store.ts +++ b/packages/postgresql/src/dead-letter-store.ts @@ -5,10 +5,12 @@ import type { DeadLetterKind, DeadLetterStatus, DeadLetterStore, + KeyedPerson, ListDeadLettersArgs, } from "@bounda-dev/core/adapter"; -import { postgresqlDialect } from "@bounda-dev/core/adapter/sql"; +import { byFirstFailure, postgresqlDialect } from "@bounda-dev/core/adapter/sql"; import type { PostgresqlDatabase } from "./database.ts"; +import { naming } from "./held-keys.ts"; export interface CreatePostgresqlDeadLetterStoreArgs { readonly db: PostgresqlDatabase; @@ -20,7 +22,7 @@ export interface CreatePostgresqlDeadLetterStoreFunction { } const COLUMNS = - '"id", "kind", "handler", "event_id", "event_type", "aggregate_type", "aggregate_id", "error_type", "error_message", "error_stack", "attempts", "first_failed_at", "last_failed_at", "status", "payload"'; + '"id", "kind", "handler", "event_id", "event_type", "aggregate_type", "aggregate_id", "error_type", "error_message", "error_stack", "attempts", "first_failed_at", "last_failed_at", "status", "payload", "persons"'; const toLetter = (row: Record): DeadLetter => ({ id: String(row.id), @@ -40,6 +42,7 @@ const toLetter = (row: Record): DeadLetter => ({ lastFailedAt: String(row.last_failed_at), status: String(row.status) as DeadLetterStatus, ...(row.payload === null || row.payload === undefined ? {} : { payload: row.payload }), + ...((row.persons as KeyedPerson[]).length === 0 ? {} : { persons: row.persons as KeyedPerson[] }), }); const filters = ( @@ -72,7 +75,7 @@ export const createPostgresqlDeadLetterStore: CreatePostgresqlDeadLetterStoreFun return { add: async (letter) => { await db.run( - `INSERT INTO ${table} (${COLUMNS}) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'failed', $14) ON CONFLICT ("id") DO NOTHING`, + `INSERT INTO ${table} (${COLUMNS}) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'failed', $14, $15) ON CONFLICT ("id") DO NOTHING`, [ letter.id, letter.kind, @@ -88,6 +91,7 @@ export const createPostgresqlDeadLetterStore: CreatePostgresqlDeadLetterStoreFun letter.firstFailedAt, letter.lastFailedAt, postgresqlDialect.encode("json", letter.payload), + postgresqlDialect.encode("json", letter.persons ?? []), ], ); return (await get(letter.id)) as DeadLetter; @@ -117,5 +121,9 @@ export const createPostgresqlDeadLetterStore: CreatePostgresqlDeadLetterStoreFun if (updated.length === 0) throw new DeadLetterSettledError({ id }); }, remove: (id) => db.run(`DELETE FROM ${table} WHERE "id" = $1`, [id]), + forget: async (person) => + (await db.all(`DELETE FROM ${table} WHERE ${naming("$1")} RETURNING ${COLUMNS}`, [person])) + .map(toLetter) + .sort(byFirstFailure), }; }; diff --git a/packages/postgresql/src/scheduler.ts b/packages/postgresql/src/scheduler.ts index 7fcc8721..4dffa9f2 100644 --- a/packages/postgresql/src/scheduler.ts +++ b/packages/postgresql/src/scheduler.ts @@ -1,12 +1,14 @@ import { type CausationContext, ScheduledClaimLostError } from "@bounda-dev/core"; import type { ClaimedCommand, + KeyedPerson, ScheduledClaim, ScheduledCommand, Scheduler, } from "@bounda-dev/core/adapter"; import { byExecuteAt, earliestDue, postgresqlDialect } from "@bounda-dev/core/adapter/sql"; import type { PostgresqlDatabase } from "./database.ts"; +import { naming } from "./held-keys.ts"; export interface CreatePostgresqlSchedulerArgs { readonly db: PostgresqlDatabase; @@ -18,19 +20,23 @@ export interface CreatePostgresqlSchedulerFunction { } const COLUMNS = - '"dedupe_key", "command_type", "aggregate_id", "payload", "execute_at", "context", "attempts"'; + '"dedupe_key", "command_type", "aggregate_id", "payload", "execute_at", "context", "attempts", "persons"'; -const toScheduled = (row: Record): ScheduledCommand => ({ - dedupeKey: String(row.dedupe_key), - command: { - type: String(row.command_type), - aggregateId: String(row.aggregate_id), - payload: row.payload, - }, - executeAt: String(row.execute_at), - context: row.context as CausationContext, - attempts: Number(row.attempts), -}); +const toScheduled = (row: Record): ScheduledCommand => { + const persons = row.persons as KeyedPerson[]; + return { + dedupeKey: String(row.dedupe_key), + command: { + type: String(row.command_type), + aggregateId: String(row.aggregate_id), + payload: row.payload, + }, + executeAt: String(row.execute_at), + context: row.context as CausationContext, + attempts: Number(row.attempts), + ...(persons.length === 0 ? {} : { persons }), + }; +}; const toClaimed = (row: Record): ClaimedCommand => ({ ...toScheduled(row), @@ -72,15 +78,16 @@ export const createPostgresqlScheduler: CreatePostgresqlSchedulerFunction = ({ d ); return { - schedule: ({ dedupeKey, command, executeAt, context, keepTimingOfSameCommand }) => + schedule: ({ dedupeKey, command, executeAt, context, keepTimingOfSameCommand, persons = [] }) => db.run( - `INSERT INTO ${table} (${COLUMNS}, "claimed_at", "last_error", "revision") VALUES ($1, $2, $3, $4, $5, $6, 0, NULL, NULL, 0) + `INSERT INTO ${table} (${COLUMNS}, "claimed_at", "last_error", "revision") VALUES ($1, $2, $3, $4, $5, $6, 0, $7, NULL, NULL, 0) ON CONFLICT ("dedupe_key") DO UPDATE SET "command_type" = excluded."command_type", "aggregate_id" = excluded."aggregate_id", "payload" = excluded."payload", "execute_at" = excluded."execute_at", "context" = excluded."context", + "persons" = excluded."persons", "attempts" = 0, "last_error" = NULL, "revision" = ${table}."revision" + 1 @@ -90,6 +97,7 @@ export const createPostgresqlScheduler: CreatePostgresqlSchedulerFunction = ({ d AND ${table}."payload" = excluded."payload" ${keepTimingOfSameCommand === true ? "" : `AND ${table}."execute_at" = excluded."execute_at"`} AND ${table}."context" = excluded."context" + AND ${table}."persons" = excluded."persons" )`, [ dedupeKey, @@ -98,6 +106,7 @@ export const createPostgresqlScheduler: CreatePostgresqlSchedulerFunction = ({ d postgresqlDialect.encode("json", command.payload), executeAt.toISOString(), postgresqlDialect.encode("json", context), + postgresqlDialect.encode("json", persons), ], ), cancel: (dedupeKey) => db.run(`DELETE FROM ${table} WHERE "dedupe_key" = $1`, [dedupeKey]), @@ -166,5 +175,9 @@ export const createPostgresqlScheduler: CreatePostgresqlSchedulerFunction = ({ d [limit ?? null, offset], ) ).map(toScheduled), + forget: async (person) => + (await db.all(`DELETE FROM ${table} WHERE ${naming("$1")} RETURNING ${COLUMNS}`, [person])) + .map(toScheduled) + .sort(byExecuteAt), }; }; diff --git a/packages/postgresql/src/schema.ts b/packages/postgresql/src/schema.ts index db4545c6..dcd2b939 100644 --- a/packages/postgresql/src/schema.ts +++ b/packages/postgresql/src/schema.ts @@ -120,7 +120,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "first_failed_at" text NOT NULL, "last_failed_at" text NOT NULL, "status" text NOT NULL, - "payload" jsonb + "payload" jsonb, + "persons" jsonb NOT NULL DEFAULT '[]' )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.deadLetters, "status")} ON ${tables.deadLetters} ("status")`, `CREATE TABLE IF NOT EXISTS ${tables.scheduledCommands} ( @@ -134,7 +135,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "claimed_at" text, "last_error" text, "revision" integer NOT NULL DEFAULT 0, - "claim_id" text + "claim_id" text, + "persons" jsonb NOT NULL DEFAULT '[]' )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.scheduledCommands, "execute_at")} ON ${tables.scheduledCommands} ("execute_at")`, `CREATE TABLE IF NOT EXISTS ${tables.idempotencyKeys} ( @@ -162,6 +164,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) // So forgetting a person finds the rows that name them without reading every row. `CREATE INDEX IF NOT EXISTS ${indexName(tables.snapshots, "persons")} ON ${tables.snapshots} USING gin ("persons" jsonb_path_ops)`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "persons")} ON ${tables.idempotencyKeys} USING gin ("persons" jsonb_path_ops)`, + `CREATE INDEX IF NOT EXISTS ${indexName(tables.scheduledCommands, "persons")} ON ${tables.scheduledCommands} USING gin ("persons" jsonb_path_ops)`, + `CREATE INDEX IF NOT EXISTS ${indexName(tables.deadLetters, "persons")} ON ${tables.deadLetters} USING gin ("persons" jsonb_path_ops)`, ]; export interface EnsureStorageSchemaArgs { @@ -195,6 +199,8 @@ export const ensureStorageSchema: EnsureStorageSchemaFunction = async ({ db, sch indexName(tables.idempotencyKeys, "expires_at"), indexName(tables.snapshots, "persons"), indexName(tables.idempotencyKeys, "persons"), + indexName(tables.scheduledCommands, "persons"), + indexName(tables.deadLetters, "persons"), ].map(unquoted); if (await relationsExist({ db, schema, names })) return; await inSchema({