From f9fa0aa49f5450e43517924caba343d1a61a025f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sat, 31 Oct 2026 10:17:14 +0100 Subject: [PATCH 1/2] fix(core): seal a person's fields only under the key the state read them under A command that loaded a person's data before forget could append after another command had given them a fresh key, and seal the old data under it, readable again. The state now records the fingerprint of the key each person's data was read under, from the events the load opened or from the snapshot, and sealing under any other key throws ConcurrencyError, so the command is decided again. Snapshots and remembered rejections are written only while every person they name still holds that key. The personal key store keeps each key's fingerprint, so the SQL guards compare it in the same statement. Persons become { person, fingerprint }, with null for data staged by the same unit of work and not sealed yet, which also lets a unit that registers a person and decides another command on the same stream commit. --- .changeset/personal-key-identity.md | 6 + docs/src/content/docs/adapters/index.mdx | 2 +- .../src/content/docs/concepts/long-streams.md | 7 +- docs/src/content/docs/guides/personal-data.md | 18 ++- docs/src/content/docs/reference/errors.md | 6 +- .../core/src/adapter/deferred-writes.test.ts | 17 ++- packages/core/src/adapter/deferred-writes.ts | 2 +- packages/core/src/adapter/index.ts | 6 +- packages/core/src/adapter/sqlite/held-keys.ts | 21 +++ .../src/adapter/sqlite/idempotency-store.ts | 23 +-- .../src/adapter/sqlite/personal-key-store.ts | 16 +- packages/core/src/adapter/sqlite/schema.ts | 3 +- .../core/src/adapter/sqlite/snapshot-store.ts | 13 +- .../src/adapter/staged-event-store.test.ts | 22 ++- .../core/src/adapter/staged-event-store.ts | 14 +- .../core/src/adapter/storage/event-store.ts | 14 +- .../src/adapter/storage/idempotency-store.ts | 14 +- .../src/adapter/storage/personal-key-store.ts | 46 +++++- .../src/adapter/storage/snapshot-store.ts | 17 ++- packages/core/src/adapter/testing/fixtures.ts | 25 ++++ .../testing/idempotency-store.contract.ts | 42 ++++-- .../testing/personal-key-store.contract.ts | 23 +-- .../testing/snapshot-store.contract.ts | 42 ++++-- .../testing/storage-transaction.contract.ts | 37 +++-- .../kernel/aggregate/load-aggregate.test.ts | 36 ++++- .../src/kernel/aggregate/load-aggregate.ts | 15 +- packages/core/src/kernel/command/pipeline.ts | 7 +- .../src/kernel/personal-data/crypto.test.ts | 23 ++- .../core/src/kernel/personal-data/crypto.ts | 33 +++-- .../kernel/personal-data/event-store.test.ts | 83 +++++++---- .../src/kernel/personal-data/event-store.ts | 28 +++- .../src/kernel/personal-data/forget.test.ts | 3 +- .../personal-data/personal-data.test.ts | 139 ++++++++++++++++-- .../src/kernel/personal-data/persons.test.ts | 18 ++- .../core/src/kernel/personal-data/persons.ts | 34 +++-- packages/core/src/memory/idempotency-store.ts | 8 +- .../core/src/memory/personal-key-store.ts | 20 ++- packages/core/src/memory/snapshot-store.ts | 8 +- packages/postgresql/src/held-keys.ts | 28 ++++ packages/postgresql/src/idempotency-store.ts | 13 +- packages/postgresql/src/personal-key-store.ts | 16 +- packages/postgresql/src/schema.ts | 7 +- packages/postgresql/src/snapshot-store.ts | 22 ++- 43 files changed, 723 insertions(+), 254 deletions(-) create mode 100644 .changeset/personal-key-identity.md create mode 100644 packages/core/src/adapter/sqlite/held-keys.ts create mode 100644 packages/postgresql/src/held-keys.ts diff --git a/.changeset/personal-key-identity.md b/.changeset/personal-key-identity.md new file mode 100644 index 00000000..48e534a2 --- /dev/null +++ b/.changeset/personal-key-identity.md @@ -0,0 +1,6 @@ +--- +"@bounda-dev/core": minor +"@bounda-dev/postgresql": minor +--- + +A command that loaded a person's data before `forget` no longer writes it back under the fresh key another command gave them meanwhile: the state it loaded remembers the key each person's data was read under, and sealing their fields under any other key fails with `ConcurrencyError`, so the command is decided again on the state without their data. Snapshots and remembered rejections keep the same: each names the key it read every person under, and is written only while that key is still theirs. Breaking for adapters: `PersonalKeyStore` keeps a `StoredPersonalKey` (`{ key, fingerprint }`) per person, and `personal_keys` gains a `fingerprint` column; the `persons` of `Snapshot`, `IdempotencyRecord` and `AppendArgs` are `KeyedPerson[]` (`{ person, fingerprint }`, `null` for data staged by the same unit of work and not sealed yet), stored as JSON in SQLite and as `jsonb` in PostgreSQL; `LoadResult` gains `fingerprints`, which the kernel fills and a store leaves out. A unit of work that registered a person and then decided another command on the same stream no longer fails its commit. diff --git a/docs/src/content/docs/adapters/index.mdx b/docs/src/content/docs/adapters/index.mdx index 4dfbaf93..5cac2841 100644 --- a/docs/src/content/docs/adapters/index.mdx +++ b/docs/src/content/docs/adapters/index.mdx @@ -89,7 +89,7 @@ On first use the adapter creates its tables, prefixed with `bounda_` by default: | `bounda_scheduled_commands` | Scheduled commands, process deadlines and timeouts, and delayed policy runs | | `bounda_dead_letters` | Handler runs that gave up, with the error and the attempt count | | `bounda_idempotency_keys` | What a command dispatched with an `idempotencyKey` answered, until the key expires | -| `bounda_personal_keys` | One key per person, which their [personal fields](/guides/personal-data/) are encrypted with; `forget` deletes it | +| `bounda_personal_keys` | One key per person, which their [personal fields](/guides/personal-data/) are encrypted with, and its fingerprint; `forget` deletes it | | `bounda_snapshots` | The last snapshot of every stream of an aggregate that takes them, with the fingerprint it is valid for | Read models get one table each, named after the read model in snake_case behind `rm_`, so no diff --git a/docs/src/content/docs/concepts/long-streams.md b/docs/src/content/docs/concepts/long-streams.md index 5d98ab60..aff9ef6d 100644 --- a/docs/src/content/docs/concepts/long-streams.md +++ b/docs/src/content/docs/concepts/long-streams.md @@ -164,9 +164,10 @@ state cannot be kept: the subscriber logs it and writes no snapshot for that str throws `SnapshotError` from `runUntilIdle()`. Keep the state to data. The state is kept as it was folded, in the clear, [personal fields](/guides/personal-data/) -included. Each snapshot names the people whose fields it holds, and forgetting one of them -removes their snapshots in the same transaction that removes their key, so no snapshot keeps -what the events no longer show. +included. Each snapshot names the people whose fields it holds and the key it read each one +under, and forgetting one of them removes their snapshots in the same transaction that removes +their key, so no snapshot keeps what the events no longer show; one folded before that +transaction is never written after it, even once the person has a new key. ## What snapshots do not do diff --git a/docs/src/content/docs/guides/personal-data.md b/docs/src/content/docs/guides/personal-data.md index 987f1416..74bd0c2e 100644 --- a/docs/src/content/docs/guides/personal-data.md +++ b/docs/src/content/docs/guides/personal-data.md @@ -80,20 +80,24 @@ empty id is refused with `ValidationError`. From then on every personal field of theirs reads as `null`, in every event, however old. A person registered again gets a fresh key; their old events stay unreadable. A command that loaded -their data before `forget` and appends after it is not the one to give them that key: its append -fails with [`ConcurrencyError`](/reference/errors/), as on a stream that moved, and the retry -decides on the state without their data. +their data before `forget` and appends after it never writes that data back, not even under the +fresh key another command gave them meanwhile: the state it loaded remembers the key each +person's data was read under, and an append that would seal their fields under any other fails +with [`ConcurrencyError`](/reference/errors/), as on a stream that moved. The retry decides on the +state without their data. 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 none is written for a person without a key. The next command on such an +people it holds and the key it read each one under, and none is written for a person who no +longer holds that key. The next command on such an aggregate folds it from its first event, with the fields forgotten, and the snapshot after it holds them as `null`. A command dispatched with an [`idempotencyKey`](/guides/idempotent-commands/) remembers what it answered, and a rejection's message may quote the state it was decided on. Such a -rejection names the people whose fields were in that state, and `forget` removes it too: the same -request sent again is decided again, on the state without them. What it cannot see is personal +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 message and say what went wrong with its code and the ids involved. @@ -142,7 +146,7 @@ typed with `PersonalDataForgottenEvent`. ## Where the keys live Each adapter keeps the keys in a table next to the events, `bounda_personal_keys`, one row per -person: the memory adapter in memory, SQLite and the Durable Object in their database, PostgreSQL +person with the key and its fingerprint, a SHA-256 of it: the memory adapter in memory, SQLite and the Durable Object in their database, PostgreSQL in its schema. A key is created when the first event with a personal field of that person is stored, by the transaction that stores it, and deleted by `forget`. Encryption is AES-256-GCM through WebCrypto, with a nonce per value, on Node and in `workerd` alike; what the store holds diff --git a/docs/src/content/docs/reference/errors.md b/docs/src/content/docs/reference/errors.md index 2e72dc57..908ebb1a 100644 --- a/docs/src/content/docs/reference/errors.md +++ b/docs/src/content/docs/reference/errors.md @@ -40,9 +40,9 @@ dead letter with `errorType: "retriable_exhausted"`. did not validate. `issues` lists each problem with its `path`. - **`CONCURRENCY_CONFLICT`** (`ConcurrencyError`; retried). An append found the stream at another version, or, with both versions equal, a person whose [personal data](/guides/personal-data/) - the command loaded was forgotten before it appended; after - `runtime.commands.concurrencyRetries` runs of the command. `streamId`, `expectedVersion`, - `actualVersion`. + the command loaded was forgotten before it appended, whether or not they hold a new key by + then; after `runtime.commands.concurrencyRetries` runs of the command. `streamId`, + `expectedVersion`, `actualVersion`. - **`HANDLER_TIMEOUT`** (no exported class; retried). A command, policy or process handler ran out of time (`runtime.commands.timeout`, `runtime.policies.timeout`, `runtime.processes.handlerTimeout`). diff --git a/packages/core/src/adapter/deferred-writes.test.ts b/packages/core/src/adapter/deferred-writes.test.ts index 4048b41f..7a550ba1 100644 --- a/packages/core/src/adapter/deferred-writes.test.ts +++ b/packages/core/src/adapter/deferred-writes.test.ts @@ -2,10 +2,15 @@ import { describe, expect, it } from "vitest"; import { silentLogger } from "../contracts/logger.ts"; import { memory } from "../memory/index.ts"; import { deferWrites } from "./deferred-writes.ts"; -import { testOutcome } from "./testing/fixtures.ts"; +import type { KeyedPerson } from "./storage/personal-key-store.ts"; +import { + testKeyedPerson as held, + testPersonalKey as stored, + testOutcome, +} from "./testing/fixtures.ts"; const now = new Date("2026-01-01T00:00:00.000Z"); -const record = (key: string, persons: readonly string[]) => ({ +const record = (key: string, persons: readonly KeyedPerson[]) => ({ key, fingerprint: "fp", outcome: testOutcome(key), @@ -17,14 +22,16 @@ const record = (key: string, persons: readonly string[]) => ({ describe("deferWrites", () => { it("forgets the idempotency records it holds back for a person, and forgets them on flush", async () => { const storage = await memory().createStorage({ logger: silentLogger }); - await storage.personalKeys.add("c-1", "k-1"); - await storage.idempotencyStore.put(record("live", ["c-1"])); + await storage.personalKeys.add("c-1", stored("1")); + await storage.idempotencyStore.put(record("live", [held("c-1", "1")])); const { stores, flush } = deferWrites(storage); - await stores.idempotencyStore.put(record("held", ["c-1"])); + await stores.idempotencyStore.put(record("held", [held("c-1", "1")])); await stores.idempotencyStore.put(record("other", [])); + await stores.idempotencyStore.put(record("bo", [held("c-2", "2")])); await stores.idempotencyStore.forget("c-1"); expect(await stores.idempotencyStore.get({ key: "held", now })).toBeNull(); expect(await stores.idempotencyStore.get({ key: "other", now })).not.toBeNull(); + expect(await stores.idempotencyStore.get({ key: "bo", now })).not.toBeNull(); expect(await storage.idempotencyStore.get({ key: "live", now })).not.toBeNull(); await flush(storage); expect(await storage.idempotencyStore.get({ key: "live", now })).toBeNull(); diff --git a/packages/core/src/adapter/deferred-writes.ts b/packages/core/src/adapter/deferred-writes.ts index ec304994..f59f1eee 100644 --- a/packages/core/src/adapter/deferred-writes.ts +++ b/packages/core/src/adapter/deferred-writes.ts @@ -103,7 +103,7 @@ export const deferWrites: DeferWritesFunction = ({ purge: idempotencyStore.purge, forget: async (person) => { for (const [key, record] of records) { - if (record.persons.includes(person)) records.delete(key); + if (record.persons.some((held) => held.person === person)) records.delete(key); } recorded.push((target) => target.idempotencyStore.forget(person)); }, diff --git a/packages/core/src/adapter/index.ts b/packages/core/src/adapter/index.ts index a943e9ee..b4a8e46d 100644 --- a/packages/core/src/adapter/index.ts +++ b/packages/core/src/adapter/index.ts @@ -53,7 +53,11 @@ export type { RenewClaimArgs, SettleClaimArgs, } from "./storage/inbox-ledger.ts"; -export type { PersonalKeyStore } from "./storage/personal-key-store.ts"; +export type { + KeyedPerson, + PersonalKeyStore, + StoredPersonalKey, +} from "./storage/personal-key-store.ts"; export type { ClaimDueArgs, ClaimedCommand, diff --git a/packages/core/src/adapter/sqlite/held-keys.ts b/packages/core/src/adapter/sqlite/held-keys.ts new file mode 100644 index 00000000..96b21899 --- /dev/null +++ b/packages/core/src/adapter/sqlite/held-keys.ts @@ -0,0 +1,21 @@ +export interface HeldKeysFunction { + (personalKeys: string): string; +} + +/** + * A condition true when every person in the JSON array of the first parameter holds, in + * `personalKeys`, the key of the fingerprint named with them, or any key for a `null` one; the + * second parameter is the array's length. A person named under two keys never holds both. + */ +export const heldKeys: HeldKeysFunction = (personalKeys) => + `(SELECT count(*) FROM ${personalKeys} AS "k" JOIN json_each(?) AS "p" ON "k"."person" = json_extract("p"."value", '$.person') AND "k"."fingerprint" = coalesce(json_extract("p"."value", '$.fingerprint'), "k"."fingerprint")) = ?`; + +export interface NamingFunction { + (table: string): string; +} + +/** + * A condition true for a row of `table` whose `persons` names the person of the parameter. + */ +export const naming: NamingFunction = (table) => + `EXISTS (SELECT 1 FROM json_each(${table}."persons") WHERE json_extract("value", '$.person') = ?)`; diff --git a/packages/core/src/adapter/sqlite/idempotency-store.ts b/packages/core/src/adapter/sqlite/idempotency-store.ts index 990ac958..73b6ba29 100644 --- a/packages/core/src/adapter/sqlite/idempotency-store.ts +++ b/packages/core/src/adapter/sqlite/idempotency-store.ts @@ -1,6 +1,12 @@ import { IdempotencyKeyTakenError } from "../../contracts/errors.ts"; -import type { DispatchOutcome, IdempotencyRecord, IdempotencyStore } from "../index.ts"; +import type { + DispatchOutcome, + IdempotencyRecord, + IdempotencyStore, + KeyedPerson, +} from "../index.ts"; import type { SqlDatabase } from "../sql/database.ts"; +import { heldKeys, naming } from "./held-keys.ts"; export interface CreateSqliteIdempotencyStoreArgs { readonly db: SqlDatabase; @@ -18,7 +24,7 @@ const toRecord = (row: Record): IdempotencyRecord => ({ fingerprint: String(row.fingerprint), outcome: JSON.parse(String(row.outcome)) as DispatchOutcome, expiresAt: String(row.expires_at), - persons: JSON.parse(String(row.persons)) as string[], + persons: JSON.parse(String(row.persons)) as KeyedPerson[], }); export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction = ({ @@ -36,12 +42,13 @@ export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction return { get, // One statement whose row comes back only when the key was free or expired and every person - // the record names has a key, so two dispatches racing for it can never both record, nor one - // record beside a forget. No row back with no live record under the key is the second case. + // the record names holds the key they were read under, so two dispatches racing for it can + // never both record, nor one record beside a forget. No row back with no live record under + // the key is the second case. put: async ({ key, fingerprint, outcome, expiresAt, now, persons }) => { const rows = await db.all( `INSERT INTO ${table} ("key", "fingerprint", "outcome", "expires_at", "persons") - SELECT ?, ?, ?, ?, ? WHERE (SELECT count(*) FROM ${personalKeys} WHERE "person" IN (SELECT "value" FROM json_each(?))) = ? + SELECT ?, ?, ?, ?, ? WHERE ${heldKeys(personalKeys)} ON CONFLICT ("key") DO UPDATE SET "fingerprint" = excluded."fingerprint", "outcome" = excluded."outcome", @@ -70,10 +77,6 @@ export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction ]); return rows.length; }, - forget: (person) => - db.run( - `DELETE FROM ${table} WHERE EXISTS (SELECT 1 FROM json_each(${table}."persons") WHERE "value" = ?)`, - [person], - ), + forget: (person) => db.run(`DELETE FROM ${table} WHERE ${naming(table)}`, [person]), }; }; diff --git a/packages/core/src/adapter/sqlite/personal-key-store.ts b/packages/core/src/adapter/sqlite/personal-key-store.ts index 3776da1d..37216feb 100644 --- a/packages/core/src/adapter/sqlite/personal-key-store.ts +++ b/packages/core/src/adapter/sqlite/personal-key-store.ts @@ -1,5 +1,5 @@ import type { SqlExecutor } from "../sql/sql-table.ts"; -import type { PersonalKeyStore } from "../storage/personal-key-store.ts"; +import type { PersonalKeyStore, StoredPersonalKey } from "../storage/personal-key-store.ts"; export interface CreateSqlitePersonalKeyStoreArgs { readonly db: SqlExecutor; @@ -14,17 +14,21 @@ export const createSqlitePersonalKeyStore: CreateSqlitePersonalKeyStoreFunction db, table, }) => { - const get = async (person: string): Promise => { - const [row] = await db.all(`SELECT "key" FROM ${table} WHERE "person" = ?`, [person]); - return row === undefined ? null : String(row.key); + const get = async (person: string): Promise => { + const [row] = await db.all(`SELECT "key", "fingerprint" FROM ${table} WHERE "person" = ?`, [ + person, + ]); + return row === undefined + ? null + : { key: String(row.key), fingerprint: String(row.fingerprint) }; }; return { get, // Whichever writer inserted first, every one answers with the key in force. add: async (person, key) => { await db.run( - `INSERT INTO ${table} ("person", "key") VALUES (?, ?) ON CONFLICT ("person") DO NOTHING`, - [person, key], + `INSERT INTO ${table} ("person", "key", "fingerprint") VALUES (?, ?, ?) ON CONFLICT ("person") DO NOTHING`, + [person, key.key, key.fingerprint], ); return (await get(person)) ?? key; }, diff --git a/packages/core/src/adapter/sqlite/schema.ts b/packages/core/src/adapter/sqlite/schema.ts index 3e664847..64e90bf0 100644 --- a/packages/core/src/adapter/sqlite/schema.ts +++ b/packages/core/src/adapter/sqlite/schema.ts @@ -116,7 +116,8 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "expires_at")} ON ${tables.idempotencyKeys} ("expires_at")`, `CREATE TABLE IF NOT EXISTS ${tables.personalKeys} ( "person" TEXT PRIMARY KEY, - "key" TEXT NOT NULL + "key" TEXT NOT NULL, + "fingerprint" TEXT NOT NULL )`, `CREATE TABLE IF NOT EXISTS ${tables.snapshots} ( "aggregate_type" TEXT NOT NULL, diff --git a/packages/core/src/adapter/sqlite/snapshot-store.ts b/packages/core/src/adapter/sqlite/snapshot-store.ts index df744413..29f91832 100644 --- a/packages/core/src/adapter/sqlite/snapshot-store.ts +++ b/packages/core/src/adapter/sqlite/snapshot-store.ts @@ -1,5 +1,6 @@ -import type { Snapshot, SnapshotStore } from "../index.ts"; +import type { KeyedPerson, Snapshot, SnapshotStore } from "../index.ts"; import type { SqlExecutor } from "../sql/sql-table.ts"; +import { heldKeys, naming } from "./held-keys.ts"; export interface CreateSqliteSnapshotStoreArgs { readonly db: SqlExecutor; @@ -18,7 +19,7 @@ const toSnapshot = (row: Record): Snapshot => ({ version: Number(row.version), fingerprint: String(row.fingerprint), state: String(row.state), - persons: JSON.parse(String(row.persons)) as string[], + persons: JSON.parse(String(row.persons)) as KeyedPerson[], }); export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ @@ -36,7 +37,7 @@ export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ // One statement, so a forget that removed a key before it runs leaves it nothing to write. put: ({ aggregateType, aggregateId, version, fingerprint, state, persons }) => db.run( - `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT ?, ?, ?, ?, ?, ? WHERE (SELECT count(*) FROM ${personalKeys} WHERE "person" IN (SELECT "value" FROM json_each(?))) = ? ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT ?, ?, ?, ?, ?, ? WHERE ${heldKeys(personalKeys)} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, [ aggregateType, aggregateId, @@ -53,9 +54,5 @@ export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ aggregateType, aggregateId, ]), - forget: (person) => - db.run( - `DELETE FROM ${table} WHERE EXISTS (SELECT 1 FROM json_each(${table}."persons") WHERE "value" = ?)`, - [person], - ), + forget: (person) => db.run(`DELETE FROM ${table} WHERE ${naming(table)}`, [person]), }); diff --git a/packages/core/src/adapter/staged-event-store.test.ts b/packages/core/src/adapter/staged-event-store.test.ts index 91319756..980fe18f 100644 --- a/packages/core/src/adapter/staged-event-store.test.ts +++ b/packages/core/src/adapter/staged-event-store.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import { ConcurrencyError } from "../contracts/errors.ts"; import { createMemoryEventStore } from "../memory/event-store.ts"; import { createStagedEventStore } from "./staged-event-store.ts"; -import { pendingEvent } from "./testing/fixtures.ts"; +import { testKeyedPerson as held, pendingEvent } from "./testing/fixtures.ts"; const order = { aggregateType: "order", aggregateId: "1" }; const customer = { aggregateType: "customer", aggregateId: "1" }; @@ -30,6 +30,17 @@ const setUp = async () => { }; describe("createStagedEventStore", () => { + it("reports the keys the base store's load opened, and none it was not given", async () => { + const { base } = await setUp(); + const fingerprints = new Map([["c-1", "f-1"]]); + const opening = createStagedEventStore({ + ...base, + load: async (args) => ({ ...(await base.load(args)), fingerprints }), + }); + expect((await opening.load(order)).fingerprints).toBe(fingerprints); + expect(await createStagedEventStore(base).load(order)).not.toHaveProperty("fingerprints"); + }); + it("loads one stream: the base store's events plus what was staged for it, at the version both make", async () => { const { staged } = await setUp(); const before = await staged.load(order); @@ -160,13 +171,13 @@ describe("createStagedEventStore", () => { ...order, expectedVersion: 1, events: [pendingEvent({ aggregateId: "1", version: 2, type: "OrderPaid" })], - persons: ["c-1"], + persons: [held("c-1", "1"), held("c-3", null)], }); await staged.append({ ...order, expectedVersion: 2, events: [pendingEvent({ aggregateId: "1", version: 3, type: "OrderArchived" })], - persons: ["c-2", "c-1"], + persons: [held("c-2", "2"), held("c-1", "1"), held("c-1", "again"), held("c-3", null)], }); const fresh = { aggregateType: "payment", aggregateId: "9" }; await staged.append({ @@ -174,7 +185,10 @@ describe("createStagedEventStore", () => { expectedVersion: 0, events: [pendingEvent({ ...fresh, version: 1, type: "PaymentRequested" })], }); - expect(staged.batches().map((batch) => batch.persons)).toEqual([["c-1", "c-2"], []]); + expect(staged.batches().map((batch) => batch.persons)).toEqual([ + [held("c-1", "1"), held("c-3", null), held("c-2", "2"), held("c-1", "again")], + [], + ]); }); it("reads the global stream from the base store alone", async () => { diff --git a/packages/core/src/adapter/staged-event-store.ts b/packages/core/src/adapter/staged-event-store.ts index 6f7aa943..3bb2ff5c 100644 --- a/packages/core/src/adapter/staged-event-store.ts +++ b/packages/core/src/adapter/staged-event-store.ts @@ -1,6 +1,7 @@ import { ConcurrencyError } from "../contracts/errors.ts"; import { type StoredEvent, streamId } from "../contracts/event.ts"; import type { AppendArgs, EventStore } from "./storage/event-store.ts"; +import { type KeyedPerson, keyedPersonId } from "./storage/personal-key-store.ts"; /** * An event store that keeps what is appended to it aside: `load` answers with the base store's @@ -38,7 +39,7 @@ interface StagedStream { readonly aggregateId: string; readonly base: number; readonly events: StoredEvent[]; - readonly persons: Set; + readonly persons: Map; } export const createStagedEventStore: CreateStagedEventStoreFunction = ( @@ -63,7 +64,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = ( aggregateId, base: version, events: [], - persons: new Set(), + persons: new Map(), }; streams.set(key, created); return created; @@ -81,7 +82,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = ( aggregateId, base: stored.version, events: [], - persons: new Set(), + persons: new Map(), }; streams.set(key, created); return created; @@ -92,6 +93,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = ( ...staged.events.filter((event) => event.version >= fromVersion), ], version: staged.base + staged.events.length, + ...(stored.fingerprints === undefined ? {} : { fingerprints: stored.fingerprints }), }; }, append: async ({ aggregateType, aggregateId, expectedVersion, events, persons = [] }) => { @@ -106,7 +108,9 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = ( } const stored = events.map((event) => ({ ...event, position: nextPosition() })); staged.events.push(...stored); - for (const person of persons) staged.persons.add(person); + for (const held of persons) { + staged.persons.set(keyedPersonId(held), held); + } return { version: actualVersion + stored.length, events: stored }; }, readAll: (args) => base.readAll(args), @@ -119,7 +123,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = ( aggregateId, expectedVersion, events: events.map(({ position: _position, ...event }) => event), - persons: [...persons], + persons: [...persons.values()], positions: events.map((event) => event.position), })), }; diff --git a/packages/core/src/adapter/storage/event-store.ts b/packages/core/src/adapter/storage/event-store.ts index fccd123a..dc3582f1 100644 --- a/packages/core/src/adapter/storage/event-store.ts +++ b/packages/core/src/adapter/storage/event-store.ts @@ -1,4 +1,5 @@ import type { StoredEvent } from "../../contracts/event.ts"; +import type { KeyedPerson } from "./personal-key-store.ts"; /** * An event the kernel hands to the store: everything except the global position, which the store @@ -15,11 +16,11 @@ export interface AppendArgs { readonly expectedVersion: number; readonly events: readonly PendingEvent[]; /** - * The people whose personal data the state the events were decided on held. The kernel reads - * it before the store does, to refuse sealing their fields once they are forgotten; a store - * ignores it. + * The people whose personal data the state the events were decided on held, with the key each + * was read under. The kernel reads it before the store does, to refuse sealing their fields + * under any other key, as after they are forgotten; a store ignores it. */ - readonly persons?: readonly string[]; + readonly persons?: readonly KeyedPerson[]; } export interface AppendResult { @@ -36,6 +37,11 @@ export interface LoadArgs { export interface LoadResult { readonly events: readonly StoredEvent[]; readonly version: number; + /** + * The fingerprint of the key that opened each person's sealed fields, set by the kernel, which + * opens them; a store leaves it out. + */ + readonly fingerprints?: ReadonlyMap; } export interface ReadAllArgs { diff --git a/packages/core/src/adapter/storage/idempotency-store.ts b/packages/core/src/adapter/storage/idempotency-store.ts index 01e13786..39f0734e 100644 --- a/packages/core/src/adapter/storage/idempotency-store.ts +++ b/packages/core/src/adapter/storage/idempotency-store.ts @@ -1,4 +1,5 @@ import type { DispatchResult, RejectedDispatch } from "../../contracts/command.ts"; +import type { KeyedPerson } from "./personal-key-store.ts"; /** * What a dispatch answered, as the idempotency store keeps it: the result of a command that was @@ -8,16 +9,16 @@ export type DispatchOutcome = DispatchResult | RejectedDispatch; /** * A dispatch remembered under its idempotency key: the fingerprint of the command it was, what it - * answered, and when the key may be used for another command. `persons` names, once each, the - * people whose personal fields were in the state a rejection was decided on, since its message - * may quote them; it is empty for any other outcome. + * answered, and when the key may be used for another command. `persons` names the people whose + * personal fields were in the state a rejection was decided on, since its message may quote + * them, each with the key they were read under; it is empty for any other outcome. */ export interface IdempotencyRecord { readonly key: string; readonly fingerprint: string; readonly outcome: DispatchOutcome; readonly expiresAt: string; - readonly persons: readonly string[]; + readonly persons: readonly KeyedPerson[]; } export interface GetIdempotencyRecordArgs { @@ -46,8 +47,9 @@ export interface PurgeIdempotencyRecordsArgs { * the recording was part of. * * A record that names people lives only as long as their keys, as a snapshot does: `put` records - * nothing unless every person it names has a key in the storage's `personalKeys`, checked in the - * same statement, and `forget` removes every record that names a person. + * nothing unless every person it names holds the key of the fingerprint it names them with, or + * any key for a `null` one, checked in the same statement, and `forget` removes every record + * that names a person. */ export interface IdempotencyStore { /** diff --git a/packages/core/src/adapter/storage/personal-key-store.ts b/packages/core/src/adapter/storage/personal-key-store.ts index caa6c9c0..c3830809 100644 --- a/packages/core/src/adapter/storage/personal-key-store.ts +++ b/packages/core/src/adapter/storage/personal-key-store.ts @@ -1,12 +1,44 @@ /** - * The keys the personal fields of events are encrypted with, one per person, as opaque text the - * kernel chooses. `get` is `null` for a person without one, as for one forgotten. `add` stores - * `key` for `person` unless one is in force already and resolves with the one in force, so two - * writers that find none keep the same key. `remove` forgets the key and reports whether there - * was one; what was encrypted with it can never be read again. + * A person's key as the key store keeps it: the key, as opaque text the kernel chooses, and its + * fingerprint, which names the key without revealing it. + */ +export interface StoredPersonalKey { + readonly key: string; + readonly fingerprint: string; +} + +/** + * A person whose personal data a state holds in the clear, with the fingerprint of the key it was + * read under; `null` when it was read from events not sealed yet, staged by the same unit of work. + * A person forgotten and given a new key since holds another fingerprint, so what the state holds + * of them must not be written back. + */ +export interface KeyedPerson { + readonly person: string; + readonly fingerprint: string | null; +} + +export interface KeyedPersonIdFunction { + (held: KeyedPerson): string; +} + +// Two entries for one person under two keys are two entries, never merged. +export const keyedPersonId: KeyedPersonIdFunction = ({ person, fingerprint }) => + JSON.stringify([person, fingerprint]); + +/** + * 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 + * force already and resolves with the one in force, so two writers that find none keep the same + * key. `remove` forgets the key and reports whether there was one; what was encrypted with it can + * never be read again. + * + * Stores that keep personal data in the clear, as snapshots do, compare a person's fingerprint + * with the one in force in the same statement as they write, so the store must keep it where + * they can read it. */ export interface PersonalKeyStore { - get(person: string): Promise; - add(person: string, key: string): Promise; + get(person: string): Promise; + add(person: string, key: StoredPersonalKey): Promise; remove(person: string): Promise; } diff --git a/packages/core/src/adapter/storage/snapshot-store.ts b/packages/core/src/adapter/storage/snapshot-store.ts index f1dc28cb..20a03ec5 100644 --- a/packages/core/src/adapter/storage/snapshot-store.ts +++ b/packages/core/src/adapter/storage/snapshot-store.ts @@ -1,15 +1,17 @@ import type { StreamIdentity } from "../../contracts/event.ts"; +import type { KeyedPerson } from "./personal-key-store.ts"; /** * The state of one aggregate stream folded up to `version`, as the kernel serialized it, valid - * only for the `fingerprint` of the code that folded it. `persons` names, once each, the people - * whose personal fields the events folded carried, since the state holds them in the clear. + * only for the `fingerprint` of the code that folded it. `persons` names the people whose personal + * fields the events folded carried, since the state holds them in the clear, each with the key + * they were read under. */ export interface Snapshot extends StreamIdentity { readonly version: number; readonly fingerprint: string; readonly state: string; - readonly persons: readonly string[]; + readonly persons: readonly KeyedPerson[]; } /** @@ -18,10 +20,11 @@ export interface Snapshot extends StreamIdentity { * instead. `get` is `null` for a stream without one. `remove` forgets a stream's snapshot. * * A snapshot holds personal data in the clear, so it lives only as long as its persons' keys: - * `put` writes nothing unless every person it names has a key in the storage's `personalKeys`, - * checked in the same statement, and `forget` removes every snapshot that names a person. A - * transaction that removes a key and then forgets its snapshots leaves none behind, whatever - * `put` runs beside it. + * `put` writes nothing unless every person it names holds, in the storage's `personalKeys`, the + * key of the fingerprint it names them with, or any key for a `null` one, checked in the same + * statement; and `forget` removes every snapshot that names a person. A transaction that removes + * a key and then forgets its snapshots leaves none behind, whatever `put` runs beside it, and a + * key given to the person afterwards is never the one a snapshot was read under. */ export interface SnapshotStore { get(stream: StreamIdentity): Promise; diff --git a/packages/core/src/adapter/testing/fixtures.ts b/packages/core/src/adapter/testing/fixtures.ts index a00a5fe6..9b85a713 100644 --- a/packages/core/src/adapter/testing/fixtures.ts +++ b/packages/core/src/adapter/testing/fixtures.ts @@ -2,6 +2,7 @@ import type { NewCommand, StoredDispatch } from "../../contracts/command.ts"; import type { CausationContext, EventMetadata } from "../../contracts/metadata.ts"; import type { NewDeadLetter } from "../storage/dead-letter-store.ts"; import type { PendingEvent } from "../storage/event-store.ts"; +import type { KeyedPerson, StoredPersonalKey } from "../storage/personal-key-store.ts"; /** * Metadata for events built in contract tests. @@ -106,3 +107,27 @@ export const testOutcome: TestOutcomeFunction = (aggregateId) => ({ eventTypes: ["OrderPlaced"], position: 1, }); + +export interface TestPersonalKeyFunction { + (name: string): StoredPersonalKey; +} + +/** + * A personal key and its fingerprint, both named after `name`, for contract tests. + */ +export const testPersonalKey: TestPersonalKeyFunction = (name) => ({ + key: `k-${name}`, + fingerprint: `f-${name}`, +}); + +export interface TestKeyedPersonFunction { + (person: string, key: string | null): KeyedPerson; +} + +/** + * A person read under the key `testPersonalKey(key)`, or under none for `null`. + */ +export const testKeyedPerson: TestKeyedPersonFunction = (person, key) => ({ + person, + fingerprint: key === null ? null : testPersonalKey(key).fingerprint, +}); diff --git a/packages/core/src/adapter/testing/idempotency-store.contract.ts b/packages/core/src/adapter/testing/idempotency-store.contract.ts index 6a4e0739..8e08172e 100644 --- a/packages/core/src/adapter/testing/idempotency-store.contract.ts +++ b/packages/core/src/adapter/testing/idempotency-store.contract.ts @@ -2,8 +2,8 @@ import { beforeEach, describe, expect, it } from "vitest"; import { IdempotencyKeyTakenError } from "../../contracts/errors.ts"; import type { Storage } from "../adapter.ts"; import type { IdempotencyStore } from "../storage/idempotency-store.ts"; -import type { PersonalKeyStore } from "../storage/personal-key-store.ts"; -import { testOutcome } from "./fixtures.ts"; +import type { KeyedPerson, PersonalKeyStore } from "../storage/personal-key-store.ts"; +import { testKeyedPerson as held, testPersonalKey as stored, testOutcome } from "./fixtures.ts"; export interface IdempotencyStoreContractArgs { /** @@ -46,11 +46,12 @@ export const idempotencyStoreContract: IdempotencyStoreContractFunction = ({ cre }; it("records a record that names people only while each of them has a key", async () => { - const named = { ...record("k1"), outcome: rejection, persons: ["c-1", "c-2"], now }; - await keys.add("c-1", "k-1"); + const persons = [held("c-1", "1"), held("c-2", null)]; + const named = { ...record("k1"), outcome: rejection, persons, now }; + await keys.add("c-1", stored("1")); await store.put(named); expect(await store.get({ key: "k1", now })).toBeNull(); - await keys.add("c-2", "k-2"); + await keys.add("c-2", stored("2")); await store.put(named); const { now: _now, ...kept } = named; expect(await store.get({ key: "k1", now })).toEqual(kept); @@ -60,12 +61,33 @@ export const idempotencyStoreContract: IdempotencyStoreContractFunction = ({ cre ); }); + it("records nothing for a person named with a key other than the one they hold", async () => { + const named = (...persons: KeyedPerson[]) => ({ + ...record("k1"), + outcome: rejection, + persons, + now, + }); + await keys.add("c-1", stored("new")); + await store.put(named(held("c-1", "old"))); + expect(await store.get({ key: "k1", now })).toBeNull(); + await store.put(named(held("c-1", "old"), held("c-1", "new"))); + expect(await store.get({ key: "k1", now })).toBeNull(); + await store.put(named(held("c-1", "new"), held("c-1", null))); + expect((await store.get({ key: "k1", now }))?.persons).toEqual([ + held("c-1", "new"), + held("c-1", null), + ]); + }); + it("forgets every record that names a person, and no other", async () => { - await keys.add("c-1", "k-1"); - await keys.add("c-2", "k-2"); - await store.put({ ...record("k1"), outcome: rejection, persons: ["c-1"], now }); - await store.put({ ...record("k2"), outcome: rejection, persons: ["c-2", "c-1"], now }); - await store.put({ ...record("k3"), outcome: rejection, persons: ["c-2"], now }); + await keys.add("c-1", stored("1")); + await keys.add("c-2", stored("2")); + const c1 = held("c-1", "1"); + const c2 = held("c-2", "2"); + await store.put({ ...record("k1"), outcome: rejection, persons: [c1], now }); + await store.put({ ...record("k2"), outcome: rejection, persons: [c2, c1], now }); + await store.put({ ...record("k3"), outcome: rejection, persons: [c2], now }); await store.put({ ...record("k4"), now }); await store.forget("c-1"); await store.forget("nobody"); diff --git a/packages/core/src/adapter/testing/personal-key-store.contract.ts b/packages/core/src/adapter/testing/personal-key-store.contract.ts index 8c2f15ff..c8d4826d 100644 --- a/packages/core/src/adapter/testing/personal-key-store.contract.ts +++ b/packages/core/src/adapter/testing/personal-key-store.contract.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it } from "vitest"; import type { PersonalKeyStore } from "../storage/personal-key-store.ts"; +import { testPersonalKey as stored } from "./fixtures.ts"; export interface PersonalKeyStoreContractArgs { readonly create: () => Promise; @@ -26,28 +27,30 @@ export const personalKeyStoreContract: PersonalKeyStoreContractFunction = ({ cre expect(await store.remove("c-1")).toBe(false); }); - it("keeps the first key added for a person and answers every add with it", async () => { - expect(await store.add("c-1", "k-first")).toBe("k-first"); - expect(await store.add("c-1", "k-second")).toBe("k-first"); - expect(await store.get("c-1")).toBe("k-first"); - expect(await store.add("c-2", "k-other")).toBe("k-other"); - expect(await store.get("c-2")).toBe("k-other"); + it("keeps the first key added for a person, with its fingerprint, and answers every add with it", async () => { + expect(await store.add("c-1", stored("first"))).toEqual(stored("first")); + expect(await store.add("c-1", stored("second"))).toEqual(stored("first")); + expect(await store.get("c-1")).toEqual(stored("first")); + expect(await store.add("c-2", stored("other"))).toEqual(stored("other")); + expect(await store.get("c-2")).toEqual(stored("other")); }); it("keeps one key when several writers add one at once", async () => { - const added = await Promise.all(["k-a", "k-b", "k-c"].map((key) => store.add("c-1", key))); + const added = await Promise.all( + ["a", "b", "c"].map((name) => store.add("c-1", stored(name))), + ); const kept = await store.get("c-1"); expect(kept).not.toBeNull(); expect(added).toEqual([kept, kept, kept]); }); it("forgets a key for good, and takes a new one for the same person afterwards", async () => { - await store.add("c-1", "k-first"); + await store.add("c-1", stored("first")); expect(await store.remove("c-1")).toBe(true); expect(await store.get("c-1")).toBeNull(); expect(await store.remove("c-1")).toBe(false); - expect(await store.add("c-1", "k-again")).toBe("k-again"); - expect(await store.get("c-1")).toBe("k-again"); + expect(await store.add("c-1", stored("again"))).toEqual(stored("again")); + expect(await store.get("c-1")).toEqual(stored("again")); }); }); }; diff --git a/packages/core/src/adapter/testing/snapshot-store.contract.ts b/packages/core/src/adapter/testing/snapshot-store.contract.ts index fdeceadf..14ba563f 100644 --- a/packages/core/src/adapter/testing/snapshot-store.contract.ts +++ b/packages/core/src/adapter/testing/snapshot-store.contract.ts @@ -1,7 +1,8 @@ import { beforeEach, describe, expect, it } from "vitest"; import type { Storage } from "../adapter.ts"; -import type { PersonalKeyStore } from "../storage/personal-key-store.ts"; +import type { KeyedPerson, PersonalKeyStore } from "../storage/personal-key-store.ts"; import type { Snapshot, SnapshotStore } from "../storage/snapshot-store.ts"; +import { testKeyedPerson as held, testPersonalKey as stored } from "./fixtures.ts"; export interface SnapshotStoreContractArgs { /** @@ -65,13 +66,13 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) }); it("writes a snapshot that names people only while each of them has a key", async () => { - const ada: Snapshot = { ...order42, persons: ["c-1", "c-2"] }; + const ada: Snapshot = { ...order42, persons: [held("c-1", "1"), held("c-2", null)] }; await store.put(ada); expect(await store.get(order42)).toBeNull(); - await keys.add("c-1", "k-1"); + await keys.add("c-1", stored("1")); await store.put(ada); expect(await store.get(order42)).toBeNull(); - await keys.add("c-2", "k-2"); + await keys.add("c-2", stored("2")); await store.put(ada); expect(await store.get(order42)).toEqual(ada); await keys.remove("c-2"); @@ -79,19 +80,36 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) expect(await store.get(order42)).toEqual(ada); }); + it("writes nothing for a person named with a key other than the one they hold", async () => { + const named = (...persons: KeyedPerson[]): Snapshot => ({ ...order42, persons }); + await keys.add("c-1", stored("new")); + await store.put(named(held("c-1", "old"))); + expect(await store.get(order42)).toBeNull(); + await store.put(named(held("c-1", "old"), held("c-1", "new"))); + expect(await store.get(order42)).toBeNull(); + await store.put(named(held("c-1", "new"), held("c-1", null))); + expect(await store.get(order42)).toEqual(named(held("c-1", "new"), held("c-1", null))); + await keys.remove("c-1"); + await keys.add("c-1", stored("newer")); + await store.put({ ...named(held("c-1", "new")), version: 300 }); + expect((await store.get(order42))?.version).toBe(order42.version); + }); + it("forgets every snapshot that names a person, and no other", async () => { - await keys.add("c-1", "k-1"); - await keys.add("c-2", "k-2"); - const named = (aggregateId: string, persons: readonly string[]): Snapshot => ({ + await keys.add("c-1", stored("1")); + await keys.add("c-2", stored("2")); + const c1 = held("c-1", "1"); + const c2 = held("c-2", "2"); + const named = (aggregateId: string, persons: readonly KeyedPerson[]): Snapshot => ({ ...order42, aggregateId, persons, }); - await store.put(named("1", ["c-1"])); - await store.put(named("2", ["c-2", "c-1"])); - await store.put(named("3", ["c-2"])); + await store.put(named("1", [c1])); + await store.put(named("2", [c2, c1])); + await store.put(named("3", [c2])); await store.put(named("4", [])); - await store.put({ ...named("1", ["c-1"]), aggregateType: "customer" }); + await store.put({ ...named("1", [c1]), aggregateType: "customer" }); await store.forget("c-1"); await store.forget("nobody"); const left = await Promise.all( @@ -99,7 +117,7 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) store.get({ aggregateType: "order", aggregateId }), ), ); - expect(left.map((snapshot) => snapshot?.persons ?? null)).toEqual([null, null, ["c-2"], []]); + expect(left.map((snapshot) => snapshot?.persons ?? null)).toEqual([null, null, [c2], []]); expect(await store.get({ aggregateType: "customer", aggregateId: "1" })).toBeNull(); }); diff --git a/packages/core/src/adapter/testing/storage-transaction.contract.ts b/packages/core/src/adapter/testing/storage-transaction.contract.ts index c5a170e9..50bde347 100644 --- a/packages/core/src/adapter/testing/storage-transaction.contract.ts +++ b/packages/core/src/adapter/testing/storage-transaction.contract.ts @@ -6,7 +6,16 @@ import { ScheduledClaimLostError, } from "../../contracts/errors.ts"; import type { Storage, StorageTransaction } from "../adapter.ts"; -import { pendingEvent, testCommand, testContext, testDeadLetter, testOutcome } from "./fixtures.ts"; +import type { KeyedPerson } from "../storage/personal-key-store.ts"; +import { + testKeyedPerson as held, + pendingEvent, + testPersonalKey as stored, + testCommand, + testContext, + testDeadLetter, + testOutcome, +} from "./fixtures.ts"; export interface StorageTransactionContractArgs { readonly create: () => Promise; @@ -131,7 +140,7 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ await expect( storage.transact(async (tx) => { await tx.inboxLedger.complete(key); - await tx.personalKeys.add("c-9", "k-9"); + await tx.personalKeys.add("c-9", stored("9")); await tx.scheduler.schedule({ dedupeKey: "command:c1", command: testCommand("1"), @@ -262,11 +271,11 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ }); it("adds and removes personal keys with the rest, or not at all", async () => { - await storage.personalKeys.add("c-1", "k-1"); + await storage.personalKeys.add("c-1", stored("1")); const work = async (tx: StorageTransaction) => { expect(await tx.personalKeys.remove("c-1")).toBe(true); - expect(await tx.personalKeys.add("c-2", "k-2")).toBe("k-2"); - expect(await tx.personalKeys.get("c-2")).toBe("k-2"); + expect(await tx.personalKeys.add("c-2", stored("2"))).toEqual(stored("2")); + expect(await tx.personalKeys.get("c-2")).toEqual(stored("2")); await tx.eventStore.append({ aggregateType: "order", aggregateId: "1", @@ -280,19 +289,19 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ throw new Error("boom"); }), ).rejects.toThrow("boom"); - expect(await storage.personalKeys.get("c-1")).toBe("k-1"); + expect(await storage.personalKeys.get("c-1")).toEqual(stored("1")); expect(await storage.personalKeys.get("c-2")).toBeNull(); expect(await storage.eventStore.lastPosition()).toBe(0); await storage.transact(work); expect(await storage.personalKeys.get("c-1")).toBeNull(); - expect(await storage.personalKeys.get("c-2")).toBe("k-2"); + expect(await storage.personalKeys.get("c-2")).toEqual(stored("2")); expect(await storage.eventStore.readAll({ afterPosition: 0, limit: 10 })).toHaveLength(1); }); it("forgets a person's snapshots and remembered rejections with their key", async () => { - await storage.personalKeys.add("c-1", "k-1"); - await storage.personalKeys.add("c-2", "k-2"); - const snapshot = (aggregateId: string, persons: readonly string[]) => ({ + await storage.personalKeys.add("c-1", stored("1")); + await storage.personalKeys.add("c-2", stored("2")); + const snapshot = (aggregateId: string, persons: readonly KeyedPerson[]) => ({ aggregateType: "customer", aggregateId, version: 1, @@ -300,8 +309,8 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ state: "[]", persons, }); - await storage.snapshotStore.put(snapshot("c-1", ["c-1"])); - await storage.snapshotStore.put(snapshot("c-2", ["c-2"])); + await storage.snapshotStore.put(snapshot("c-1", [held("c-1", "1")])); + await storage.snapshotStore.put(snapshot("c-2", [held("c-2", "2")])); const remembered = { key: "req-1", fingerprint: "fp", @@ -312,7 +321,7 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ aggregateId: "c-1", }, expiresAt: "2026-01-02T00:00:00.000Z", - persons: ["c-1"], + persons: [held("c-1", "1")], now, }; await storage.idempotencyStore.put(remembered); @@ -325,7 +334,7 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ expect(await storage.personalKeys.get("c-1")).toBeNull(); expect(await storage.snapshotStore.get(snapshot("c-1", []))).toBeNull(); expect(await storage.snapshotStore.get(snapshot("c-2", []))).toEqual( - snapshot("c-2", ["c-2"]), + snapshot("c-2", [held("c-2", "2")]), ); }); diff --git a/packages/core/src/kernel/aggregate/load-aggregate.test.ts b/packages/core/src/kernel/aggregate/load-aggregate.test.ts index 4802b64e..714d8bf8 100644 --- a/packages/core/src/kernel/aggregate/load-aggregate.test.ts +++ b/packages/core/src/kernel/aggregate/load-aggregate.test.ts @@ -1,10 +1,15 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { EventStore } from "../../adapter/storage/event-store.ts"; import type { Snapshot } from "../../adapter/storage/snapshot-store.ts"; -import { pendingEvent } from "../../adapter/testing/fixtures.ts"; +import { + testKeyedPerson as held, + pendingEvent, + testPersonalKey, +} from "../../adapter/testing/fixtures.ts"; import { resolveConfig } from "../../config/schema.ts"; import { SnapshotError } from "../../contracts/errors.ts"; import { memory } from "../../memory/index.ts"; +import type { EventPayloadArgs } from "../../modules/payload.ts"; import type { Registry } from "../../modules/registry.ts"; import { ATTRIBUTES, METRICS } from "../telemetry.ts"; import { type FakeTelemetry, installFakeTelemetry } from "../telemetry-fake.ts"; @@ -51,6 +56,11 @@ const setUp = async ({ orderPlaced: { ...orderAggregate.events.orderPlaced, evolve: evolves.placed }, orderPaid: { ...orderAggregate.events.orderPaid, evolve: evolves.paid }, orderArchived: { evolve: evolves.archived }, + orderSigned: { + payload: ({ z, personal, person }: EventPayloadArgs) => + z.object({ customerId: person(z.string()), signature: personal(z.string()) }), + evolve: ({ state }: { state: object }) => state, + }, }, }, }, @@ -144,6 +154,30 @@ describe("createAggregateLoader", () => { expect(outcomes()).toEqual(["used"]); }); + it("names whose personal data the state holds, with the key its load opened it with", async () => { + const { storage, snapshot, append, load } = await setUp(); + await storage.personalKeys.add("c-0", testPersonalKey("0")); + await storage.snapshotStore.put(snapshot({ persons: [held("c-0", "0")] })); + await append( + [3, "OrderSigned", { customerId: "c-1", signature: "Ana" }], + [4, "OrderSigned", { customerId: "c-2", signature: "Bo" }], + ); + const opening: EventStore = { + ...storage.eventStore, + load: async (args) => ({ + ...(await storage.eventStore.load(args)), + fingerprints: new Map([["c-1", testPersonalKey("1").fingerprint]]), + }), + }; + expect((await load(opening)).persons).toEqual([ + held("c-0", "0"), + held("c-1", "1"), + held("c-2", null), + ]); + await storage.snapshotStore.remove(stream); + expect((await load(opening)).persons).toEqual([held("c-1", "1"), held("c-2", null)]); + }); + it("ignores, but keeps, a snapshot taken by other code", async () => { const { storage, snapshot, load, outcomes, evolves } = await setUp(); await storage.snapshotStore.put(snapshot({ fingerprint: "fp:2" })); diff --git a/packages/core/src/kernel/aggregate/load-aggregate.ts b/packages/core/src/kernel/aggregate/load-aggregate.ts index 26747c3c..36433696 100644 --- a/packages/core/src/kernel/aggregate/load-aggregate.ts +++ b/packages/core/src/kernel/aggregate/load-aggregate.ts @@ -1,4 +1,5 @@ import type { EventStore } from "../../adapter/storage/event-store.ts"; +import type { KeyedPerson } from "../../adapter/storage/personal-key-store.ts"; import type { SnapshotStore } from "../../adapter/storage/snapshot-store.ts"; import { SnapshotError } from "../../contracts/errors.ts"; import type { StreamIdentity } from "../../contracts/event.ts"; @@ -26,9 +27,10 @@ export interface LoadedAggregate { */ readonly version: number; /** - * Whose personal fields went into the state, as a snapshot of it records them. + * Whose personal fields went into the state, and under which key, as a snapshot of it records + * them. */ - readonly persons: readonly string[]; + readonly persons: readonly KeyedPerson[]; } /** @@ -103,7 +105,7 @@ export const createAggregateLoader: CreateAggregateLoaderFunction = ({ return { folded: foldState({ aggregate, events }), version: Math.min(loaded.version, through), - persons: personsOf({ aggregate, events }), + persons: personsOf({ aggregate, events, fingerprints: loaded.fingerprints }), }; }; const dropped = async ( @@ -159,7 +161,12 @@ export const createAggregateLoader: CreateAggregateLoaderFunction = ({ ); } const folded = foldState({ aggregate, events: loaded.events, from }); - const persons = personsOf({ aggregate, events: loaded.events, from: snapshot.persons }); + const persons = personsOf({ + aggregate, + events: loaded.events, + fingerprints: loaded.fingerprints, + from: snapshot.persons, + }); if (!due(aggregate, settings.verify)) { count(aggregate, "used"); return { folded, version: loaded.version, persons }; diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index d2977e43..f3b20346 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -4,6 +4,7 @@ import type { DispatchOutcome, PutIdempotencyRecordArgs, } from "../../adapter/storage/idempotency-store.ts"; +import type { KeyedPerson } from "../../adapter/storage/personal-key-store.ts"; import type { Scheduler } from "../../adapter/storage/scheduler.ts"; import type { ResolvedConfig } from "../../config/types.ts"; import type { Clock } from "../../contracts/clock.ts"; @@ -322,7 +323,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ // Only streams written before `begin` existed land here, so the set stays small. const warnedOpenings = new Set(); // Whose personal data a rejection's state held, which its message may quote. - const rejectedPersons = new WeakMap(); + const rejectedPersons = new WeakMap(); const execute = async ( aggregate: AggregateRuntime, runtime: CommandRuntime, @@ -494,7 +495,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ const recordOf = ( { key, fingerprint }: Idempotency, outcome: DispatchOutcome, - persons: readonly string[] = [], + persons: readonly KeyedPerson[] = [], ): PutIdempotencyRecordArgs => { const now = clock.now(); return { @@ -509,7 +510,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ }; }; - const personsOf = (answer: DispatchResult | DomainError): readonly string[] => + const personsOf = (answer: DispatchResult | DomainError): readonly KeyedPerson[] => answer instanceof DomainError ? (rejectedPersons.get(answer) ?? []) : []; const outcomeOf = (answer: DispatchResult | DomainError): DispatchOutcome => diff --git a/packages/core/src/kernel/personal-data/crypto.test.ts b/packages/core/src/kernel/personal-data/crypto.test.ts index 169386aa..0709843c 100644 --- a/packages/core/src/kernel/personal-data/crypto.test.ts +++ b/packages/core/src/kernel/personal-data/crypto.test.ts @@ -8,7 +8,7 @@ describe("personal data crypto", () => { const sealed = await seal({ key, person: "c-1", value }); expect(isSealed(sealed)).toBe(true); expect(sealed.$personal.person).toBe("c-1"); - expect(sealed.$personal.key).toBe(key.id); + expect(sealed.$personal.key).toBe(key.fingerprint); expect(sealed.$personal.data).not.toContain(JSON.stringify(value)); expect(await unseal({ key, sealed })).toEqual(value); } @@ -16,9 +16,11 @@ describe("personal data crypto", () => { it("gives every key and every sealing a different text", async () => { const keys = await Promise.all([generatePersonalKey(), generatePersonalKey()]); - expect(keys[0]).not.toBe(keys[1]); - expect(keys[0]).toHaveLength(44); - const key = await importPersonalKey(keys[0] ?? ""); + expect(keys[0]?.key).not.toBe(keys[1]?.key); + expect(keys[0]?.fingerprint).not.toBe(keys[1]?.fingerprint); + expect(keys[0]?.key).toHaveLength(44); + expect(keys[0]?.fingerprint).toHaveLength(44); + const key = await importPersonalKey(keys[0] ?? { key: "", fingerprint: "" }); const [first, second] = await Promise.all([ seal({ key, person: "c-1", value: "same" }), seal({ key, person: "c-1", value: "same" }), @@ -27,14 +29,19 @@ describe("personal data crypto", () => { expect(first.$personal.data).not.toBe(second.$personal.data); }); - it("imports a key that cannot be exported again, with a fingerprint of its own", async () => { + it("fingerprints a key with the SHA-256 of its raw bytes", async () => { + const stored = await generatePersonalKey(); + const raw = Uint8Array.from(atob(stored.key), (character) => character.charCodeAt(0)); + const digest = new Uint8Array(await crypto.subtle.digest("SHA-256", raw)); + expect(stored.fingerprint).toBe(btoa(String.fromCharCode(...digest))); + }); + + it("imports a key that cannot be exported again, named by its stored fingerprint", async () => { const stored = await generatePersonalKey(); const key = await importPersonalKey(stored); expect(key.cryptoKey.extractable).toBe(false); expect(key.cryptoKey.algorithm).toEqual({ name: "AES-GCM", length: 256 }); - expect((await importPersonalKey(stored)).id).toBe(key.id); - expect((await importPersonalKey(await generatePersonalKey())).id).not.toBe(key.id); - expect(key.id).not.toBe(stored); + expect(key.fingerprint).toBe(stored.fingerprint); }); it("opens nothing without a key or with another one, and refuses a value its key should open", async () => { diff --git a/packages/core/src/kernel/personal-data/crypto.ts b/packages/core/src/kernel/personal-data/crypto.ts index 2d3c87c3..54207a30 100644 --- a/packages/core/src/kernel/personal-data/crypto.ts +++ b/packages/core/src/kernel/personal-data/crypto.ts @@ -1,3 +1,5 @@ +import type { StoredPersonalKey } from "../../adapter/storage/personal-key-store.ts"; + const ALGORITHM = "AES-GCM"; const KEY_BITS = 256; const IV_BYTES = 12; @@ -18,7 +20,7 @@ export interface PersonalEnvelope { * A person's key as the store uses it, with the fingerprint envelopes name it by. */ export interface PersonalKey { - readonly id: string; + readonly fingerprint: string; readonly cryptoKey: CryptoKey; } @@ -60,32 +62,33 @@ const fromBase64 = (text: string): Uint8Array => Uint8Array.from(atob(text), (character) => character.charCodeAt(0)); export interface GeneratePersonalKeyFunction { - (): Promise; + (): Promise; } /** - * A fresh key, as the text the key store keeps. + * A fresh key, as the key store keeps it: the raw key and its SHA-256, both in base64. */ export const generatePersonalKey: GeneratePersonalKeyFunction = async () => { const key = await crypto.subtle.generateKey({ name: ALGORITHM, length: KEY_BITS }, true, [ "encrypt", "decrypt", ]); - return toBase64(new Uint8Array(await crypto.subtle.exportKey("raw", key))); + const raw = new Uint8Array(await crypto.subtle.exportKey("raw", key)); + const digest = await crypto.subtle.digest("SHA-256", raw); + return { key: toBase64(raw), fingerprint: toBase64(new Uint8Array(digest)) }; }; export interface ImportPersonalKeyFunction { - (key: string): Promise; + (stored: StoredPersonalKey): Promise; } -export const importPersonalKey: ImportPersonalKeyFunction = async (key) => { - const raw = fromBase64(key); - const [digest, cryptoKey] = await Promise.all([ - crypto.subtle.digest("SHA-256", raw), - crypto.subtle.importKey("raw", raw, ALGORITHM, false, ["encrypt", "decrypt"]), - ]); - return { id: toBase64(new Uint8Array(digest)), cryptoKey }; -}; +export const importPersonalKey: ImportPersonalKeyFunction = async ({ key, fingerprint }) => ({ + fingerprint, + cryptoKey: await crypto.subtle.importKey("raw", fromBase64(key), ALGORITHM, false, [ + "encrypt", + "decrypt", + ]), +}); export interface SealArgs { readonly key: PersonalKey; @@ -110,7 +113,7 @@ export const seal: SealFunction = async ({ key, person, value }) => { return { $personal: { person, - key: key.id, + key: key.fingerprint, iv: toBase64(iv), data: toBase64(new Uint8Array(encrypted)), }, @@ -131,7 +134,7 @@ export interface UnsealFunction { * and may have been given a new key since. A value the key should open and does not throws. */ export const unseal: UnsealFunction = async ({ key, sealed: { $personal: envelope } }) => { - if (key === null || key.id !== envelope.key) return null; + if (key === null || key.fingerprint !== envelope.key) return null; const decrypted = await crypto.subtle.decrypt( { name: ALGORITHM, iv: fromBase64(envelope.iv) }, key.cryptoKey, diff --git a/packages/core/src/kernel/personal-data/event-store.test.ts b/packages/core/src/kernel/personal-data/event-store.test.ts index 57fb8dcc..e5f2553e 100644 --- a/packages/core/src/kernel/personal-data/event-store.test.ts +++ b/packages/core/src/kernel/personal-data/event-store.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import type { PersonalKeyStore } from "../../adapter/storage/personal-key-store.ts"; +import type { KeyedPerson, PersonalKeyStore } from "../../adapter/storage/personal-key-store.ts"; import { pendingEvent } from "../../adapter/testing/fixtures.ts"; import { ConcurrencyError, ValidationError } from "../../contracts/errors.ts"; import { silentLogger } from "../../contracts/logger.ts"; @@ -7,7 +7,7 @@ import { memory } from "../../memory/index.ts"; import type { EventPayloadArgs } from "../../modules/payload.ts"; import type { Registry } from "../../modules/registry.ts"; import { buildAggregates } from "../aggregate/build-aggregates.ts"; -import { isSealed } from "./crypto.ts"; +import { generatePersonalKey, isSealed } from "./crypto.ts"; import { withPersonalData } from "./event-store.ts"; const registry: Registry = { @@ -94,7 +94,7 @@ describe("withPersonalData", () => { expect(isSealed(payload.email)).toBe(true); expect(JSON.stringify(stored)).not.toContain("ana@example.com"); expect((payload.email as { $personal: { person: string } }).$personal.person).toBe("c-1"); - expect(await storage.personalKeys.get("c-1")).toBeTypeOf("string"); + expect((await storage.personalKeys.get("c-1"))?.fingerprint).toHaveLength(44); const loaded = await store.load({ aggregateType: "customer", aggregateId: "c-1" }); expect(loaded.events[0]?.payload).toEqual({ email: "ana@example.com", plan: "pro" }); expect((await store.readAll({ afterPosition: 0, limit: 10 }))[0]?.payload).toEqual({ @@ -117,7 +117,7 @@ describe("withPersonalData", () => { expect(payload.customerId).toBe("c-7"); expect(isSealed(payload.address)).toBe(true); expect((payload.address as { $personal: { person: string } }).$personal.person).toBe("c-7"); - expect(await storage.personalKeys.get("c-7")).toBeTypeOf("string"); + expect((await storage.personalKeys.get("c-7"))?.fingerprint).toHaveLength(44); expect(await storage.personalKeys.get("o-1")).toBeNull(); const loaded = await store.load({ aggregateType: "order", aggregateId: "o-1" }); expect(loaded.events[0]?.payload).toEqual({ @@ -149,6 +149,8 @@ describe("withPersonalData", () => { { email: null, plan: "pro" }, { customerId: "c-1", note: null, address: null, total: 1 }, ]); + const loaded = await store.load({ aggregateType: "customer", aggregateId: "c-1" }); + expect(loaded.fingerprints).toEqual(new Map()); }); it("reads what was sealed under a key the person no longer has as null, next to what their new key opens", async () => { @@ -171,9 +173,11 @@ describe("withPersonalData", () => { null, "ana@again.example", ]); + const again = await storage.personalKeys.get("c-1"); + expect(loaded.fingerprints).toEqual(new Map([["c-1", again?.fingerprint]])); }); - it("refuses to give a new key to a person the events were decided on, forgotten since", async () => { + it("seals a person the events were decided on only under the key they were read under", async () => { const { storage, store, calls, registered } = await setup(); await store.append({ aggregateType: "customer", @@ -181,39 +185,59 @@ describe("withPersonalData", () => { expectedVersion: 0, events: [registered(1, "ana@example.com")], }); - await storage.personalKeys.remove("c-1"); - calls.length = 0; - const refused = await store - .append({ + const { fingerprints } = await store.load({ aggregateType: "customer", aggregateId: "c-1" }); + const read = [ + { person: "c-2", fingerprint: "another key" }, + { person: "c-1", fingerprint: fingerprints?.get("c-1") ?? "" }, + ]; + const append = (version: number, persons: readonly KeyedPerson[]) => + store.append({ aggregateType: "customer", aggregateId: "c-1", - expectedVersion: 1, - events: [registered(2, "ana@example.com")], - persons: ["c-2", "c-1"], - }) - .catch((error: unknown) => error); + expectedVersion: version - 1, + events: [registered(version, "ana@example.com")], + persons, + }); + await append(2, read); + await storage.personalKeys.remove("c-1"); + calls.length = 0; + const refused = await append(3, read).catch((error: unknown) => error); expect(refused).toBeInstanceOf(ConcurrencyError); expect(refused).toMatchObject({ code: "CONCURRENCY_CONFLICT", streamId: "customer:c-1", - expectedVersion: 1, - actualVersion: 1, - message: "Person c-1 was forgotten since customer:c-1 was loaded at version 1", + expectedVersion: 2, + actualVersion: 2, + message: "Person c-1 was forgotten since customer:c-1 was loaded at version 2", }); expect(calls).toEqual(["get:c-1"]); expect(await storage.personalKeys.get("c-1")).toBeNull(); + + const fresh = await generatePersonalKey(); + await storage.personalKeys.add("c-1", fresh); + await expect(append(3, read)).rejects.toBeInstanceOf(ConcurrencyError); + await expect( + append(3, [...read, { person: "c-1", fingerprint: fresh.fingerprint }]), + ).rejects.toBeInstanceOf(ConcurrencyError); + expect(await storage.personalKeys.get("c-1")).toEqual(fresh); expect( (await storage.eventStore.load({ aggregateType: "customer", aggregateId: "c-1" })).version, - ).toBe(1); + ).toBe(2); - await store.append({ - aggregateType: "customer", - aggregateId: "c-1", - expectedVersion: 1, - events: [registered(2, "ana@again.example")], - persons: ["c-2"], - }); - expect(await storage.personalKeys.get("c-1")).not.toBeNull(); + await append(3, [{ person: "c-1", fingerprint: fresh.fingerprint }, ...read.slice(0, 1)]); + await append(4, [{ person: "c-1", fingerprint: null }]); + await storage.personalKeys.remove("c-1"); + await append(5, [{ person: "c-1", fingerprint: null }]); + const created = await storage.personalKeys.get("c-1"); + expect(created?.fingerprint).not.toBe(fresh.fingerprint); + const loaded = await store.load({ aggregateType: "customer", aggregateId: "c-1" }); + expect(loaded.events.map((event) => (event.payload as { email: unknown }).email)).toEqual([ + null, + null, + null, + null, + "ana@example.com", + ]); }); it("reuses a person's key across events and appends, and fetches it once per call", async () => { @@ -251,7 +275,8 @@ describe("withPersonalData", () => { it("keeps the key another writer added first", async () => { const { storage, store, registered } = await setup(); - await storage.personalKeys.add("c-1", "Zmlyc3Q="); + const first = { key: "Zmlyc3Q=", fingerprint: "first" }; + await storage.personalKeys.add("c-1", first); const concurrentKeys: PersonalKeyStore = { get: async () => null, add: (person, key) => storage.personalKeys.add(person, key), @@ -269,7 +294,7 @@ describe("withPersonalData", () => { events: [registered(1, "ana@example.com")], }), ).rejects.toThrow(); - expect(await storage.personalKeys.get("c-1")).toBe("Zmlyc3Q="); + expect(await storage.personalKeys.get("c-1")).toEqual(first); void store; }); @@ -347,6 +372,6 @@ describe("withPersonalData", () => { expectedVersion: 0, events: [placed({ customerId: 42, address: { street: "x" }, total: 1 })], }); - expect(await storage.personalKeys.get("42")).toBeTypeOf("string"); + expect((await storage.personalKeys.get("42"))?.fingerprint).toHaveLength(44); }); }); diff --git a/packages/core/src/kernel/personal-data/event-store.ts b/packages/core/src/kernel/personal-data/event-store.ts index 62afa7df..da3f122b 100644 --- a/packages/core/src/kernel/personal-data/event-store.ts +++ b/packages/core/src/kernel/personal-data/event-store.ts @@ -38,6 +38,7 @@ const personOf = (event: PendingEvent, payload: Record, field: // A conflict like a moved stream: the events were decided on a state that held the person's data, // and the person was forgotten since, so the command is decided again on the state without it. +// Forgotten even when they hold a key again: one given to them after `forget` is another key. class ForgottenSinceLoadError extends ConcurrencyError { constructor(args: AppendArgs, person: string) { const { expectedVersion } = args; @@ -71,13 +72,17 @@ export const withPersonalData: WithPersonalDataFunction = ({ eventStore, aggrega const marksOf = (event: Pick): PersonalMarks | null => aggregates.byName[event.aggregateType]?.eventsByType[event.type]?.personal ?? null; - // Only a person the state did not hold gets a new key: one it held and who has none now was - // forgotten after the load, and a new key would make their old data readable again. + // A person the state held under a key is sealed under that key alone: without it now, they were + // forgotten after the load, and any other key would make their old data readable again. const keyInForce = async (args: AppendArgs, person: string): Promise => { const stored = await keys.get(person); - if (stored !== null) return importPersonalKey(stored); - if (args.persons?.includes(person)) throw new ForgottenSinceLoadError(args, person); - return importPersonalKey(await keys.add(person, await generatePersonalKey())); + const readUnder = (args.persons ?? []).filter( + (held) => held.person === person && held.fingerprint !== null, + ); + if (readUnder.some((held) => held.fingerprint !== stored?.fingerprint)) { + throw new ForgottenSinceLoadError(args, person); + } + return importPersonalKey(stored ?? (await keys.add(person, await generatePersonalKey()))); }; const sealed = async ( @@ -100,13 +105,17 @@ export const withPersonalData: WithPersonalDataFunction = ({ eventStore, aggrega const unsealed = async ( event: StoredEvent, key: (person: string) => Promise, + opened: Map, ): Promise => { if (!isRecord(event.payload)) return event; const entries = Object.entries(event.payload).filter(([, value]) => isSealed(value)); if (entries.length === 0) return event; const payload: Record = { ...event.payload }; for (const [field, value] of entries as [string, SealedValue][]) { - payload[field] = await unseal({ key: await key(value.$personal.person), sealed: value }); + const { person, key: sealedUnder } = value.$personal; + const personal = await key(person); + payload[field] = await unseal({ key: personal, sealed: value }); + if (personal?.fingerprint === sealedUnder) opened.set(person, sealedUnder); } return { ...event, payload }; }; @@ -138,11 +147,14 @@ export const withPersonalData: WithPersonalDataFunction = ({ eventStore, aggrega load: async (args) => { const loaded = await eventStore.load(args); const key = opening(); - return { ...loaded, events: await Promise.all(loaded.events.map((e) => unsealed(e, key))) }; + const fingerprints = new Map(); + const events = await Promise.all(loaded.events.map((e) => unsealed(e, key, fingerprints))); + return { ...loaded, events, fingerprints }; }, readAll: async (args) => { const key = opening(); - return Promise.all((await eventStore.readAll(args)).map((e) => unsealed(e, key))); + const ignored = new Map(); + return Promise.all((await eventStore.readAll(args)).map((e) => unsealed(e, key, ignored))); }, lastPosition: () => eventStore.lastPosition(), }; diff --git a/packages/core/src/kernel/personal-data/forget.test.ts b/packages/core/src/kernel/personal-data/forget.test.ts index 4fd072c8..adb4b521 100644 --- a/packages/core/src/kernel/personal-data/forget.test.ts +++ b/packages/core/src/kernel/personal-data/forget.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; import type { Storage } from "../../adapter/adapter.ts"; +import { 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"; @@ -26,7 +27,7 @@ const setup = async (storage?: Storage) => { describe("forgetPersonalData", () => { it("removes the person's key and appends PersonalDataForgotten to their stream, answering like a command", async () => { const { storage, forget, logs } = await setup(); - await storage.personalKeys.add("c-1", "a2V5"); + await storage.personalKeys.add("c-1", testPersonalKey("1")); expect(await forget("c-1")).toEqual({ scheduled: false, aggregateType: "bounda", diff --git a/packages/core/src/kernel/personal-data/personal-data.test.ts b/packages/core/src/kernel/personal-data/personal-data.test.ts index a62555da..4f8c9988 100644 --- a/packages/core/src/kernel/personal-data/personal-data.test.ts +++ b/packages/core/src/kernel/personal-data/personal-data.test.ts @@ -105,6 +105,23 @@ const registry = { }, }, policies: { + registerOnInvited: { + module: { + on: "Invited", + handler: async ({ + event, + commands, + }: { + event: { payload: { customerId: string; email: string } }; + commands: Record Promise>; + }) => { + const { customerId, email } = event.payload; + await commands.registerCustomer?.({ customerId, email, name: "Invited" }); + await commands.remindCustomer?.({ customerId }); + }, + }, + source: "newsletter", + }, boundaEraseOnPersonalDataForgotten: { module: { handler: async ({ event }: { event: PersonalDataForgottenEvent }) => { @@ -116,6 +133,50 @@ const registry = { }, processes: {}, }, + newsletter: { + events: { + subscribed: { + payload: ({ z, personal, person }: EventPayloadArgs) => + z.object({ customerId: person(z.string()), email: personal(z.email()) }), + evolve: () => ({}), + }, + invited: { + payload: ({ z }: EventPayloadArgs) => + z.object({ customerId: z.string(), email: z.email() }), + evolve: () => ({}), + }, + }, + commands: { + subscribe: { + module: { + payload: ({ z }: PayloadArgs) => + z.object({ newsletterId: z.string(), customerId: z.string(), email: z.email() }), + handler: ({ + command, + events, + }: { + command: { payload: { customerId: string; email: string } }; + events: Record unknown>; + }) => [events.subscribed?.(command.payload)], + }, + }, + invite: { + module: { + payload: ({ z }: PayloadArgs) => + z.object({ newsletterId: z.string(), customerId: z.string(), email: z.email() }), + handler: ({ + command, + events, + }: { + command: { payload: { customerId: string; email: string } }; + events: Record unknown>; + }) => [events.invited?.(command.payload)], + }, + }, + }, + policies: {}, + processes: {}, + }, }, readModels: { customerDirectory: { @@ -172,6 +233,17 @@ const registry = { const ana = { customerId: "c-1", email: "ana@example.com", name: "Ana" }; +const snapshotted = { + registry: { + ...registry, + aggregates: { + ...registry.aggregates, + customer: { ...registry.aggregates.customer, stateFingerprint: "fp" }, + }, + }, + config: { runtime: { overrides: { customer: { snapshot: { every: 1 } } } } }, +}; + describe("personal data in an app", () => { it("stores personal fields sealed, projects them in the clear, and forgets a person end to end", async () => { erased.length = 0; @@ -266,20 +338,15 @@ describe("personal data in an app", () => { it("forgets a person's snapshots too, so their aggregate loads with the fields forgotten", async () => { const adapter = memory(); const storage = await adapter.createStorage({ logger: silentLogger }); - const { app } = await createTestApp({ - registry: { - ...registry, - aggregates: { - customer: { ...registry.aggregates.customer, stateFingerprint: "fp" }, - }, - }, - config: { runtime: { overrides: { customer: { snapshot: { every: 1 } } } } }, - adapter, - }); + const { app } = await createTestApp({ ...snapshotted, adapter }); await app.commands.registerCustomer(ana); await app.runUntilIdle(); const stream = { aggregateType: "customer", aggregateId: "c-1" }; - expect(await storage.snapshotStore.get(stream)).toMatchObject({ version: 1, persons: ["c-1"] }); + const key = await storage.personalKeys.get("c-1"); + expect(await storage.snapshotStore.get(stream)).toMatchObject({ + version: 1, + persons: [{ person: "c-1", fingerprint: key?.fingerprint }], + }); expect((await storage.snapshotStore.get(stream))?.state).toContain("ana@example.com"); await app.forget("c-1"); @@ -307,10 +374,11 @@ describe("personal data in an app", () => { rejected: "NotYet", message: "ana@example.com cannot be confirmed yet", }); + const key = await storage.personalKeys.get("c-1"); expect( await storage.idempotencyStore.get({ key: "confirm-1", now: clock.now() }), ).toMatchObject({ - persons: ["c-1"], + persons: [{ person: "c-1", fingerprint: key?.fingerprint }], }); expect(await confirm()).toMatchObject({ message: "ana@example.com cannot be confirmed yet" }); @@ -343,6 +411,53 @@ describe("personal data in an app", () => { await app.stop(); }); + it.each([ + ["its stream", {}], + ["a snapshot", snapshotted], + ])( + "decides a command again when its person is forgotten and keyed anew between its load from %s and its append", + async (_, setup) => { + const adapter = memory(); + const storage = await adapter.createStorage({ logger: silentLogger }); + const { app } = await createTestApp({ registry, ...setup, adapter }); + await app.commands.registerCustomer(ana); + await app.runUntilIdle(); + const before = await storage.personalKeys.get("c-1"); + whileDeciding = async () => { + await app.forget("c-1"); + await app.commands.subscribe({ + newsletterId: "n-1", + customerId: "c-1", + email: "ana@again.example", + }); + }; + const reminded = await app.commands.remindCustomer({ customerId: "c-1" }); + expect(whileDeciding).toBeNull(); + const [reminder] = await app.events.read({ after: reminded.position - 1 }); + expect(reminder?.type).toBe("CustomerReminded"); + expect(reminder?.payload).toEqual({}); + const after = await storage.personalKeys.get("c-1"); + expect(after).not.toBeNull(); + expect(after?.fingerprint).not.toBe(before?.fingerprint); + await app.stop(); + }, + ); + + it("seals what a unit of work staged for a new person, with what it decided on it", async () => { + const { app } = await createTestApp({ registry }); + await app.commands.invite({ newsletterId: "n-1", customerId: "c-3", email: "cy@example.com" }); + expect((await app.runUntilIdle()).idle).toBe(true); + const events = await app.events.read(); + expect(events.map((event) => event.type)).toEqual([ + "Invited", + "CustomerRegistered", + "CustomerReminded", + ]); + expect(isSealed((events[2]?.payload as { to?: unknown } | undefined)?.to)).toBe(true); + expect(await app.deadLetters.list()).toEqual([]); + await app.stop(); + }); + it("refuses to forget nobody", async () => { const { app } = await createTestApp({ registry }); await expect(app.forget("")).rejects.toBeInstanceOf(ValidationError); diff --git a/packages/core/src/kernel/personal-data/persons.test.ts b/packages/core/src/kernel/personal-data/persons.test.ts index 01c923e1..246ba0b2 100644 --- a/packages/core/src/kernel/personal-data/persons.test.ts +++ b/packages/core/src/kernel/personal-data/persons.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { pendingEvent } from "../../adapter/testing/fixtures.ts"; +import { testKeyedPerson as held, pendingEvent } from "../../adapter/testing/fixtures.ts"; import type { StoredEvent } from "../../contracts/event.ts"; import type { EventPayloadArgs } from "../../modules/payload.ts"; import type { Registry } from "../../modules/registry.ts"; @@ -42,12 +42,16 @@ const event = (version: number, type: string, payload: unknown): StoredEvent => }); describe("personsOf", () => { - it("names whoever an event's readable personal fields belong to, once each, after those known", () => { + it("names whoever an event's readable personal fields belong to, once per key, after those known", () => { if (aggregate === undefined) throw new Error("no aggregate"); expect( personsOf({ aggregate, - from: ["c-0"], + from: [held("c-0", "0"), held("c-1", "old"), held("c-1", "1")], + fingerprints: new Map([ + ["c-1", "f-1"], + ["c-9", "f-9"], + ]), events: [ event(1, "OrderPlaced", { customerId: "c-1", note: "ring" }), event(2, "OrderSigned", { signature: "x" }), @@ -56,7 +60,13 @@ describe("personsOf", () => { event(5, "OrderPaid", {}), ], }), - ).toEqual(["c-0", "c-1", "o-1", "7"]); + ).toEqual([ + held("c-0", "0"), + held("c-1", "old"), + held("c-1", "1"), + held("o-1", null), + held("7", null), + ]); }); it("names no one for an event whose personal fields are forgotten or absent", () => { diff --git a/packages/core/src/kernel/personal-data/persons.ts b/packages/core/src/kernel/personal-data/persons.ts index 898ebf6c..adc8e33b 100644 --- a/packages/core/src/kernel/personal-data/persons.ts +++ b/packages/core/src/kernel/personal-data/persons.ts @@ -1,3 +1,4 @@ +import { type KeyedPerson, keyedPersonId } from "../../adapter/storage/personal-key-store.ts"; import type { StoredEvent } from "../../contracts/event.ts"; import type { AggregateRuntime } from "../aggregate/runtime.ts"; import { isRecord } from "../shared/merge-fields.ts"; @@ -5,22 +6,33 @@ import { isRecord } from "../shared/merge-fields.ts"; export interface PersonsOfArgs { readonly aggregate: AggregateRuntime; readonly events: readonly StoredEvent[]; + // The key each person's sealed fields were opened with, as the load that read them reports it. + readonly fingerprints?: ReadonlyMap | undefined; // Persons already known, such as a snapshot's, kept first. - readonly from?: readonly string[]; + readonly from?: readonly KeyedPerson[]; } export interface PersonsOfFunction { - (args: PersonsOfArgs): readonly string[]; + (args: PersonsOfArgs): readonly KeyedPerson[]; } /** - * The people whose personal data the events carry, once each: whoever an event names, by its - * person field or its aggregate's id, when one of its marked fields still holds a value. A field - * forgotten reads as `null` and names no one, so a forgotten person's stream can be snapshotted - * again without them. + * The people whose personal data the events carry, once each per key: whoever an event names, by + * its person field or its aggregate's id, when one of its marked fields still holds a value, with + * the key it was opened with. A field forgotten reads as `null` and names no one, so a forgotten + * person's stream can be snapshotted again without them. */ -export const personsOf: PersonsOfFunction = ({ aggregate, events, from = [] }) => { - const persons = new Set(from); +export const personsOf: PersonsOfFunction = ({ + aggregate, + events, + fingerprints = new Map(), + from = [], +}) => { + const persons = new Map(); + const add = (held: KeyedPerson): void => { + persons.set(keyedPersonId(held), held); + }; + for (const held of from) add(held); for (const event of events) { const marks = aggregate.eventsByType[event.type]?.personal; if (marks === undefined || marks === null || !isRecord(event.payload)) continue; @@ -29,7 +41,9 @@ export const personsOf: PersonsOfFunction = ({ aggregate, events, from = [] }) = continue; } const named = marks.person === null ? event.aggregateId : payload[marks.person]; - if (typeof named === "string" || typeof named === "number") persons.add(String(named)); + if (typeof named !== "string" && typeof named !== "number") continue; + const person = String(named); + add({ person, fingerprint: fingerprints.get(person) ?? null }); } - return [...persons]; + return [...persons.values()]; }; diff --git a/packages/core/src/memory/idempotency-store.ts b/packages/core/src/memory/idempotency-store.ts index 18cca8bf..47c415a8 100644 --- a/packages/core/src/memory/idempotency-store.ts +++ b/packages/core/src/memory/idempotency-store.ts @@ -1,4 +1,5 @@ import type { IdempotencyRecord, IdempotencyStore } from "../adapter/storage/idempotency-store.ts"; +import type { KeyedPerson } from "../adapter/storage/personal-key-store.ts"; import { IdempotencyKeyTakenError } from "../contracts/errors.ts"; import { createStoreEntries, type WithEntries } from "./entries.ts"; import { jsonCopy } from "./json-copy.ts"; @@ -9,9 +10,10 @@ export interface CreateMemoryIdempotencyStoreFunction { export interface CreateKeptIdempotencyStoreArgs { /** - * Whether a person has a personal key; synchronous, so `put` checks and writes in one step. + * Whether a person holds the key a record names; synchronous, so `put` checks and writes in one + * step. */ - readonly holds: (person: string) => boolean; + readonly holds: (held: KeyedPerson) => boolean; } export interface CreateKeptIdempotencyStoreFunction { @@ -52,7 +54,7 @@ export const createKeptIdempotencyStore: CreateKeptIdempotencyStoreFunction = ({ }, forget: async (person) => { for (const [key, record] of records) { - if (record.persons.includes(person)) records.delete(key); + if (record.persons.some((held) => held.person === person)) records.delete(key); } }, }; diff --git a/packages/core/src/memory/personal-key-store.ts b/packages/core/src/memory/personal-key-store.ts index b73a91f3..f7905ecc 100644 --- a/packages/core/src/memory/personal-key-store.ts +++ b/packages/core/src/memory/personal-key-store.ts @@ -1,12 +1,17 @@ -import type { PersonalKeyStore } from "../adapter/storage/personal-key-store.ts"; +import type { + KeyedPerson, + PersonalKeyStore, + StoredPersonalKey, +} from "../adapter/storage/personal-key-store.ts"; import { createStoreEntries, type WithEntries } from "./entries.ts"; export interface KeptPersonalKeyStore extends WithEntries { /** - * Whether `person` has a key, answered at once: the memory snapshot store checks it in the - * same step as it writes, so no forget runs between the two. + * Whether the person holds the key of that fingerprint, or any key for a `null` one, answered + * at once: the memory snapshot store checks it in the same step as it writes, so no forget runs + * between the two. */ - holds(person: string): boolean; + holds(held: KeyedPerson): boolean; } export interface CreateKeptPersonalKeyStoreFunction { @@ -14,9 +19,12 @@ export interface CreateKeptPersonalKeyStoreFunction { } export const createKeptPersonalKeyStore: CreateKeptPersonalKeyStoreFunction = () => { - const keys = new Map(); + const keys = new Map(); return { - holds: (person) => keys.has(person), + holds: ({ person, fingerprint }) => { + const stored = keys.get(person); + return stored !== undefined && (fingerprint === null || stored.fingerprint === fingerprint); + }, store: { get: async (person) => keys.get(person) ?? null, add: async (person, key) => { diff --git a/packages/core/src/memory/snapshot-store.ts b/packages/core/src/memory/snapshot-store.ts index 8ff9881f..a489f5d6 100644 --- a/packages/core/src/memory/snapshot-store.ts +++ b/packages/core/src/memory/snapshot-store.ts @@ -1,11 +1,13 @@ +import type { KeyedPerson } from "../adapter/storage/personal-key-store.ts"; import type { Snapshot, SnapshotStore } from "../adapter/storage/snapshot-store.ts"; import { streamId } from "../contracts/event.ts"; export interface CreateMemorySnapshotStoreArgs { /** - * Whether a person has a personal key; synchronous, so `put` checks and writes in one step. + * Whether a person holds the key a snapshot names; synchronous, so `put` checks and writes in + * one step. */ - readonly holds: (person: string) => boolean; + readonly holds: (held: KeyedPerson) => boolean; } export interface CreateMemorySnapshotStoreFunction { @@ -24,7 +26,7 @@ export const createMemorySnapshotStore: CreateMemorySnapshotStoreFunction = ({ h }, forget: async (person) => { for (const [key, snapshot] of snapshots) { - if (snapshot.persons.includes(person)) snapshots.delete(key); + if (snapshot.persons.some((held) => held.person === person)) snapshots.delete(key); } }, }; diff --git a/packages/postgresql/src/held-keys.ts b/packages/postgresql/src/held-keys.ts new file mode 100644 index 00000000..9c37b0fc --- /dev/null +++ b/packages/postgresql/src/held-keys.ts @@ -0,0 +1,28 @@ +export interface HeldKeysArgs { + readonly personalKeys: string; + // The placeholder of the `jsonb` array of persons. + readonly persons: string; +} + +export interface HeldKeysFunction { + (args: HeldKeysArgs): string; +} + +/** + * A condition true when every person in `persons` holds, in `personalKeys`, the key of the + * fingerprint named with them, or any key for a `null` one. A person named under two keys never + * holds both. The keys it finds stay locked until the transaction ends: a forget that removes one + * waits for the write, and then finds the row to remove. + */ +export const heldKeys: HeldKeysFunction = ({ personalKeys, persons }) => + `(SELECT count(*) FROM (SELECT 1 FROM ${personalKeys} AS "k" JOIN jsonb_to_recordset(${persons}::jsonb) AS "p"("person" text, "fingerprint" text) ON "k"."person" = "p"."person" AND "k"."fingerprint" = coalesce("p"."fingerprint", "k"."fingerprint") FOR SHARE OF "k") AS "held") = jsonb_array_length(${persons}::jsonb)`; + +export interface NamingFunction { + (person: string): string; +} + +/** + * A condition true for a row whose `persons` names the person of the placeholder `person`. + */ +export const naming: NamingFunction = (person) => + `"persons" @> jsonb_build_array(jsonb_build_object('person', ${person}::text))`; diff --git a/packages/postgresql/src/idempotency-store.ts b/packages/postgresql/src/idempotency-store.ts index 84362277..2cb5467d 100644 --- a/packages/postgresql/src/idempotency-store.ts +++ b/packages/postgresql/src/idempotency-store.ts @@ -3,9 +3,11 @@ import type { DispatchOutcome, IdempotencyRecord, IdempotencyStore, + KeyedPerson, } from "@bounda-dev/core/adapter"; import { postgresqlDialect } from "@bounda-dev/core/adapter/sql"; import type { PostgresqlDatabase } from "./database.ts"; +import { heldKeys, naming } from "./held-keys.ts"; export interface CreatePostgresqlIdempotencyStoreArgs { readonly db: PostgresqlDatabase; @@ -23,14 +25,15 @@ const toRecord = (row: Record): IdempotencyRecord => ({ fingerprint: String(row.fingerprint), outcome: row.outcome as DispatchOutcome, expiresAt: String(row.expires_at), - persons: (row.persons as readonly string[]).map(String), + persons: row.persons as KeyedPerson[], }); /** * `put` is a single upsert: PostgreSQL locks the conflicting row, so of two dispatches racing for * one key exactly one gets its row back. It locks the keys of the persons the record names too, * as a snapshot's `put` does, so a forget beside it waits and then removes it, or removed a key - * first and leaves it nothing to record; no row back with no live record under the key is that. + * first and leaves it nothing to record, even once the person has another key; no row back with + * no live record under the key is that. */ export const createPostgresqlIdempotencyStore: CreatePostgresqlIdempotencyStoreFunction = ({ db, @@ -49,7 +52,7 @@ export const createPostgresqlIdempotencyStore: CreatePostgresqlIdempotencyStoreF put: async ({ key, fingerprint, outcome, expiresAt, now, persons }) => { const rows = await db.all( `INSERT INTO ${table} ("key", "fingerprint", "outcome", "expires_at", "persons") - SELECT $1, $2, $3, $4, $6::text[] WHERE (SELECT count(*) FROM (SELECT 1 FROM ${personalKeys} WHERE "person" = ANY($6::text[]) FOR SHARE) AS "held") = cardinality($6::text[]) + SELECT $1, $2, $3, $4, $6::jsonb WHERE ${heldKeys({ personalKeys, persons: "$6" })} ON CONFLICT ("key") DO UPDATE SET "fingerprint" = excluded."fingerprint", "outcome" = excluded."outcome", @@ -63,7 +66,7 @@ export const createPostgresqlIdempotencyStore: CreatePostgresqlIdempotencyStoreF postgresqlDialect.encode("json", outcome), expiresAt, now.toISOString(), - [...persons], + postgresqlDialect.encode("json", persons), ], ); if (rows.length === 0 && (await get({ key, now })) !== null) { @@ -76,6 +79,6 @@ export const createPostgresqlIdempotencyStore: CreatePostgresqlIdempotencyStoreF ]); return rows.length; }, - forget: (person) => db.run(`DELETE FROM ${table} WHERE $1 = ANY("persons")`, [person]), + forget: (person) => db.run(`DELETE FROM ${table} WHERE ${naming("$1")}`, [person]), }; }; diff --git a/packages/postgresql/src/personal-key-store.ts b/packages/postgresql/src/personal-key-store.ts index 72f4ba3d..42004b77 100644 --- a/packages/postgresql/src/personal-key-store.ts +++ b/packages/postgresql/src/personal-key-store.ts @@ -1,4 +1,4 @@ -import type { PersonalKeyStore } from "@bounda-dev/core/adapter"; +import type { PersonalKeyStore, StoredPersonalKey } from "@bounda-dev/core/adapter"; import type { SqlExecutor } from "@bounda-dev/core/adapter/sql"; export interface CreatePostgresqlPersonalKeyStoreArgs { @@ -14,17 +14,21 @@ export const createPostgresqlPersonalKeyStore: CreatePostgresqlPersonalKeyStoreF db, table, }) => { - const get = async (person: string): Promise => { - const [row] = await db.all(`SELECT "key" FROM ${table} WHERE "person" = $1`, [person]); - return row === undefined ? null : String(row.key); + const get = async (person: string): Promise => { + const [row] = await db.all(`SELECT "key", "fingerprint" FROM ${table} WHERE "person" = $1`, [ + person, + ]); + return row === undefined + ? null + : { key: String(row.key), fingerprint: String(row.fingerprint) }; }; return { get, // Whichever writer inserted first, every one answers with the key in force. add: async (person, key) => { await db.run( - `INSERT INTO ${table} ("person", "key") VALUES ($1, $2) ON CONFLICT ("person") DO NOTHING`, - [person, key], + `INSERT INTO ${table} ("person", "key", "fingerprint") VALUES ($1, $2, $3) ON CONFLICT ("person") DO NOTHING`, + [person, key.key, key.fingerprint], ); return (await get(person)) ?? key; }, diff --git a/packages/postgresql/src/schema.ts b/packages/postgresql/src/schema.ts index 67de6952..a780d624 100644 --- a/packages/postgresql/src/schema.ts +++ b/packages/postgresql/src/schema.ts @@ -142,12 +142,13 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "fingerprint" text NOT NULL, "outcome" jsonb NOT NULL, "expires_at" text NOT NULL, - "persons" text[] NOT NULL + "persons" jsonb NOT NULL )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "expires_at")} ON ${tables.idempotencyKeys} ("expires_at")`, `CREATE TABLE IF NOT EXISTS ${tables.personalKeys} ( "person" text PRIMARY KEY, - "key" text NOT NULL + "key" text NOT NULL, + "fingerprint" text NOT NULL )`, `CREATE TABLE IF NOT EXISTS ${tables.snapshots} ( "aggregate_type" text NOT NULL, @@ -155,7 +156,7 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "version" integer NOT NULL, "fingerprint" text NOT NULL, "state" text NOT NULL, - "persons" text[] NOT NULL, + "persons" jsonb NOT NULL, PRIMARY KEY ("aggregate_type", "aggregate_id") )`, ]; diff --git a/packages/postgresql/src/snapshot-store.ts b/packages/postgresql/src/snapshot-store.ts index 502bdf05..1dd3d44b 100644 --- a/packages/postgresql/src/snapshot-store.ts +++ b/packages/postgresql/src/snapshot-store.ts @@ -1,5 +1,6 @@ -import type { Snapshot, SnapshotStore } from "@bounda-dev/core/adapter"; -import type { SqlExecutor } from "@bounda-dev/core/adapter/sql"; +import type { KeyedPerson, Snapshot, SnapshotStore } from "@bounda-dev/core/adapter"; +import { postgresqlDialect, type SqlExecutor } from "@bounda-dev/core/adapter/sql"; +import { heldKeys, naming } from "./held-keys.ts"; export interface CreatePostgresqlSnapshotStoreArgs { readonly db: SqlExecutor; @@ -18,7 +19,7 @@ const toSnapshot = (row: Record): Snapshot => ({ version: Number(row.version), fingerprint: String(row.fingerprint), state: String(row.state), - persons: (row.persons as readonly string[]).map(String), + persons: row.persons as KeyedPerson[], }); export const createPostgresqlSnapshotStore: CreatePostgresqlSnapshotStoreFunction = ({ @@ -33,17 +34,22 @@ export const createPostgresqlSnapshotStore: CreatePostgresqlSnapshotStoreFunctio ); return row === undefined ? null : toSnapshot(row); }, - // The keys are locked until the write commits: a forget that removes one waits for it, and - // then finds the snapshot to remove; one that removed it first leaves the write nothing. put: ({ aggregateType, aggregateId, version, fingerprint, state, persons }) => db.run( - `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT $1, $2, $3, $4, $5, $6::text[] WHERE (SELECT count(*) FROM (SELECT 1 FROM ${personalKeys} WHERE "person" = ANY($6::text[]) FOR SHARE) AS "held") = cardinality($6::text[]) ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, - [aggregateType, aggregateId, version, fingerprint, state, [...persons]], + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT $1, $2, $3, $4, $5, $6::jsonb WHERE ${heldKeys({ personalKeys, persons: "$6" })} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, + [ + aggregateType, + aggregateId, + version, + fingerprint, + state, + postgresqlDialect.encode("json", persons), + ], ), remove: ({ aggregateType, aggregateId }) => db.run(`DELETE FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2`, [ aggregateType, aggregateId, ]), - forget: (person) => db.run(`DELETE FROM ${table} WHERE $1 = ANY("persons")`, [person]), + forget: (person) => db.run(`DELETE FROM ${table} WHERE ${naming("$1")}`, [person]), }); From bda41506acf71409ea662baad0eb7691e91d6271 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sat, 31 Oct 2026 10:58:26 +0100 Subject: [PATCH 2/2] chore: describe personal key identity in the personal data changeset Everything this branch changes is unreleased, so the minor changeset of the feature describes it instead of a changeset of its own. --- .changeset/personal-data.md | 2 +- .changeset/personal-key-identity.md | 6 ------ 2 files changed, 1 insertion(+), 7 deletions(-) delete mode 100644 .changeset/personal-key-identity.md diff --git a/.changeset/personal-data.md b/.changeset/personal-data.md index 84326115..9fb3f359 100644 --- a/.changeset/personal-data.md +++ b/.changeset/personal-data.md @@ -7,4 +7,4 @@ "@bounda-dev/react-router": minor --- -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 in a new `personal_keys` table, `Storage` and `StorageTransaction` gain `personalKeys`, and the storage contracts cover it. The Cloudflare Worker answers `POST /forget/`. 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 a new key; `AppendArgs` gains `persons` for this, which a store ignores. +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 Cloudflare Worker answers `POST /forget/`. 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. diff --git a/.changeset/personal-key-identity.md b/.changeset/personal-key-identity.md deleted file mode 100644 index 48e534a2..00000000 --- a/.changeset/personal-key-identity.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -"@bounda-dev/core": minor -"@bounda-dev/postgresql": minor ---- - -A command that loaded a person's data before `forget` no longer writes it back under the fresh key another command gave them meanwhile: the state it loaded remembers the key each person's data was read under, and sealing their fields under any other key fails with `ConcurrencyError`, so the command is decided again on the state without their data. Snapshots and remembered rejections keep the same: each names the key it read every person under, and is written only while that key is still theirs. Breaking for adapters: `PersonalKeyStore` keeps a `StoredPersonalKey` (`{ key, fingerprint }`) per person, and `personal_keys` gains a `fingerprint` column; the `persons` of `Snapshot`, `IdempotencyRecord` and `AppendArgs` are `KeyedPerson[]` (`{ person, fingerprint }`, `null` for data staged by the same unit of work and not sealed yet), stored as JSON in SQLite and as `jsonb` in PostgreSQL; `LoadResult` gains `fingerprints`, which the kernel fills and a store leaves out. A unit of work that registered a person and then decided another command on the same stream no longer fails its commit.