From 4760e8c5f1c90505ae657498b16f9c4eadd3c59c Mon Sep 17 00:00:00 2001
From: "Garen J. Torikian" <gjtorikian@users.noreply.github.com>
Date: Fri, 6 Nov 2026 09:09:39 -0500
Subject: [PATCH 1/5] fix: redact every column that names a user, verify
 unrecorded MySQL columns

`redact --user` in Postgres, MySQL and MongoDB erased rows by `user_id`
only. An invitation names a user as its inviter and as the user who
accepted it, so those rows, and the archived and quarantined events
carrying those payloads, kept the subject's id and email after the
command reported success. Every column and payload field that names a
user is matched now, as in the ClickHouse package.

MySQL recorded a column with no `_columns` record as current without a
look at the table, so after a schema applied by hand a changed path was
recorded as installed with the column still reading the old path. The
catalog's clause now goes onto a temporary table, and `SHOW CREATE
TABLE` renders that and the live table with the same server, so the two
compare as text: a match is recorded, anything else is drift.
---
 packages/sync-mongodb/src/destination.ts   | 24 +++++++--
 packages/sync-mongodb/test/engine.test.ts  |  8 +++
 packages/sync-mysql/src/destination.ts     | 31 ++++++++----
 packages/sync-mysql/src/schema/drift.ts    | 58 ++++++++++++++++++++--
 packages/sync-mysql/test/drift.test.ts     | 27 ++++++++++
 packages/sync-mysql/test/engine.test.ts    |  9 ++++
 packages/sync-postgres/src/destination.ts  | 21 ++++++--
 packages/sync-postgres/test/engine.test.ts |  9 ++++
 8 files changed, 165 insertions(+), 22 deletions(-)

diff --git a/packages/sync-mongodb/src/destination.ts b/packages/sync-mongodb/src/destination.ts
index ac3d9e2..c1a466e 100644
--- a/packages/sync-mongodb/src/destination.ts
+++ b/packages/sync-mongodb/src/destination.ts
@@ -35,6 +35,14 @@ import {
 import { migrate, type MigrateResult } from './schema/migrate.js'
 import { fieldPath, keyPath, Names } from './schema/names.js'
 
+/**
+ * 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 MongoDestinationOptions {
   /** Connection string (`mongodb://` or `mongodb+srv://`). Alternatively pass a pre-built `client`. */
   url?: string
@@ -240,15 +248,21 @@ 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 = stream.columns.filter(
+        (c) => USER_REFERENCES.includes(c.name) || (c.name === 'id' && stream.name === 'users'),
+      )
+      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 = USER_REFERENCES.map((f) => ({ [`_raw.data.${f}`]: 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
@@ -256,7 +270,7 @@ export class MongoDestination implements Destination {
     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
diff --git a/packages/sync-mongodb/test/engine.test.ts b/packages/sync-mongodb/test/engine.test.ts
index fe1f90f..5160b18 100644
--- a/packages/sync-mongodb/test/engine.test.ts
+++ b/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.
diff --git a/packages/sync-mysql/src/destination.ts b/packages/sync-mysql/src/destination.ts
index 4cf86c1..1ceb58a 100644
--- a/packages/sync-mysql/src/destination.ts
+++ b/packages/sync-mysql/src/destination.ts
@@ -28,6 +28,14 @@ import { jsonPath, quoteIdent, safeIdentifier, KEY_LENGTH } from './schema/ddl.j
 
 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 MysqlDestinationOptions {
   /** Connection string, `mysql://user:password@host:3306/db`. Alternatively pass a pre-built `db`. */
   databaseUrl?: string
@@ -205,20 +213,25 @@ 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 = stream.columns.filter(
+        (c) => USER_REFERENCES.includes(c.name) || (c.name === 'id' && stream.name === 'users'),
+      )
+      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 namedInData = USER_REFERENCES.map(
+      (f) => `json_value(\`_raw\`, ${jsonPath(['data', f])} returning char(${KEY_LENGTH}) null on error) = ?`,
+    ).join(' or ')
+    const named = USER_REFERENCES.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\` = ? or ${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
@@ -226,8 +239,8 @@ export class MysqlDestination implements Destination {
     // 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 = ? or ${namedInData})`,
+      [this.syncId, env, userId, ...named],
     )
     if (rj.affectedRows > 0) out.push({ table: '_rejected', rows: rj.affectedRows })
     return out
diff --git a/packages/sync-mysql/src/schema/drift.ts b/packages/sync-mysql/src/schema/drift.ts
index b488ce5..0c72846 100644
--- a/packages/sync-mysql/src/schema/drift.ts
+++ b/packages/sync-mysql/src/schema/drift.ts
@@ -13,8 +13,11 @@
  * `information_schema` shows: MySQL rewrites a generated expression into its
  * own normal form, which differs between versions, so text written by the
  * catalog and text read back from the server never compare equal. A
- * generated column that has no record (a table made by hand to the right
- * shape) is recorded as current. The live type is always compared with the
+ * generated column that has no record (a schema applied by hand from `init
+ * --print-sql`) is judged by the table itself: the catalog's clause goes
+ * onto a temporary table, and `SHOW CREATE TABLE` renders that and the
+ * live table with the same server, so the two compare as text. Only a
+ * match is recorded; anything else is drift. The live type is always compared with the
  * catalog's, record or no record.
  *
  * Never touched, only reported: key columns (a rebuild would drop the
@@ -24,6 +27,7 @@
  * names, which still compute from `_raw` with their old expression and are
  * harmless to leave.
  */
+import { randomBytes } from 'node:crypto'
 import type { Logger, TableDef } from '@workos/sync-engine'
 import type { SqlClient } from '../db/client.js'
 import {
@@ -97,12 +101,44 @@ export async function recordColumn(db: SqlClient, schema: string, table: string,
   )
 }
 
+/** The generated expressions of a table as `SHOW CREATE TABLE` prints them, by column. */
+async function createdExpressions(db: SqlClient, ref: string): Promise<Map<string, string>> {
+  const r = await db.query<{ 'Create Table': string }>(`show create table ${ref}`)
+  const out = new Map<string, string>()
+  for (const line of (r.rows[0]?.['Create Table'] ?? '').split('\n')) {
+    const m = /^\s*`((?:[^`]|``)+)`\s.*?\sGENERATED ALWAYS AS \((.*)\) (?:STORED|VIRTUAL)(?: NOT NULL)?,?$/.exec(line)
+    if (m) out.set(m[1]!.replace(/``/g, '`'), m[2]!)
+  }
+  return out
+}
+
+/**
+ * The server's own spelling of each of a stream's generated expressions: a
+ * temporary table built from the catalog's clauses, read back the way the
+ * live table is. On one pinned connection, since a temporary table lives in
+ * its session.
+ */
+async function canonicalExpressions(db: SqlClient, schema: string, stream: TableDef): Promise<Map<string, string>> {
+  const ref = `${quoteIdent(schema)}.${quoteIdent(`_probe_${randomBytes(4).toString('hex')}`)}`
+  const clauses = stream.columns.map((c) => generatedColumnSql(c, stream))
+  return db.pinned(async (conn) => {
+    await conn.exec(`create temporary table ${ref} (\`_raw\` json, ${clauses.join(', ')})`)
+    try {
+      return await createdExpressions(conn, ref)
+    } finally {
+      await conn.exec(`drop temporary table if exists ${ref}`)
+    }
+  })
+}
+
 export async function detectColumnDrift(db: SqlClient, streams: TableDef[], schema: string): Promise<ColumnDrift[]> {
   const out: ColumnDrift[] = []
   for (const stream of streams) {
     const table = safeIdentifier(stream.name)
     const live = await liveColumns(db, schema, table)
     const recorded = await recordedColumns(db, schema, table)
+    let created: Map<string, string> | undefined
+    let canonical: Map<string, string> | undefined
     const catalog = new Set<string>()
     for (const col of stream.columns) {
       const name = safeIdentifier(col.name)
@@ -133,7 +169,23 @@ export async function detectColumnDrift(db: SqlClient, streams: TableDef[], sche
       }
       const record = recorded.get(name)
       if (record === undefined) {
-        await recordColumn(db, schema, table, name, expected)
+        // Nothing says what this column was created with, so the table is
+        // asked, in its own spelling of the clause the catalog would write.
+        created ??= await createdExpressions(db, `${quoteIdent(schema)}.${quoteIdent(table)}`)
+        canonical ??= await canonicalExpressions(db, schema, stream)
+        const spelled = canonical.get(name)
+        if (spelled !== undefined && spelled === created.get(name)) {
+          await recordColumn(db, schema, table, name, expected)
+          continue
+        }
+        out.push({
+          stream: stream.name,
+          column: name,
+          kind: 'expression',
+          expected,
+          actual: created.get(name) ?? actual.generation_expression,
+          key,
+        })
         continue
       }
       if (record !== expected) {
diff --git a/packages/sync-mysql/test/drift.test.ts b/packages/sync-mysql/test/drift.test.ts
index d638e98..6953152 100644
--- a/packages/sync-mysql/test/drift.test.ts
+++ b/packages/sync-mysql/test/drift.test.ts
@@ -140,6 +140,33 @@ describeDb('column drift detection', () => {
     expect(await recorded([changed])).toBe(true)
   })
 
+  it('judges an unrecorded column by the table itself, as after a schema applied by hand', async () => {
+    await migrate(db, [users], db.schema)
+    const forgetAll = async () => {
+      await db.query(`delete from \`${db.schema}\`.\`_columns\``)
+      await db.query(`delete from \`${db.schema}\`.\`_migrations\``)
+    }
+    // The same catalog again: every column checks out and is recorded.
+    await forgetAll()
+    const same = await migrate(db, [users], db.schema)
+    expect(same.drift).toEqual([])
+    expect(await recorded([users])).toBe(true)
+
+    // A changed path with no record to compare against. `ADD COLUMN IF NOT
+    // EXISTS` leaves the column as it was; the table says so, and that is
+    // drift to report, not a clause to record as installed.
+    await forgetAll()
+    await insertUser({ id: 'user_1', email: 'old@example.com', primary_email: 'new@example.com' })
+    const moved = await migrate(db, [usersMovedEmail], db.schema)
+    expect(moved.drift.map((d) => [d.column, d.kind])).toEqual([['email', 'expression']])
+    expect(await recorded([usersMovedEmail])).toBe(false)
+    const fixed = await migrate(db, [usersMovedEmail], db.schema, undefined, { repair: true })
+    expect(fixed.drift).toEqual([])
+    const row = await db.query<{ email: string }>(`select email from \`${db.schema}\`.users`)
+    expect(row.rows[0]!.email).toBe('new@example.com')
+    expect(await recorded([usersMovedEmail])).toBe(true)
+  })
+
   it('refuses to repair a key column', async () => {
     await migrate(db, [users], db.schema)
     const rekeyed = withColumn(users, { name: 'id', type: 'text', path: 'uid' })
diff --git a/packages/sync-mysql/test/engine.test.ts b/packages/sync-mysql/test/engine.test.ts
index 0e88fa0..154e0fd 100644
--- a/packages/sync-mysql/test/engine.test.ts
+++ b/packages/sync-mysql/test/engine.test.ts
@@ -494,12 +494,21 @@ describeDb('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(db, fake)
     await engine.backfill()
     await engine.tick()
 
     const removed = await engine.redact(user['id'] as string)
     expect(removed).toContainEqual({ table: 'users', rows: 1 })
+    expect(removed).toContainEqual({ table: 'invitations', rows: 2 })
+    const invitations = await db.query<{ n: number }>(`select count(*) as n from \`${db.schema}\`.invitations`)
+    expect(Number(invitations.rows[0]!.n)).toBe(1)
     expect(removed).toContainEqual({ table: 'auth_attempts', rows: 1 })
     expect(removed.find((r) => r.table === 'events')!.rows).toBeGreaterThan(0)
 
diff --git a/packages/sync-postgres/src/destination.ts b/packages/sync-postgres/src/destination.ts
index 6a7b3be..97037e4 100644
--- a/packages/sync-postgres/src/destination.ts
+++ b/packages/sync-postgres/src/destination.ts
@@ -28,6 +28,14 @@ import { 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 PostgresDestinationOptions {
   /** Connection string. Alternatively pass a pre-built `db`. */
   databaseUrl?: string
@@ -204,19 +212,22 @@ export class PostgresDestination 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 = stream.columns.filter(
+        (c) => USER_REFERENCES.includes(c.name) || (c.name === 'id' && stream.name === 'users'),
+      )
+      if (refs.length === 0) continue
       const r = await this.client.query(
         `delete from ${this.table(stream)}
-         where "_env" = $1 and ${ident(col.name)} = $2 returning 1`,
+         where "_env" = $1 and (${refs.map((c) => `${ident(c.name)} = $2`).join(' or ')}) returning 1`,
         [env, userId],
       )
       if (r.rows.length > 0) out.push({ table: stream.name, rows: r.rows.length })
     }
+    const namedInData = (param: string) => USER_REFERENCES.map((f) => `"_raw"->'data'->>'${f}' = ${param}`).join(' or ')
     // 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" = $1 and ("object_id" = $2 or "_raw"->'data'->>'user_id' = $2) returning 1`,
+       where "_env" = $1 and ("object_id" = $2 or ${namedInData('$2')}) returning 1`,
       [env, userId],
     )
     if (ev.rows.length > 0) out.push({ table: 'events', rows: ev.rows.length })
@@ -225,7 +236,7 @@ export class PostgresDestination implements Destination {
     // in `events`.
     const rj = await this.client.query(
       `delete from ${quoteIdent(this.schema)}."_rejected"
-       where sync_id = $1 and env = $2 and ("object_id" = $3 or "_raw"->'data'->>'user_id' = $3) returning 1`,
+       where sync_id = $1 and env = $2 and ("object_id" = $3 or ${namedInData('$3')}) returning 1`,
       [this.syncId, env, userId],
     )
     if (rj.rows.length > 0) out.push({ table: '_rejected', rows: rj.rows.length })
diff --git a/packages/sync-postgres/test/engine.test.ts b/packages/sync-postgres/test/engine.test.ts
index d0b3fe1..7829cfd 100644
--- a/packages/sync-postgres/test/engine.test.ts
+++ b/packages/sync-postgres/test/engine.test.ts
@@ -1393,12 +1393,21 @@ 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(db, 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 })
+    const invitations = await db.query<{ n: string }>(`select count(*)::text as n from workos.invitations`)
+    expect(Number(invitations.rows[0]!.n)).toBe(1)
 
     const remaining = await db.query<{ n: string }>(`select count(*)::text as n from workos.users`)
     expect(Number(remaining.rows[0]!.n)).toBe(0)

From a803bfb7e7565ea48c0da866a0e2883d1d11c410 Mon Sep 17 00:00:00 2001
From: "Garen J. Torikian" <gjtorikian@users.noreply.github.com>
Date: Fri, 6 Nov 2026 09:23:09 -0500
Subject: [PATCH 2/5] fix: let the catalog say which columns name a user, for
 every destination

Which columns hold a user id is a fact about the catalog, not about any
database, yet each destination carried its own list. A column can now
declare what it references (`user_id` references `users`), and the
engine answers `userReferenceColumns(stream)` for the mirror rows and
`userReferenceFields()` for the archived payloads; the four `redact`
implementations ask it instead of repeating the list.

The MySQL README's "no DDL after the first init" promise now says what
checking an unrecorded column does: a temporary table created and
dropped within the session, which is session state, not a schema change.
---
 packages/sync-clickhouse/src/destination.ts | 20 +++--------
 packages/sync-engine/src/catalog.ts         | 38 ++++++++++++++++++---
 packages/sync-engine/src/index.ts           | 11 +++++-
 packages/sync-engine/src/types.ts           |  6 ++++
 packages/sync-engine/test/catalog.test.ts   | 25 +++++++++++++-
 packages/sync-mongodb/src/destination.ts    | 16 ++-------
 packages/sync-mysql/README.md               | 12 ++++---
 packages/sync-mysql/src/destination.ts      | 22 ++++--------
 packages/sync-postgres/src/destination.ts   | 19 ++++-------
 9 files changed, 101 insertions(+), 68 deletions(-)

diff --git a/packages/sync-clickhouse/src/destination.ts b/packages/sync-clickhouse/src/destination.ts
index 78b2a3e..0d8f5df 100644
--- a/packages/sync-clickhouse/src/destination.ts
+++ b/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, userReferenceFields } 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,9 +231,9 @@ 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 = userReferenceFields()
+      .map((field) => `JSONExtractString(\`_raw\`, ${jsonPathArgs(['data', field])}) = {id:String}`)
+      .join(' OR ')
     // Archived events embed the payload, so they must be scrubbed too.
     await erase(
       'events',
diff --git a/packages/sync-engine/src/catalog.ts b/packages/sync-engine/src/catalog.ts
index 5998af5..6144fee 100644
--- a/packages/sync-engine/src/catalog.ts
+++ b/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,34 @@ export const STREAMS: StreamDef[] = [
   },
 ] satisfies CatalogStreamDef[]
 
+/**
+ * The columns of `stream` that name a user: those declared as referencing
+ * `users`, 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. `redact`
+ * erases a row that names the subject in any of them.
+ */
+export function userReferenceColumns(stream: StreamDef): ColumnDef[] {
+  return stream.columns.filter(
+    (c) => c.references === 'users' || (stream.name === 'users' && stream.key.includes(c.name)),
+  )
+}
+
+/**
+ * The top-level payload fields that name a user, across the catalog. An
+ * event's `data` carries the object, so a field that references `users` on
+ * any stream does there too; `redact` scrubs archived and quarantined
+ * events by them.
+ */
+export function userReferenceFields(): string[] {
+  const out = new Set<string>()
+  for (const stream of STREAMS) {
+    for (const col of stream.columns) {
+      if (col.references === 'users' && typeof col.path === 'string') out.add(col.path)
+    }
+  }
+  return [...out]
+}
+
 export const streamByName = (name: string): StreamDef | undefined => STREAMS.find((s) => s.name === name)
 
 /**
diff --git a/packages/sync-engine/src/index.ts b/packages/sync-engine/src/index.ts
index 8d8c173..60e00d1 100644
--- a/packages/sync-engine/src/index.ts
+++ b/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,
+  userReferenceFields,
+} from './catalog.js'
 export { routeEvent, sourceUpdatedAt } from './router.js'
 export type { RouteAction } from './router.js'
 
diff --git a/packages/sync-engine/src/types.ts b/packages/sync-engine/src/types.ts
index bd4930f..599c6e8 100644
--- a/packages/sync-engine/src/types.ts
+++ b/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
 }
 
 /**
diff --git a/packages/sync-engine/test/catalog.test.ts b/packages/sync-engine/test/catalog.test.ts
index 4a2ce46..e923074 100644
--- a/packages/sync-engine/test/catalog.test.ts
+++ b/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,
+  userReferenceFields,
+} from '../src/catalog.js'
 
 describe('selectStreams', () => {
   it('selects the defaults, with events, when given nothing', () => {
@@ -28,3 +35,19 @@ 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 fields that name a user across the catalog', () => {
+    expect(userReferenceFields().sort()).toEqual(['accepted_user_id', 'inviter_user_id', 'user_id'])
+  })
+})
diff --git a/packages/sync-mongodb/src/destination.ts b/packages/sync-mongodb/src/destination.ts
index c1a466e..aba7862 100644
--- a/packages/sync-mongodb/src/destination.ts
+++ b/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, userReferenceFields } 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'
@@ -35,14 +35,6 @@ import {
 import { migrate, type MigrateResult } from './schema/migrate.js'
 import { fieldPath, keyPath, Names } from './schema/names.js'
 
-/**
- * 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 MongoDestinationOptions {
   /** Connection string (`mongodb://` or `mongodb+srv://`). Alternatively pass a pre-built `client`. */
   url?: string
@@ -248,9 +240,7 @@ 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 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
       const r = await this.mirror(stream).deleteMany({
         _env: env,
@@ -258,7 +248,7 @@ export class MongoDestination implements Destination {
       })
       if (r.deletedCount > 0) out.push({ table: stream.name, rows: r.deletedCount })
     }
-    const namedInData = USER_REFERENCES.map((f) => ({ [`_raw.data.${f}`]: userId }))
+    const namedInData = userReferenceFields().map((f) => ({ [`_raw.data.${f}`]: userId }))
     // Archived events embed the payload, so they must be scrubbed too.
     const ev = await this.store
       .collection<MirrorDoc>(this.names.mirror('events'))
diff --git a/packages/sync-mysql/README.md b/packages/sync-mysql/README.md
index 69ce2ed..a3de7c1 100644
--- a/packages/sync-mysql/README.md
+++ b/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
diff --git a/packages/sync-mysql/src/destination.ts b/packages/sync-mysql/src/destination.ts
index 1ceb58a..59af78b 100644
--- a/packages/sync-mysql/src/destination.ts
+++ b/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, userReferenceFields } 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'
@@ -28,14 +28,6 @@ import { jsonPath, quoteIdent, safeIdentifier, KEY_LENGTH } from './schema/ddl.j
 
 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 MysqlDestinationOptions {
   /** Connection string, `mysql://user:password@host:3306/db`. Alternatively pass a pre-built `db`. */
   databaseUrl?: string
@@ -213,9 +205,7 @@ 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 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
       const r = await this.client.query(
         `delete from ${this.table(stream)} where \`_env\` = ? and (${refs.map((c) => `${ident(c.name)} = ?`).join(' or ')})`,
@@ -223,10 +213,10 @@ export class MysqlDestination implements Destination {
       )
       if (r.affectedRows > 0) out.push({ table: stream.name, rows: r.affectedRows })
     }
-    const namedInData = USER_REFERENCES.map(
-      (f) => `json_value(\`_raw\`, ${jsonPath(['data', f])} returning char(${KEY_LENGTH}) null on error) = ?`,
-    ).join(' or ')
-    const named = USER_REFERENCES.map(() => userId)
+    const namedInData = userReferenceFields()
+      .map((f) => `json_value(\`_raw\`, ${jsonPath(['data', f])} returning char(${KEY_LENGTH}) null on error) = ?`)
+      .join(' or ')
+    const named = userReferenceFields().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\`
diff --git a/packages/sync-postgres/src/destination.ts b/packages/sync-postgres/src/destination.ts
index 97037e4..7e4a39a 100644
--- a/packages/sync-postgres/src/destination.ts
+++ b/packages/sync-postgres/src/destination.ts
@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey } from '@workos/sync-engine'
+import { encodeKey, userReferenceColumns, userReferenceFields } 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'
@@ -28,14 +28,6 @@ import { 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 PostgresDestinationOptions {
   /** Connection string. Alternatively pass a pre-built `db`. */
   databaseUrl?: string
@@ -212,9 +204,7 @@ export class PostgresDestination 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 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
       const r = await this.client.query(
         `delete from ${this.table(stream)}
@@ -223,7 +213,10 @@ export class PostgresDestination implements Destination {
       )
       if (r.rows.length > 0) out.push({ table: stream.name, rows: r.rows.length })
     }
-    const namedInData = (param: string) => USER_REFERENCES.map((f) => `"_raw"->'data'->>'${f}' = ${param}`).join(' or ')
+    const namedInData = (param: string) =>
+      userReferenceFields()
+        .map((f) => `"_raw"->'data'->>'${f}' = ${param}`)
+        .join(' or ')
     // Archived events embed the payload, so they must be scrubbed too.
     const ev = await this.client.query(
       `delete from ${quoteIdent(this.schema)}."events"

From fdd0943d2e41707a92d1b9d2781411a0a4d58597 Mon Sep 17 00:00:00 2001
From: "Garen J. Torikian" <gjtorikian@users.noreply.github.com>
Date: Fri, 6 Nov 2026 09:47:52 -0500
Subject: [PATCH 3/5] fix(engine): keep name-based user references, derive
 payload paths from the streams

An extra column named `user_id` counted as naming a user before and
still does, without declaring anything. The payload paths `redact`
scrubs archived and quarantined events by come from the effective
streams, extra columns included, as path steps rather than top-level
field names, so a nested reference such as `metadata.owner_user_id`
is scrubbed too. Each destination builds its JSON path from the steps:
a parameterised array for Postgres, the JSON path helpers elsewhere.
---
 packages/sync-clickhouse/src/destination.ts | 12 +++---
 packages/sync-engine/src/catalog.ts         | 43 +++++++++++++--------
 packages/sync-engine/src/index.ts           |  2 +-
 packages/sync-engine/test/catalog.test.ts   | 24 ++++++++++--
 packages/sync-mongodb/src/destination.ts    |  4 +-
 packages/sync-mysql/src/destination.ts      | 17 ++++----
 packages/sync-postgres/src/destination.ts   | 21 +++++-----
 7 files changed, 79 insertions(+), 44 deletions(-)

diff --git a/packages/sync-clickhouse/src/destination.ts b/packages/sync-clickhouse/src/destination.ts
index 0d8f5df..d18d371 100644
--- a/packages/sync-clickhouse/src/destination.ts
+++ b/packages/sync-clickhouse/src/destination.ts
@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey, userReferenceColumns, userReferenceFields } 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'
@@ -231,15 +231,15 @@ export class ClickHouseDestination implements Destination {
         { env, id: userId },
       )
     }
-    const namedInData = userReferenceFields()
-      .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
@@ -249,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
diff --git a/packages/sync-engine/src/catalog.ts b/packages/sync-engine/src/catalog.ts
index 6144fee..9918275 100644
--- a/packages/sync-engine/src/catalog.ts
+++ b/packages/sync-engine/src/catalog.ts
@@ -592,31 +592,42 @@ export const STREAMS: StreamDef[] = [
 ] satisfies CatalogStreamDef[]
 
 /**
- * The columns of `stream` that name a user: those declared as referencing
- * `users`, 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. `redact`
- * erases a row that names the subject in any of them.
+ * 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) => c.references === 'users' || (stream.name === 'users' && stream.key.includes(c.name)),
-  )
+  return stream.columns.filter((c) => namesUser(c) || (stream.name === 'users' && stream.key.includes(c.name)))
 }
 
 /**
- * The top-level payload fields that name a user, across the catalog. An
- * event's `data` carries the object, so a field that references `users` on
- * any stream does there too; `redact` scrubs archived and quarantined
- * events by them.
+ * The payload paths that name a user across `streams`, each as its steps
+ * under the object: `['user_id']`, or `['metadata', 'owner_user_id']` for a
+ * nested extra column. An event's `data` carries the object, so `redact`
+ * scrubs archived and quarantined events by `data` followed by each.
  */
-export function userReferenceFields(): string[] {
-  const out = new Set<string>()
-  for (const stream of STREAMS) {
+export function userReferencePaths(streams: StreamDef[]): string[][] {
+  const seen = new Set<string>()
+  const out: string[][] = []
+  for (const stream of streams) {
     for (const col of stream.columns) {
-      if (col.references === 'users' && typeof col.path === 'string') out.add(col.path)
+      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]
+  return out
 }
 
 export const streamByName = (name: string): StreamDef | undefined => STREAMS.find((s) => s.name === name)
diff --git a/packages/sync-engine/src/index.ts b/packages/sync-engine/src/index.ts
index 60e00d1..db6c82d 100644
--- a/packages/sync-engine/src/index.ts
+++ b/packages/sync-engine/src/index.ts
@@ -45,7 +45,7 @@ export {
   selectStreams,
   subscribedEventTypes,
   userReferenceColumns,
-  userReferenceFields,
+  userReferencePaths,
 } from './catalog.js'
 export { routeEvent, sourceUpdatedAt } from './router.js'
 export type { RouteAction } from './router.js'
diff --git a/packages/sync-engine/test/catalog.test.ts b/packages/sync-engine/test/catalog.test.ts
index e923074..1dbf7ba 100644
--- a/packages/sync-engine/test/catalog.test.ts
+++ b/packages/sync-engine/test/catalog.test.ts
@@ -5,7 +5,7 @@ import {
   defaultStreams,
   selectStreams,
   userReferenceColumns,
-  userReferenceFields,
+  userReferencePaths,
 } from '../src/catalog.js'
 
 describe('selectStreams', () => {
@@ -47,7 +47,25 @@ describe('user references', () => {
     expect(userReferenceColumns(streamByName('organizations')!)).toEqual([])
   })
 
-  it('collects the payload fields that name a user across the catalog', () => {
-    expect(userReferenceFields().sort()).toEqual(['accepted_user_id', 'inviter_user_id', 'user_id'])
+  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'])
+    expect(
+      userReferencePaths(withExtra)
+        .map((p) => p.join('.'))
+        .sort(),
+    ).toEqual(['metadata.owner_user_id', 'user_id'])
   })
 })
diff --git a/packages/sync-mongodb/src/destination.ts b/packages/sync-mongodb/src/destination.ts
index aba7862..bb3bc1b 100644
--- a/packages/sync-mongodb/src/destination.ts
+++ b/packages/sync-mongodb/src/destination.ts
@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey, userReferenceColumns, userReferenceFields } 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'
@@ -248,7 +248,7 @@ export class MongoDestination implements Destination {
       })
       if (r.deletedCount > 0) out.push({ table: stream.name, rows: r.deletedCount })
     }
-    const namedInData = userReferenceFields().map((f) => ({ [`_raw.data.${f}`]: userId }))
+    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'))
diff --git a/packages/sync-mysql/src/destination.ts b/packages/sync-mysql/src/destination.ts
index 59af78b..7f5fffa 100644
--- a/packages/sync-mysql/src/destination.ts
+++ b/packages/sync-mysql/src/destination.ts
@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey, userReferenceColumns, userReferenceFields } 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'
@@ -213,14 +213,17 @@ export class MysqlDestination implements Destination {
       )
       if (r.affectedRows > 0) out.push({ table: stream.name, rows: r.affectedRows })
     }
-    const namedInData = userReferenceFields()
-      .map((f) => `json_value(\`_raw\`, ${jsonPath(['data', f])} returning char(${KEY_LENGTH}) null on error) = ?`)
-      .join(' or ')
-    const named = userReferenceFields().map(() => userId)
+    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 ${namedInData})`,
+       where \`_env\` = ? and (\`object_id\` = ?${namedInData})`,
       [env, userId, ...named],
     )
     if (ev.affectedRows > 0) out.push({ table: 'events', rows: ev.affectedRows })
@@ -229,7 +232,7 @@ export class MysqlDestination implements Destination {
     // in `events`.
     const rj = await this.client.query(
       `delete from ${quoteIdent(this.schema)}.\`_rejected\`
-       where sync_id = ? and env = ? and (object_id = ? or ${namedInData})`,
+       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 })
diff --git a/packages/sync-postgres/src/destination.ts b/packages/sync-postgres/src/destination.ts
index 7e4a39a..6b06250 100644
--- a/packages/sync-postgres/src/destination.ts
+++ b/packages/sync-postgres/src/destination.ts
@@ -16,7 +16,7 @@ import type {
   WriteMode,
   WriteStats,
 } from '@workos/sync-engine'
-import { encodeKey, userReferenceColumns, userReferenceFields } 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'
@@ -213,15 +213,18 @@ export class PostgresDestination implements Destination {
       )
       if (r.rows.length > 0) out.push({ table: stream.name, rows: r.rows.length })
     }
-    const namedInData = (param: string) =>
-      userReferenceFields()
-        .map((f) => `"_raw"->'data'->>'${f}' = ${param}`)
-        .join(' or ')
+    // Each path as a Postgres array literal passed as a parameter, so a
+    // nested one works and no segment is interpolated into SQL.
+    const paths = userReferencePaths(streams).map(
+      (p) => `{${['data', ...p].map((step) => `"${step.replace(/(["\\])/g, '\\$1')}"`).join(',')}}`,
+    )
+    const namedInData = (id: string, from: number) =>
+      paths.map((_, i) => ` or "_raw" #>> $${from + i}::text[] = ${id}`).join('')
     // 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" = $1 and ("object_id" = $2 or ${namedInData('$2')}) returning 1`,
-      [env, userId],
+       where "_env" = $1 and ("object_id" = $2${namedInData('$2', 3)}) returning 1`,
+      [env, userId, ...paths],
     )
     if (ev.rows.length > 0) out.push({ table: 'events', rows: ev.rows.length })
     // Quarantined events embed the payload too, and they may be the only copy
@@ -229,8 +232,8 @@ export class PostgresDestination implements Destination {
     // in `events`.
     const rj = await this.client.query(
       `delete from ${quoteIdent(this.schema)}."_rejected"
-       where sync_id = $1 and env = $2 and ("object_id" = $3 or ${namedInData('$3')}) returning 1`,
-      [this.syncId, env, userId],
+       where sync_id = $1 and env = $2 and ("object_id" = $3${namedInData('$3', 4)}) returning 1`,
+      [this.syncId, env, userId, ...paths],
     )
     if (rj.rows.length > 0) out.push({ table: '_rejected', rows: rj.rows.length })
     return out

From 6e8c51e8c8fe363d7f1337cb33b79941149d447b Mon Sep 17 00:00:00 2001
From: "Garen J. Torikian" <gjtorikian@users.noreply.github.com>
Date: Fri, 6 Nov 2026 09:56:09 -0500
Subject: [PATCH 4/5] fix(engine): scrub archived events by every catalog path,
 whichever streams are selected

The archive keeps every event type the tail subscribed to, so a run
mirroring only `users` still archives `invitation.accepted` with the
subject's id in it. The payload paths `redact` scrubs by now come from
the whole catalog, with the selection adding only its extra columns.
---
 packages/sync-engine/src/catalog.ts | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/packages/sync-engine/src/catalog.ts b/packages/sync-engine/src/catalog.ts
index 9918275..32c2dbb 100644
--- a/packages/sync-engine/src/catalog.ts
+++ b/packages/sync-engine/src/catalog.ts
@@ -609,15 +609,18 @@ export function userReferenceColumns(stream: StreamDef): ColumnDef[] {
 }
 
 /**
- * The payload paths that name a user across `streams`, each as its steps
- * under the object: `['user_id']`, or `['metadata', 'owner_user_id']` for a
- * nested extra column. An event's `data` carries the object, so `redact`
- * scrubs archived and quarantined events by `data` followed by each.
+ * 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) {
+  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]

From ced612896807e5018b06436ceade2a22a2c8a9b6 Mon Sep 17 00:00:00 2001
From: "Garen J. Torikian" <gjtorikian@users.noreply.github.com>
Date: Fri, 6 Nov 2026 10:04:13 -0500
Subject: [PATCH 5/5] test(engine): expect the catalog's user paths alongside a
 selection's extra columns

---
 packages/sync-engine/test/catalog.test.ts | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/sync-engine/test/catalog.test.ts b/packages/sync-engine/test/catalog.test.ts
index 1dbf7ba..70fc2c0 100644
--- a/packages/sync-engine/test/catalog.test.ts
+++ b/packages/sync-engine/test/catalog.test.ts
@@ -62,10 +62,12 @@ describe('user references', () => {
     })
     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(['metadata.owner_user_id', 'user_id'])
+    ).toEqual(['accepted_user_id', 'inviter_user_id', 'metadata.owner_user_id', 'user_id'])
   })
 })