improvement(knowledge): finish Confluence passes past unreadable attachments, rank unauthenticated keyword search in one statement (#8975)

* improvement(knowledge): finish Confluence passes past unreadable attachments, rank unauthenticated keyword search in one statement

- Confluence: a 403/404 on one parent's attachment listing is skipped with a
  warning instead of recorded as a listing failure, so the pass can finish and
  advance lastSyncAt. The page still marks the listing reconciliation-unsafe,
  so attachments it did not list are kept, never deleted. Missing scopes and
  server failures stay listing failures.
- Connector counts: the member engine keeps the recorded document count on an
  unfinished slice once a pass has finished, the same gate the content engine
  uses. The count query and gate now live in sync-primitives.
- Keyword search: callers with no person behind them rank through the same
  materialized candidate statement signed-in readers use, instead of ORDER BY
  ts_rank_cd over a per-chunk join. Same ACL predicate and global ordering,
  ties now broken by chunk id; still unbudgeted.

* test(knowledge): compare single-statement keyword ranking against the two-step reference

Author: waleedlatif1

apps/sim/connectors/confluence/attachments.test.ts

@@ -215,41 +215,73 @@ describe('Confluence attachment listing', () => {
     expect(last.hasMore).toBe(false)
   })
 
-  it.each([403, 401])(
-    'preserves parent pages and cumulative partial progress for attachment access failure %s',
+  it('preserves parent pages and cumulative partial progress when the attachment scope is missing', async () => {
+    fetchMock.mockImplementation(async () =>
+      Response.json({ code: 401, message: 'Unauthorized; scope does not match' }, { status: 401 })
+    )
+    const listParents = vi.fn(
+      async (cursor: string | undefined): Promise<ExternalDocumentList> => ({
+        documents: [parent(cursor ? 'p2' : 'p1')],
+        hasMore: !cursor,
+        nextCursor: cursor ? undefined : 'next',
+      })
+    )
+    const first = await listConfluenceAttachments({ ...INPUT, listParents })
+    const context: Record<string, unknown> = {}
+    const last = await listConfluenceAttachments({
+      ...INPUT,
+      listParents,
+      cursor: first.nextCursor,
+      syncContext: context,
+    })
+    expect(first.documents.map((doc) => doc.externalId)).toEqual(['p1'])
+    expect(last.documents.map((doc) => doc.externalId)).toEqual(['p2'])
+    expect(last.listingFailures).toEqual({
+      count: 2,
+      samples: ['p1', 'p2'].map((scope) => ({
+        scope,
+        operation: 'confluence.attachments.list',
+        status: 401,
+        reasons: ['attachment_scope_mismatch'],
+      })),
+    })
+    expect(last.reconciliationSafe).toBe(false)
+    expect(context.reconciliationUnsafe).toBe(true)
+  })
+
+  /**
+   * One page whose attachments the credential cannot open must not leave every pass incomplete,
+   * which would keep the connector from ever finishing a sync. Its attachments were not listed,
+   * so the page still withholds deletion of anything it did not list.
+   */
+  it.each([403, 404])(
+    'skips a parent whose attachments answer %s without failing the listing or deleting what it did not list',
     async (status) => {
-      fetchMock.mockImplementation(async () =>
-        status === 401
-          ? Response.json({ code: 401, message: 'Unauthorized; scope does not match' }, { status })
-          : new Response(null, { status })
-      )
+      fetchMock.mockImplementation(async (input) => {
+        const parentId = new URL(String(input)).pathname.split('/').at(-2) ?? ''
+        if (parentId === 'p1') return new Response(null, { status })
+        return Response.json({ results: [file({ id: `${parentId}-file`, pageId: parentId })] })
+      })
       const listParents = vi.fn(
-        async (cursor: string | undefined): Promise<ExternalDocumentList> => ({
-          documents: [parent(cursor ? 'p2' : 'p1')],
-          hasMore: !cursor,
-          nextCursor: cursor ? undefined : 'next',
+        async (): Promise<ExternalDocumentList> => ({
+          documents: [parent('p1'), parent('p2')],
+          hasMore: false,
         })
       )
-      const first = await listConfluenceAttachments({ ...INPUT, listParents })
       const context: Record<string, unknown> = {}
-      const last = await listConfluenceAttachments({
+      const result = await listConfluenceAttachments({
         ...INPUT,
         listParents,
-        cursor: first.nextCursor,
         syncContext: context,
       })
-      expect(first.documents.map((doc) => doc.externalId)).toEqual(['p1'])
-      expect(last.documents.map((doc) => doc.externalId)).toEqual(['p2'])
-      expect(last.listingFailures).toEqual({
-        count: 2,
-        samples: ['p1', 'p2'].map((scope) => ({
-          scope,
-          operation: 'confluence.attachments.list',
-          status,
-          reasons: [status === 401 ? 'attachment_scope_mismatch' : 'attachment_access_denied'],
-        })),
-      })
-      expect(last.reconciliationSafe).toBe(false)
+      expect(result.documents.map((doc) => doc.externalId)).toEqual([
+        'p1',
+        'p2',
+        'attachment:page:p2:p2-file',
+      ])
+      expect(result.hasMore).toBe(false)
+      expect(result.listingFailures).toBeUndefined()
+      expect(result.reconciliationSafe).toBe(false)
       expect(context.reconciliationUnsafe).toBe(true)
     }
   )

apps/sim/connectors/confluence/attachments.ts

@@ -378,6 +378,7 @@ export async function listConfluenceAttachments(
 ): Promise<ExternalDocumentList> {
   const state = decodeCursor(input.cursor, input.syncContext?.totalDocsFetched)
   const documents: ExternalDocument[] = []
+  let skippedUnreadableParent = false
   if (state.parents.length === 0 && !state.parentsDone) {
     const parentContext = { ...input.syncContext, totalDocsFetched: state.parentsListed }
     const page = await input.listParents(state.parentCursor, parentContext)
@@ -413,15 +414,30 @@ export async function listConfluenceAttachments(
         })
       }
       if (state.serverFailureStreak >= MAX_CONSECUTIVE_SERVER_FAILURES) throw listing.error
-      state.failures ??= { count: 0, samples: [] }
-      state.failures.count += 1
-      if (state.failures.samples.length < MAX_LISTING_FAILURE_SAMPLES) {
-        state.failures.samples.push({
-          scope: parent.id,
-          operation: 'confluence.attachments.list',
+      if (listing.reason === 'attachment_access_denied') {
+        /**
+         * The parent was listed a moment ago, so a 403 or 404 on its attachments is that one
+         * item's restriction, not a scope the source failed to list: skipping it leaves the pass
+         * complete. Its unlisted attachments are not evidence of deletion, so the page still
+         * marks the listing unsafe and reconciliation keeps them.
+         */
+        logger.warn('Skipping a Confluence parent whose attachments are not readable', {
+          parentId: parent.id,
+          parentType: parent.type,
           status: listing.status,
-          reasons: [listing.reason],
         })
+        skippedUnreadableParent = true
+      } else {
+        state.failures ??= { count: 0, samples: [] }
+        state.failures.count += 1
+        if (state.failures.samples.length < MAX_LISTING_FAILURE_SAMPLES) {
+          state.failures.samples.push({
+            scope: parent.id,
+            operation: 'confluence.attachments.list',
+            status: listing.status,
+            reasons: [listing.reason],
+          })
+        }
       }
       state.parents.shift()
       state.attachmentCursor = undefined
@@ -453,7 +469,8 @@ export async function listConfluenceAttachments(
     state.attachmentCursor = nextCursor
     if (!nextCursor) state.parents.shift()
   }
-  if (state.failures && input.syncContext) input.syncContext.reconciliationUnsafe = true
+  const reconciliationUnsafe = Boolean(state.failures) || skippedUnreadableParent
+  if (reconciliationUnsafe && input.syncContext) input.syncContext.reconciliationUnsafe = true
   const hasMore = state.parents.length > 0 || !state.parentsDone
   const nextCursor = hasMore ? CURSOR_PREFIX + JSON.stringify(state) : undefined
   if (nextCursor && Buffer.byteLength(nextCursor, 'utf8') > MAX_CURSOR_BYTES) {
@@ -463,7 +480,8 @@ export async function listConfluenceAttachments(
     documents,
     hasMore,
     nextCursor,
-    ...(state.failures ? { listingFailures: state.failures, reconciliationSafe: false } : {}),
+    ...(state.failures ? { listingFailures: state.failures } : {}),
+    ...(reconciliationUnsafe ? { reconciliationSafe: false } : {}),
   }
 }