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/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]), });