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({