fix(knowledge): refresh unchanged mirrored ACLs without firing the projection fan-out

.github/workflows/test-build.yml

@@ -297,7 +297,8 @@ jobs:
           bunx vitest run --mode integration lib/knowledge/access/group-membership.integration.ts
           bunx vitest run \
             lib/knowledge/access/predicate.postgres.test.ts \
-            lib/knowledge/connectors/external-directory.postgres.test.ts
+            lib/knowledge/connectors/external-directory.postgres.test.ts \
+            lib/knowledge/connectors/sync-persistence.postgres.test.ts
 
   test-build:
     name: Lint and Test

apps/sim/lib/knowledge/connectors/sync-persistence.postgres.test.ts

@@ -0,0 +1,235 @@
+/**
+ * @vitest-environment node
+ */
+import { installProjectionSourceAcl } from '@sim/db/script-migrations/0021_embedding_search_connector'
+import { generateId } from '@sim/utils/id'
+import postgres, { type Sql } from 'postgres'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.unmock('drizzle-orm')
+vi.unmock('@sim/db/schema')
+vi.mock('@/lib/knowledge/documents/service', () => ({ hardDeleteDocuments: vi.fn() }))
+vi.mock('@/lib/uploads', () => ({ StorageService: {} }))
+vi.mock('@/lib/uploads/core/storage-service', () => ({ deleteFile: vi.fn() }))
+vi.mock('@/lib/uploads/server/metadata', () => ({ deleteFileMetadata: vi.fn() }))
+vi.mock('@/connectors/registry.server', () => ({ CONNECTOR_REGISTRY: {} }))
+
+const { drizzle } = await import('drizzle-orm/postgres-js')
+const schema = await import('@sim/db/schema')
+const { persistDocumentAcls } = await import('@/lib/knowledge/connectors/sync-persistence')
+
+const databaseUrl = process.env.KNOWLEDGE_ACL_TEST_DATABASE_URL
+
+const ALICE = 'u:alice@corp.com'
+const BOB = 'u:bob@corp.com'
+
+/**
+ * The document and projections carry only the columns the ACL write and the projection trigger
+ * touch. A projection row whose `acl` is NULL is one the backfill has not filled yet: the trigger
+ * rewrites it on any ACL assignment, because NULL is distinct from every ACL.
+ */
+describe.runIf(Boolean(databaseUrl))('persistDocumentAcls in PostgreSQL', () => {
+  let admin: Sql
+  let sql: Sql
+  const schemaName = `acl_write_${generateId().replaceAll('-', '')}`
+
+  const projected = () =>
+    sql<{ id: string; acl: string[] | null }[]>`
+      SELECT id, acl FROM embedding_search
+      UNION ALL SELECT id, acl FROM embedding_keyword_tin ORDER BY id`
+
+  const persist = (acls: Map<string, string[]>) =>
+    persistDocumentAcls('admin', acls, drizzle(sql, { schema }))
+
+  beforeAll(async () => {
+    const url = new URL(databaseUrl!)
+    if (
+      !['localhost', '127.0.0.1'].includes(url.hostname) ||
+      !url.pathname.startsWith('/sim_acl_test')
+    ) {
+      throw new Error('ACL write tests require a disposable local integration database')
+    }
+    admin = postgres(url.toString(), { max: 1, onnotice: () => undefined })
+    await admin.unsafe(`CREATE SCHEMA "${schemaName}"`)
+    sql = postgres(url.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+      connection: { search_path: schemaName },
+    })
+    await sql`CREATE TABLE document (
+      id text PRIMARY KEY, external_id text, connector_id text,
+      acl text[] NOT NULL DEFAULT '{ws}', acl_requirements jsonb NOT NULL DEFAULT '[]',
+      acl_verified_at timestamp
+    )`
+    for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
+      await sql`CREATE TABLE ${sql(projection)} (
+        id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true,
+        connector_id text, acl text[]
+      )`
+    }
+    await installProjectionSourceAcl(sql)
+    await sql`ALTER TABLE embedding_search DISABLE TRIGGER embedding_search_source_acl_set`
+    await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set`
+  }, 60_000)
+
+  afterAll(async () => {
+    await sql?.end()
+    await admin?.unsafe(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`)
+    await admin?.end()
+  })
+
+  beforeEach(async () => {
+    await sql`TRUNCATE embedding_search, embedding_keyword_tin, document`
+    await sql`INSERT INTO document (id, external_id, connector_id, acl, acl_verified_at) VALUES
+      ('doc-same', 'file-same', 'admin', ARRAY[${ALICE}], now() - interval '1 day'),
+      ('doc-moved', 'file-moved', 'admin', ARRAY[${ALICE}], now() - interval '1 day')`
+    for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
+      const prefix = projection === 'embedding_search' ? 'vec' : 'kw'
+      await sql`INSERT INTO ${sql(projection)} (id, document_id, connector_id, acl) VALUES
+        (${`${prefix}-same-unfilled`}, 'doc-same', NULL, NULL),
+        (${`${prefix}-same-filled`}, 'doc-same', 'admin', ARRAY[${ALICE}]),
+        (${`${prefix}-moved-unfilled`}, 'doc-moved', NULL, NULL),
+        (${`${prefix}-moved-filled`}, 'doc-moved', 'admin', ARRAY[${ALICE}])`
+    }
+  })
+
+  it('refreshes the evidence of an unchanged ACL without rewriting any chunk projection row', async () => {
+    await expect(persist(new Map([['file-same', [ALICE]]]))).resolves.toEqual({
+      updated: 1,
+      rejected: 0,
+    })
+
+    const [stored] = await sql<{ acl: string[]; fresh: boolean }[]>`
+      SELECT acl, acl_verified_at > now() AT TIME ZONE 'UTC' - interval '1 minute' AS fresh
+      FROM document WHERE id = 'doc-same'`
+    expect(stored).toEqual({ acl: [ALICE], fresh: true })
+    expect((await projected()).filter((row) => row.id.includes('-same-'))).toEqual([
+      { id: 'kw-same-filled', acl: [ALICE] },
+      { id: 'kw-same-unfilled', acl: null },
+      { id: 'vec-same-filled', acl: [ALICE] },
+      { id: 'vec-same-unfilled', acl: null },
+    ])
+  })
+
+  it('propagates a changed ACL to every chunk projection row, filled or not', async () => {
+    await expect(persist(new Map([['file-moved', [BOB]]]))).resolves.toEqual({
+      updated: 1,
+      rejected: 0,
+    })
+
+    const [stored] = await sql<{ acl: string[] }[]>`SELECT acl FROM document WHERE id = 'doc-moved'`
+    expect(stored.acl).toEqual([BOB])
+    expect((await projected()).filter((row) => row.id.includes('-moved-'))).toEqual([
+      { id: 'kw-moved-filled', acl: [BOB] },
+      { id: 'kw-moved-unfilled', acl: [BOB] },
+      { id: 'vec-moved-filled', acl: [BOB] },
+      { id: 'vec-moved-unfilled', acl: [BOB] },
+    ])
+  })
+
+  it('treats a changed restriction under the same primary ACL as a change', async () => {
+    await persistDocumentAcls(
+      'admin',
+      new Map([['file-same', { acl: [ALICE], requirements: [['g:confluence:tenant:space']] }]]),
+      drizzle(sql, { schema })
+    )
+
+    const [stored] = await sql<{ requirements: string[][] }[]>`
+      SELECT acl_requirements AS requirements FROM document WHERE id = 'doc-same'`
+    expect(stored.requirements).toEqual([[ALICE], ['g:confluence:tenant:space']])
+  })
+
+  it('writes each document once when one page mixes unchanged and changed ACLs', async () => {
+    await expect(
+      persist(
+        new Map([
+          ['file-same', [ALICE]],
+          ['file-moved', [ALICE, BOB]],
+        ])
+      )
+    ).resolves.toEqual({ updated: 2, rejected: 0 })
+
+    const rows = await sql<
+      { id: string; acl: string[] }[]
+    >`SELECT id, acl FROM document ORDER BY id`
+    expect(rows).toEqual([
+      { id: 'doc-moved', acl: [ALICE, BOB] },
+      { id: 'doc-same', acl: [ALICE] },
+    ])
+    expect(
+      (await projected()).filter((row) => row.id.endsWith('-unfilled')).map((row) => row.acl)
+    ).toEqual([[ALICE, BOB], null, [ALICE, BOB], null])
+  })
+  it('writes a changed ACL group larger than one change batch completely', async () => {
+    const ids = Array.from(
+      { length: 60 },
+      (_unused, index) => `bulk-${String(index).padStart(2, '0')}`
+    )
+    await sql`INSERT INTO document ${sql(
+      ids.map((id) => ({ id: `doc-${id}`, external_id: id, connector_id: 'admin', acl: [ALICE] }))
+    )}`
+    await sql`INSERT INTO embedding_search ${sql(
+      ids.map((id) => ({
+        id: `vec-${id}`,
+        document_id: `doc-${id}`,
+        connector_id: 'admin',
+        acl: [ALICE],
+      }))
+    )}`
+
+    await expect(persist(new Map(ids.map((id) => [id, [BOB]])))).resolves.toEqual({
+      updated: ids.length,
+      rejected: 0,
+    })
+
+    const documents = await sql<{ acl: string[] }[]>`
+      SELECT acl FROM document WHERE id LIKE 'doc-bulk-%'`
+    expect(documents).toHaveLength(ids.length)
+    expect(documents.every((row) => row.acl.join() === BOB)).toBe(true)
+    const chunks = await sql<{ acl: string[] }[]>`
+      SELECT acl FROM embedding_search WHERE id LIKE 'vec-bulk-%'`
+    expect(chunks).toHaveLength(ids.length)
+    expect(chunks.every((row) => row.acl.join() === BOB)).toBe(true)
+  })
+
+  it.each([
+    { name: 'verified within this generation', verifiedOffsetMs: 1_000, preserved: true },
+    { name: 'verified before this generation', verifiedOffsetMs: -1_000, preserved: false },
+    { name: 'never verified', verifiedOffsetMs: null, preserved: false },
+  ])(
+    'applies the unresolved-evidence guard to an ACL the source could not answer: $name',
+    async ({ verifiedOffsetMs, preserved }) => {
+      const [clock] = await sql<{ now: string }[]>`
+        SELECT (now() AT TIME ZONE 'UTC')::text AS now`
+      const generationStartedAt = new Date(`${clock.now}Z`)
+      generationStartedAt.setTime(generationStartedAt.getTime() - 60_000)
+      const verifiedAt =
+        verifiedOffsetMs === null
+          ? null
+          : new Date(generationStartedAt.getTime() + verifiedOffsetMs).toISOString()
+      await sql`UPDATE document SET acl_verified_at = ${verifiedAt}::timestamptz AT TIME ZONE 'UTC'
+        WHERE id = 'doc-same'`
+
+      const result = await persistDocumentAcls(
+        'admin',
+        new Map([['file-same', []]]),
+        drizzle(sql, { schema }),
+        {
+          unresolvedExternalIds: new Set(['file-same']),
+          generationStartedAt,
+        }
+      )
+
+      expect(result).toEqual({ updated: preserved ? 0 : 1, rejected: 0 })
+      const [stored] = await sql<{ acl: string[]; verified: boolean }[]>`
+        SELECT acl, acl_verified_at IS NOT NULL AS verified FROM document WHERE id = 'doc-same'`
+      expect(stored).toEqual(
+        preserved ? { acl: [ALICE], verified: true } : { acl: [], verified: false }
+      )
+      const filled = (await projected())
+        .filter((row) => row.id.endsWith('-same-filled'))
+        .map((row) => row.acl)
+      expect(filled).toEqual(preserved ? [[ALICE], [ALICE]] : [[], []])
+    }
+  )
+})

apps/sim/lib/knowledge/connectors/sync-persistence.test.ts

@@ -71,13 +71,13 @@ describe('persistDocumentAcls', () => {
    * corpus every time somebody joined a group.
    */
   it('refreshes only access fields, so no document is re-embedded', async () => {
-    queueUpdatedCounts(1)
+    queueUpdatedCounts(0, 1)
 
     await persistDocumentAcls(CONNECTOR, new Map([['file-1', ['u:alice@corp.com']]]))
 
     expect(dbChainMockFns.update).toHaveBeenCalledWith(schemaMock.document)
-    expect(dbChainMockFns.set).toHaveBeenCalledTimes(1)
-    expect(dbChainMockFns.set).toHaveBeenCalledWith({
+    expect(dbChainMockFns.set).toHaveBeenCalledTimes(2)
+    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
       acl: ['u:alice@corp.com'],
       aclRequirements: [],
       aclVerifiedAt: expect.objectContaining({
@@ -87,6 +87,25 @@ describe('persistDocumentAcls', () => {
     })
   })
 
+  /**
+   * Assigning `acl` fires the projection trigger, which rewrites every chunk row whose copy
+   * differs, so a document whose ACL did not change must only have its evidence refreshed.
+   */
+  it('refreshes the evidence of an unchanged ACL without assigning it', async () => {
+    queueUpdatedCounts(1, 0)
+
+    await expect(
+      persistDocumentAcls(CONNECTOR, new Map([['file-1', ['u:alice@corp.com']]]))
+    ).resolves.toEqual({ updated: 1, rejected: 0 })
+
+    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, {
+      aclVerifiedAt: expect.objectContaining({
+        strings: ["statement_timestamp() AT TIME ZONE 'UTC'"],
+        values: [],
+      }),
+    })
+  })
+
   it('reports how many documents received current permission evidence', async () => {
     queueUpdatedCounts(2)
 
@@ -105,8 +124,8 @@ describe('persistDocumentAcls', () => {
    * Files under one folder overwhelmingly share an ACL, so grouping is what
    * keeps a crawl of thousands to a handful of statements.
    */
-  it('writes one statement per distinct ACL, not per document', async () => {
-    queueUpdatedCounts(2, 1)
+  it('writes one refresh and one change statement per distinct ACL, not per document', async () => {
+    queueUpdatedCounts(0, 2, 0, 1)
 
     await persistDocumentAcls(
       CONNECTOR,
@@ -117,16 +136,16 @@ describe('persistDocumentAcls', () => {
       ])
     )
 
-    expect(dbChainMockFns.set).toHaveBeenCalledTimes(2)
-    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, {
+    expect(dbChainMockFns.set).toHaveBeenCalledTimes(4)
+    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
       acl: ['u:alice@corp.com'],
       aclRequirements: [],
       aclVerifiedAt: expect.objectContaining({
         strings: ["statement_timestamp() AT TIME ZONE 'UTC'"],
         values: [],
       }),
     })
-    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
+    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(4, {
       acl: ['u:bob@corp.com'],
       aclRequirements: [],
       aclVerifiedAt: expect.objectContaining({
@@ -147,8 +166,8 @@ describe('persistDocumentAcls', () => {
       ])
     )
 
-    expect(dbChainMockFns.set).toHaveBeenCalledTimes(1)
-    expect(dbChainMockFns.set).toHaveBeenCalledWith({
+    expect(dbChainMockFns.set).toHaveBeenCalledTimes(2)
+    expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
       acl: ['u:alice@corp.com', 'u:bob@corp.com'],
       aclRequirements: [],
       aclVerifiedAt: expect.objectContaining({
@@ -174,24 +193,24 @@ describe('persistDocumentAcls', () => {
     })
 
     it('retains an empty restriction and separately persists different clauses', async () => {
-      queueUpdatedCounts(1, 1)
+      queueUpdatedCounts(0, 1, 0, 1)
       await persistDocumentAcls(
         CONNECTOR,
         new Map([
           ['file-1', { acl: ['u:alice@corp.com'], requirements: [[]] }],
           ['file-2', { acl: ['u:alice@corp.com'], requirements: [['g:confluence:site:team']] }],
         ])
       )
-      expect(dbChainMockFns.set).toHaveBeenCalledTimes(2)
-      expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, {
+      expect(dbChainMockFns.set).toHaveBeenCalledTimes(4)
+      expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
         acl: ['u:alice@corp.com'],
         aclRequirements: [['u:alice@corp.com'], []],
         aclVerifiedAt: expect.objectContaining({
           strings: ["statement_timestamp() AT TIME ZONE 'UTC'"],
           values: [],
         }),
       })
-      expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
+      expect(dbChainMockFns.set).toHaveBeenNthCalledWith(4, {
         acl: ['u:alice@corp.com'],
         aclRequirements: [['u:alice@corp.com'], ['g:confluence:site:team']],
         aclVerifiedAt: expect.objectContaining({