fix(core): seal a person's fields only under the key the state read them under

.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:<personId>`, 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/<personId>`. 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:<personId>`, 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/<personId>`. 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.

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

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
 

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

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`).

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();

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

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,

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') = ?)`;

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<string, unknown>): 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]),
   };
 };

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<string | null> => {
-    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<StoredPersonalKey | null> => {
+    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;
     },

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,

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<string, unknown>): 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]),
 });