fix: redact every column that names a user, verify unrecorded MySQL columns

packages/sync-clickhouse/src/destination.ts

@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey } from '@workos/sync-engine'
+import { encodeKey, userReferenceColumns, userReferencePaths } from '@workos/sync-engine'
 import { connect, isoOf, type ChClient } from './db/client.js'
 import { upsertRows, deleteRows, relinkRows, unlinkRows, keyExpression, tableRef, KEY_SEPARATOR } from './db/upsert.js'
 import { diffIdsInClickHouse } from './db/reconcile.js'
@@ -28,14 +28,6 @@ import { jsonPathArgs, quoteIdent, safeIdentifier } from './schema/ddl.js'
 
 const ident = (name: string) => quoteIdent(safeIdentifier(name))
 
-/**
- * The columns that name a user: a membership's or sign-in's `user_id`, and
- * an invitation's inviter and the user who accepted it. `redact` erases a
- * row that names the subject in any of them, and an archived or quarantined
- * event whose payload does.
- */
-const USER_REFERENCES = ['user_id', 'inviter_user_id', 'accepted_user_id']
-
 export interface ClickHouseDestinationOptions {
   /**
    * The HTTP interface, `https://user:password@host:8443`. Alternatively
@@ -229,9 +221,7 @@ export class ClickHouseDestination implements Destination {
       out.push({ table: label, rows: n })
     }
     for (const stream of streams) {
-      const refs = stream.columns.filter(
-        (c) => USER_REFERENCES.includes(c.name) || (c.name === 'id' && stream.name === 'users'),
-      )
+      const refs = userReferenceColumns(stream)
       if (refs.length === 0) continue
       await erase(
         stream.name,
@@ -241,15 +231,15 @@ export class ClickHouseDestination implements Destination {
         { env, id: userId },
       )
     }
-    const namedInData = USER_REFERENCES.map(
-      (field) => `JSONExtractString(\`_raw\`, ${jsonPathArgs(['data', field])}) = {id:String}`,
-    ).join(' OR ')
+    const namedInData = userReferencePaths(streams)
+      .map((p) => ` OR JSONExtractString(\`_raw\`, ${jsonPathArgs(['data', ...p])}) = {id:String}`)
+      .join('')
     // Archived events embed the payload, so they must be scrubbed too.
     await erase(
       'events',
       `${quoteIdent(this.schema)}.\`events\``,
       ' FINAL',
-      `\`_env\` = {env:String} AND (\`object_id\` = {id:String} OR ${namedInData})`,
+      `\`_env\` = {env:String} AND (\`object_id\` = {id:String}${namedInData})`,
       { env, id: userId },
     )
     // Quarantined events embed the payload too, and they may be the only copy
@@ -259,7 +249,7 @@ export class ClickHouseDestination implements Destination {
       '_rejected',
       `${quoteIdent(this.schema)}.\`_rejected\``,
       ' FINAL',
-      `sync_id = {syncId:String} AND env = {env:String} AND (object_id = {id:String} OR ${namedInData})`,
+      `sync_id = {syncId:String} AND env = {env:String} AND (object_id = {id:String}${namedInData})`,
       { syncId: this.syncId, env, id: userId },
     )
     return out

packages/sync-engine/src/catalog.ts

@@ -116,7 +116,7 @@ export const STREAMS: StreamDef[] = [
     prefixes: ['om_'],
     columns: [
       { name: 'id', type: 'text', path: 'id' },
-      { name: 'user_id', type: 'text', path: 'user_id', index: true },
+      { name: 'user_id', type: 'text', path: 'user_id', index: true, references: 'users' },
       { name: 'organization_id', type: 'text', path: 'organization_id', index: true },
       { name: 'status', type: 'text', path: 'status' },
       { name: 'role_slug', type: 'text', path: ['role', 'slug'] },
@@ -183,8 +183,8 @@ export const STREAMS: StreamDef[] = [
       { name: 'email', type: 'text', path: 'email', index: true },
       { name: 'state', type: 'text', path: 'state' },
       { name: 'organization_id', type: 'text', path: 'organization_id', index: true },
-      { name: 'inviter_user_id', type: 'text', path: 'inviter_user_id' },
-      { name: 'accepted_user_id', type: 'text', path: 'accepted_user_id' },
+      { name: 'inviter_user_id', type: 'text', path: 'inviter_user_id', references: 'users' },
+      { name: 'accepted_user_id', type: 'text', path: 'accepted_user_id', references: 'users' },
       { name: 'role_slug', type: 'text', path: ['role', 'slug'] },
       { name: 'expires_at', type: 'timestamptz', path: 'expires_at' },
       { name: 'accepted_at', type: 'timestamptz', path: 'accepted_at' },
@@ -251,7 +251,7 @@ export const STREAMS: StreamDef[] = [
       { name: 'event', type: 'text', path: 'event', index: true },
       { name: 'type', type: 'text', path: 'type' },
       { name: 'status', type: 'text', path: 'status' },
-      { name: 'user_id', type: 'text', path: 'user_id', index: true },
+      { name: 'user_id', type: 'text', path: 'user_id', index: true, references: 'users' },
       { name: 'email', type: 'text', path: 'email', index: true },
       { name: 'ip_address', type: 'text', path: 'ip_address' },
       { name: 'user_agent', type: 'text', path: 'user_agent' },
@@ -271,7 +271,7 @@ export const STREAMS: StreamDef[] = [
     prefixes: ['session_'],
     columns: [
       { name: 'id', type: 'text', path: 'id' },
-      { name: 'user_id', type: 'text', path: 'user_id', index: true },
+      { name: 'user_id', type: 'text', path: 'user_id', index: true, references: 'users' },
       { name: 'organization_id', type: 'text', path: 'organization_id', index: true },
       { name: 'status', type: 'text', path: 'status' },
       { name: 'auth_method', type: 'text', path: 'auth_method' },
@@ -591,6 +591,48 @@ export const STREAMS: StreamDef[] = [
   },
 ] satisfies CatalogStreamDef[]
 
+/**
+ * Whether `col` holds a user id: declared with `references: 'users'`, or
+ * named `user_id`, which an `extraColumns` entry may be without declaring
+ * anything, and which always counted.
+ */
+const namesUser = (col: ColumnDef): boolean => col.references === 'users' || col.name === 'user_id'
+
+/**
+ * The columns of `stream` that name a user, and the `users` stream's own
+ * key: a membership's or sign-in's `user_id`, an invitation's inviter and
+ * the user who accepted it, an extra column declared as referencing
+ * `users`. `redact` erases a row that names the subject in any of them.
+ */
+export function userReferenceColumns(stream: StreamDef): ColumnDef[] {
+  return stream.columns.filter((c) => namesUser(c) || (stream.name === 'users' && stream.key.includes(c.name)))
+}
+
+/**
+ * The payload paths that name a user, each as its steps under the object:
+ * `['user_id']`, or `['metadata', 'owner_user_id']` for a nested extra
+ * column. Every stream of the catalog contributes, whether selected or
+ * not, because the archive keeps every event type the tail subscribed to;
+ * `streams` adds the extra columns of the selection. An event's `data`
+ * carries the object, so `redact` scrubs archived and quarantined events
+ * by `data` followed by each.
+ */
+export function userReferencePaths(streams: StreamDef[]): string[][] {
+  const seen = new Set<string>()
+  const out: string[][] = []
+  for (const stream of [...STREAMS, ...streams]) {
+    for (const col of stream.columns) {
+      if (!namesUser(col)) continue
+      const path = Array.isArray(col.path) ? col.path : [col.path]
+      const key = path.join('\u0000')
+      if (seen.has(key)) continue
+      seen.add(key)
+      out.push(path)
+    }
+  }
+  return out
+}
+
 export const streamByName = (name: string): StreamDef | undefined => STREAMS.find((s) => s.name === name)
 
 /**

packages/sync-engine/src/index.ts

@@ -37,7 +37,16 @@ export type {
 export { StateStore } from './state.js'
 export { redactPayload, redactRows, redactEvent, RedactingDestination, redacting } from './redact.js'
 
-export { STREAMS, streamByName, streamForId, defaultStreams, selectStreams, subscribedEventTypes } from './catalog.js'
+export {
+  STREAMS,
+  streamByName,
+  streamForId,
+  defaultStreams,
+  selectStreams,
+  subscribedEventTypes,
+  userReferenceColumns,
+  userReferencePaths,
+} from './catalog.js'
 export { routeEvent, sourceUpdatedAt } from './router.js'
 export type { RouteAction } from './router.js'
 

packages/sync-engine/src/types.ts

@@ -17,6 +17,12 @@ export interface ColumnDef {
   path: string | string[]
   /** Create a btree index on this column. */
   index?: boolean
+  /**
+   * The stream whose key this column holds, when it holds one: `user_id`
+   * references `users`. What `redact` follows to every row that names a
+   * subject, on every destination alike.
+   */
+  references?: string
 }
 
 /**

packages/sync-engine/test/catalog.test.ts

@@ -1,5 +1,12 @@
 import { describe, it, expect } from 'vitest'
-import { STREAMS, defaultStreams, selectStreams } from '../src/catalog.js'
+import {
+  streamByName,
+  STREAMS,
+  defaultStreams,
+  selectStreams,
+  userReferenceColumns,
+  userReferencePaths,
+} from '../src/catalog.js'
 
 describe('selectStreams', () => {
   it('selects the defaults, with events, when given nothing', () => {
@@ -28,3 +35,39 @@ describe('selectStreams', () => {
     )
   })
 })
+
+describe('user references', () => {
+  it('names the columns of a stream that hold a user id, and the users key itself', () => {
+    expect(userReferenceColumns(streamByName('users')!).map((c) => c.name)).toEqual(['id'])
+    expect(userReferenceColumns(streamByName('organization_memberships')!).map((c) => c.name)).toEqual(['user_id'])
+    expect(userReferenceColumns(streamByName('invitations')!).map((c) => c.name)).toEqual([
+      'inviter_user_id',
+      'accepted_user_id',
+    ])
+    expect(userReferenceColumns(streamByName('organizations')!)).toEqual([])
+  })
+
+  it('collects the payload paths that name a user across the given streams, extra and nested ones included', () => {
+    expect(
+      userReferencePaths(STREAMS)
+        .map((p) => p.join('.'))
+        .sort(),
+    ).toEqual(['accepted_user_id', 'inviter_user_id', 'user_id'])
+    const withExtra = selectStreams(['organizations'], {
+      organizations: [
+        { name: 'owner', type: 'text', path: ['metadata', 'owner_user_id'], references: 'users' },
+        // Named `user_id` without declaring anything: counted, as it always was.
+        { name: 'user_id', type: 'text', path: 'user_id' },
+      ],
+    })
+    const organizations = withExtra.find((s) => s.name === 'organizations')!
+    expect(userReferenceColumns(organizations).map((c) => c.name)).toEqual(['owner', 'user_id'])
+    // The catalog's own paths stay: the archive holds every event type,
+    // whichever streams were selected.
+    expect(
+      userReferencePaths(withExtra)
+        .map((p) => p.join('.'))
+        .sort(),
+    ).toEqual(['accepted_user_id', 'inviter_user_id', 'metadata.owner_user_id', 'user_id'])
+  })
+})

packages/sync-mongodb/src/destination.ts

@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey } from '@workos/sync-engine'
+import { encodeKey, userReferenceColumns, userReferencePaths } from '@workos/sync-engine'
 import type { Collection, Db, Document, MongoClient } from 'mongodb'
 import { connect, databaseFromUrl } from './db/client.js'
 import { isTransientMongoError } from './db/errors.js'
@@ -240,23 +240,27 @@ export class MongoDestination implements Destination {
   async redact(streams: TableDef[], env: string, userId: string): Promise<{ table: string; rows: number }[]> {
     const out: { table: string; rows: number }[] = []
     for (const stream of streams) {
-      const col = stream.columns.find((c) => c.name === 'user_id' || (c.name === 'id' && stream.name === 'users'))
-      if (!col) continue
-      const r = await this.mirror(stream).deleteMany({ _env: env, [fieldPath(col.path)]: userId })
+      const refs = userReferenceColumns(stream)
+      if (refs.length === 0) continue
+      const r = await this.mirror(stream).deleteMany({
+        _env: env,
+        $or: refs.map((c) => ({ [fieldPath(c.path)]: userId })),
+      })
       if (r.deletedCount > 0) out.push({ table: stream.name, rows: r.deletedCount })
     }
+    const namedInData = userReferencePaths(streams).map((p) => ({ [`_raw.data.${p.join('.')}`]: userId }))
     // Archived events embed the payload, so they must be scrubbed too.
     const ev = await this.store
       .collection<MirrorDoc>(this.names.mirror('events'))
-      .deleteMany({ _env: env, $or: [{ '_raw.data.id': userId }, { '_raw.data.user_id': userId }] })
+      .deleteMany({ _env: env, $or: [{ '_raw.data.id': userId }, ...namedInData] })
     if (ev.deletedCount > 0) out.push({ table: 'events', rows: ev.deletedCount })
     // Quarantined events embed the payload too, and they may be the only copy
     // of it: an event rejected before its archive write succeeded is nowhere
     // in `events`.
     const rj = await this.quarantine.deleteMany({
       sync_id: this.syncId,
       env,
-      $or: [{ object_id: userId }, { '_raw.data.user_id': userId }],
+      $or: [{ object_id: userId }, ...namedInData],
     })
     if (rj.deletedCount > 0) out.push({ table: '_rejected', rows: rj.deletedCount })
     return out

packages/sync-mongodb/test/engine.test.ts

@@ -785,12 +785,20 @@ describe('redaction', () => {
   it('erases a subject from the mirror and the event archive', async () => {
     const user = fake.createUser({ email: 'forget@example.com' })
     fake.emitAuthAttempt(user['id'] as string, true)
+    // Invitations name a user as the one who accepted and as the inviter.
+    const org = fake.createOrganization()
+    const accepted = fake.createInvitation(org['id'] as string)
+    fake.acceptInvitation(accepted['id'] as string, user['id'] as string)
+    fake.createInvitation(org['id'] as string, { inviter_user_id: user['id'] })
+    fake.createInvitation(org['id'] as string)
     const engine = makeEngine(t, fake)
     await engine.backfill()
     await engine.tick()
 
     const removed = await engine.redact(user['id'] as string)
     expect(removed.length).toBeGreaterThan(0)
+    expect(removed).toContainEqual({ table: 'invitations', rows: 2 })
+    expect(await count(t, 'invitations')).toBe(1)
     expect(await count(t, 'users')).toBe(0)
     expect(await count(t, 'auth_attempts')).toBe(0)
     // The archived payloads carry the same personal data and must go too.

packages/sync-mysql/README.md

@@ -105,10 +105,14 @@ npx workos-sync-mysql init --print-sql --schema <branch database> > schema.sql
 npx workos-sync-mysql init --schema <branch database>
 ```
 
-Once the tables and views are in place, `init` runs no DDL at all: it checks
-the live columns, records what it finds, and records the fingerprint. A later
-catalog change that adds a column is the one thing that needs another deploy
-request; `init` reports the missing column and the DDL for it.
+Once the tables and views are in place, `init` changes no schema: it checks
+the live columns, records what it finds, and records the fingerprint. To
+check a column it has no record of, it creates a temporary table from the
+catalog's clauses and drops it again within the session, which is session
+state rather than a schema change, needs only the `CREATE TEMPORARY TABLES`
+grant, and passes a host that gates schema changes. A later catalog change
+that adds a column is the one thing that needs another deploy request;
+`init` reports the missing column and the DDL for it.
 
 `listen` with a health port answers `GET /healthz` with `200` while the tail
 is polling and `503` once a poll is overdue, and `GET /metrics` in Prometheus

packages/sync-mysql/src/destination.ts

@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey } from '@workos/sync-engine'
+import { encodeKey, userReferenceColumns, userReferencePaths } from '@workos/sync-engine'
 import { connect, type SqlClient } from './db/client.js'
 import { upsertRows, deleteRows, relinkRows, unlinkRows } from './db/upsert.js'
 import { diffIdsInSql } from './db/reconcile.js'
@@ -205,29 +205,35 @@ export class MysqlDestination implements Destination {
   async redact(streams: TableDef[], env: string, userId: string): Promise<{ table: string; rows: number }[]> {
     const out: { table: string; rows: number }[] = []
     for (const stream of streams) {
-      const col = stream.columns.find((c) => c.name === 'user_id' || (c.name === 'id' && stream.name === 'users'))
-      if (!col) continue
+      const refs = userReferenceColumns(stream)
+      if (refs.length === 0) continue
       const r = await this.client.query(
-        `delete from ${this.table(stream)} where \`_env\` = ? and ${ident(col.name)} = ?`,
-        [env, userId],
+        `delete from ${this.table(stream)} where \`_env\` = ? and (${refs.map((c) => `${ident(c.name)} = ?`).join(' or ')})`,
+        [env, ...refs.map(() => userId)],
       )
       if (r.affectedRows > 0) out.push({ table: stream.name, rows: r.affectedRows })
     }
-    const userIdInData = `json_value(\`_raw\`, ${jsonPath(['data', 'user_id'])} returning char(${KEY_LENGTH}) null on error)`
+    const paths = userReferencePaths(streams)
+    const namedInData = paths
+      .map(
+        (p) => ` or json_value(\`_raw\`, ${jsonPath(['data', ...p])} returning char(${KEY_LENGTH}) null on error) = ?`,
+      )
+      .join('')
+    const named = paths.map(() => userId)
     // Archived events embed the payload, so they must be scrubbed too.
     const ev = await this.client.query(
       `delete from ${quoteIdent(this.schema)}.\`events\`
-       where \`_env\` = ? and (\`object_id\` = ? or ${userIdInData} = ?)`,
-      [env, userId, userId],
+       where \`_env\` = ? and (\`object_id\` = ?${namedInData})`,
+      [env, userId, ...named],
     )
     if (ev.affectedRows > 0) out.push({ table: 'events', rows: ev.affectedRows })
     // Quarantined events embed the payload too, and they may be the only copy
     // of it: an event rejected before its archive write succeeded is nowhere
     // in `events`.
     const rj = await this.client.query(
       `delete from ${quoteIdent(this.schema)}.\`_rejected\`
-       where sync_id = ? and env = ? and (object_id = ? or ${userIdInData} = ?)`,
-      [this.syncId, env, userId, userId],
+       where sync_id = ? and env = ? and (object_id = ?${namedInData})`,
+      [this.syncId, env, userId, ...named],
     )
     if (rj.affectedRows > 0) out.push({ table: '_rejected', rows: rj.affectedRows })
     return out