diff --git a/packages/sync-clickhouse/src/destination.ts b/packages/sync-clickhouse/src/destination.ts
index 78b2a3e..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 } 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
diff --git a/packages/sync-engine/src/catalog.ts b/packages/sync-engine/src/catalog.ts
index 5998af5..32c2dbb 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,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)
 
 /**
diff --git a/packages/sync-engine/src/index.ts b/packages/sync-engine/src/index.ts
index 8d8c173..db6c82d 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,
+  userReferencePaths,
+} 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..70fc2c0 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,
+  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'])
+  })
+})
diff --git a/packages/sync-mongodb/src/destination.ts b/packages/sync-mongodb/src/destination.ts
index ac3d9e2..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 } 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,15 +240,19 @@ 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
@@ -256,7 +260,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/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 4cf86c1..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 } 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,20 +205,26 @@ 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
@@ -226,8 +232,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 = ?${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..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 } 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'
@@ -204,20 +204,27 @@ 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 = userReferenceColumns(stream)
+      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 })
     }
+    // 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 "_raw"->'data'->>'user_id' = $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
@@ -225,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 "_raw"->'data'->>'user_id' = $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
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)