diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml index b7d6b0d6dc5..d2c1834bb86 100644 --- a/.github/workflows/test-build.yml +++ b/.github/workflows/test-build.yml @@ -271,6 +271,7 @@ jobs: lib/knowledge/__integration__/connector-deferral.integration.ts lib/knowledge/__integration__/stored-document-recovery.integration.ts lib/knowledge/__integration__/connector-partition-work.integration.ts + lib/knowledge/__integration__/user-document-visibility.integration.ts lib/knowledge/__integration__/listing-continuation.integration.ts lib/knowledge/__integration__/member-scope-renewal.integration.ts lib/knowledge/__integration__/slack-empty-threads.integration.ts diff --git a/apps/sim/connectors/google-workspace/company-crawl.test.ts b/apps/sim/connectors/google-workspace/company-crawl.test.ts index 40125bce020..cf4f7198a49 100644 --- a/apps/sim/connectors/google-workspace/company-crawl.test.ts +++ b/apps/sim/connectors/google-workspace/company-crawl.test.ts @@ -2,9 +2,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { GoogleApiError, readGoogleApiError } from '@/connectors/google-workspace/api-errors' import { + GoogleWorkspaceMailboxNotSetup, getGoogleWorkspaceDocument, InvalidGoogleWorkspaceCursor, listGoogleWorkspaceDocuments, + serviceNotEnabledFailure, validateGoogleWorkspaceConfig, } from '@/connectors/google-workspace/company-crawl' import type { ConnectorConfig, ExternalDocument } from '@/connectors/types' @@ -483,18 +485,18 @@ describe('Google Workspace per-user central crawl', () => { } ) - it('skips an explicitly unprovisioned Gmail mailbox before requesting a token', async () => { + it('leaves an explicitly unprovisioned Gmail mailbox to the scheduler before requesting a token', async () => { directory([USER('alice', undefined, { isMailboxSetup: false }), USER('bob')]) - const ctx = context() - const first = await list(ctx) - expect(first.listingFailures?.samples[0]).toEqual({ - scope: 'alice@corp.com', + const ctx: Record = context() + const error = await list(ctx).catch((caught: unknown) => caught) + expect(error).toBeInstanceOf(GoogleWorkspaceMailboxNotSetup) + expect(serviceNotEnabledFailure(error)).toEqual({ operation: 'directory.users.get', reasons: ['mailboxNotSetup'], }) + expect(ctx.reconciliationUnsafe).toBeUndefined() expect(ctx.getDelegatedAccessToken).not.toHaveBeenCalled() - const second = await list(context(), first.nextCursor) - expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) + expect(listUserDocuments).not.toHaveBeenCalled() }) it('does not use Gmail mailbox eligibility for Calendar', async () => { @@ -502,18 +504,18 @@ describe('Google Workspace per-user central crawl', () => { expect((await list(context(), undefined, CONFIG, 'google_calendar')).documents).toHaveLength(1) }) - it.each(['forbidden', 'notACalendarUser'])( - 'isolates explicit Calendar list access failures (%s) without claiming a disabled service', - async (reason) => { + it.each([{ reasons: ['forbidden'] }, { reasons: ['forbidden', 'notACalendarUser'] }])( + 'isolates explicit Calendar list access failures ($reasons) without claiming a disabled service', + async ({ reasons }) => { listUserDocuments.mockRejectedValueOnce( - new GoogleApiError('calendar.events.list', 403, [reason]) + new GoogleApiError('calendar.events.list', 403, reasons) ) const first = await list(context(), undefined, CONFIG, 'google_calendar') expect(first.listingFailures?.samples[0]).toEqual({ scope: 'alice@corp.com', operation: 'calendar.events.list', status: 403, - reasons: [reason], + reasons, }) const second = await list(context(), first.nextCursor, CONFIG, 'google_calendar') expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) @@ -521,6 +523,14 @@ describe('Google Workspace per-user central crawl', () => { } ) + it('leaves a user without the Calendar service to the scheduler instead of recording a failure', async () => { + const error = new GoogleApiError('calendar.events.list', 403, ['notACalendarUser']) + listUserDocuments.mockRejectedValueOnce(error) + const ctx: Record = context() + await expect(list(ctx, undefined, CONFIG, 'google_calendar')).rejects.toBe(error) + expect(ctx.reconciliationUnsafe).toBeUndefined() + }) + it.each([{ error: { code: 403 } }, { error: { code: 403, errors: [], details: [] } }])( 'propagates a Calendar 403 without reason codes: %j', async (body) => { @@ -626,10 +636,9 @@ describe('Google Workspace per-user central crawl', () => { }) it('bounds retained failure samples while counting every unavailable user', async () => { - directory( - Array.from({ length: 15 }, (_, index) => - USER(`user-${index}`, undefined, { isMailboxSetup: false }) - ) + directory(Array.from({ length: 15 }, (_, index) => USER(`user-${index}`))) + listUserDocuments.mockRejectedValue( + new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) ) let cursor: string | undefined let final @@ -643,7 +652,7 @@ describe('Google Workspace per-user central crawl', () => { reconciliationSafe: false, }) expect(final?.listingFailures?.samples).toHaveLength(10) - expect(listUserDocuments).not.toHaveBeenCalled() + expect(listUserDocuments).toHaveBeenCalledTimes(15) }) }) diff --git a/apps/sim/connectors/google-workspace/company-crawl.ts b/apps/sim/connectors/google-workspace/company-crawl.ts index 241828d10ee..7bcbca3722e 100644 --- a/apps/sim/connectors/google-workspace/company-crawl.ts +++ b/apps/sim/connectors/google-workspace/company-crawl.ts @@ -161,7 +161,35 @@ function ownerDocument(document: ExternalDocument, access: DelegatedUser): Exter return { ...document, acl: [`u:${access.user.email}`] } } -/** Isolates narrow user-list failures; delegation, known scope errors and quota errors still fail. */ +/** A Directory user without a Gmail mailbox; the user scheduler decides whether to skip them. */ +export class GoogleWorkspaceMailboxNotSetup extends Error { + constructor() { + super('Google Workspace user has no Gmail mailbox') + this.name = 'GoogleWorkspaceMailboxNotSetup' + } +} + +/** + * Evidence that a user lacks the service itself: no Gmail mailbox, or a Calendar list 403 whose + * only reason is `notACalendarUser`. Admin changes to a user's services can take up to a day to + * settle, so the user scheduler, which can see indexed documents, decides when this is a skip. + */ +export function serviceNotEnabledFailure( + error: unknown +): Omit | null { + if (error instanceof GoogleWorkspaceMailboxNotSetup) + return { operation: 'directory.users.get', reasons: ['mailboxNotSetup'] } + return error instanceof GoogleApiError && + error.reasonsComplete && + error.status === 403 && + error.diagnostic?.operation === 'calendar.events.list' && + error.diagnostic.reasons.length === 1 && + error.diagnostic.reasons[0] === 'notACalendarUser' + ? { operation: 'calendar.events.list', status: 403, reasons: ['notACalendarUser'] } + : null +} + +/** Isolates narrow user-list failures; a missing Calendar service propagates to the scheduler. */ function userListingFailure( error: unknown, provider: GoogleWorkspaceProvider @@ -169,7 +197,8 @@ function userListingFailure( if (!(error instanceof GoogleApiError) || !error.diagnostic || !error.reasonsComplete) return null const reasons = error.diagnostic.reasons const isolated = - provider === 'gmail' + !serviceNotEnabledFailure(error) && + (provider === 'gmail' ? error.diagnostic.operation === 'gmail.threads.list' && error.status === 400 && reasons.length > 0 && @@ -177,7 +206,7 @@ function userListingFailure( : error.diagnostic.operation === 'calendar.events.list' && error.status === 403 && reasons.length > 0 && - reasons.every((reason) => reason === 'forbidden' || reason === 'notACalendarUser') + reasons.every((reason) => reason === 'forbidden' || reason === 'notACalendarUser')) return isolated ? { operation: error.diagnostic.operation, status: error.status, reasons: [...reasons] } : null @@ -317,9 +346,8 @@ export async function listGoogleWorkspaceDocuments( }) return emptyPage(advance()) } - if (provider === 'gmail' && user.isMailboxSetup === false) { - return failedUser({ operation: 'directory.users.get', reasons: ['mailboxNotSetup'] }) - } + if (provider === 'gmail' && user.isMailboxSetup === false) + throw new GoogleWorkspaceMailboxNotSetup() const access: PageAccess = { provider, user, diff --git a/apps/sim/connectors/listing-failures.ts b/apps/sim/connectors/listing-failures.ts index 22dedb2e9d1..1c3e546bc66 100644 --- a/apps/sim/connectors/listing-failures.ts +++ b/apps/sim/connectors/listing-failures.ts @@ -4,17 +4,17 @@ import { CONNECTOR_SOURCE_REASON_STATES } from '@/connectors/source-error' export const MAX_LISTING_FAILURE_SAMPLES = 10 /** Persist only bounded scope identifiers and provider-owned codes, never raw errors or content. */ +export const listingFailureSampleSchema = z.object({ + scope: z.string().min(1).max(254), + operation: z.string().min(1).max(96), + status: z.number().int().min(100).max(599).optional(), + reasons: z.array(z.string().min(1).max(64)).max(16), + reasonState: z.enum(CONNECTOR_SOURCE_REASON_STATES).optional(), + /** When a standing account condition was first observed, bounding how long it is retained. */ + since: z.string().datetime().optional(), +}) + export const listingFailuresSchema = z.object({ count: z.number().int().positive().max(Number.MAX_SAFE_INTEGER), - samples: z - .array( - z.object({ - scope: z.string().min(1).max(254), - operation: z.string().min(1).max(96), - status: z.number().int().min(100).max(599).optional(), - reasons: z.array(z.string().min(1).max(64)).max(16), - reasonState: z.enum(CONNECTOR_SOURCE_REASON_STATES).optional(), - }) - ) - .max(MAX_LISTING_FAILURE_SAMPLES), + samples: z.array(listingFailureSampleSchema).max(MAX_LISTING_FAILURE_SAMPLES), }) diff --git a/apps/sim/connectors/types.ts b/apps/sim/connectors/types.ts index 2ea305bc5f3..b5493524cd4 100644 --- a/apps/sim/connectors/types.ts +++ b/apps/sim/connectors/types.ts @@ -205,6 +205,8 @@ export interface ExternalListingFailures { status?: number reasons: string[] reasonState?: ConnectorSourceReasonState + /** When a standing account condition was first observed, bounding how long it is retained. */ + since?: string }[] } diff --git a/apps/sim/lib/knowledge/__integration__/user-document-visibility.integration.ts b/apps/sim/lib/knowledge/__integration__/user-document-visibility.integration.ts new file mode 100644 index 00000000000..4e551f06cfb --- /dev/null +++ b/apps/sim/lib/knowledge/__integration__/user-document-visibility.integration.ts @@ -0,0 +1,90 @@ +/** Real PostgreSQL coverage for the probe that keeps a user's visible documents from being skipped. */ +import { db } from '@sim/db' +import { document, knowledgeConnector, organization, user, workspace } from '@sim/db/schema' +import { generateId } from '@sim/utils/id' +import { eq, inArray } from 'drizzle-orm' +import { afterAll, afterEach, beforeEach, describe, expect, it } from 'vitest' +import { seedKnowledgeAclFixture } from '@/lib/knowledge/__integration__/seed-source-access-fixture' +import { hasVisibleUserDocuments } from '@/lib/knowledge/connectors/user-document-visibility' + +const DAY_MS = 24 * 60 * 60 * 1000 + +describe('Visible user documents in PostgreSQL', () => { + let owner: Awaited> + let otherConnectorId: string + const email = () => `${owner.aliceId}@fixture.test` + + beforeEach(async () => { + owner = await seedKnowledgeAclFixture(undefined, { connectorType: 'google_drive' }) + otherConnectorId = generateId() + await db.insert(knowledgeConnector).values({ + id: otherConnectorId, + knowledgeBaseId: owner.knowledgeBaseId, + connectorType: 'google_drive', + sourceConfig: {}, + accessMode: 'admin', + status: 'active', + credentialId: owner.credentialId, + }) + }) + afterEach(async () => { + await db.delete(workspace).where(eq(workspace.id, owner.workspaceId)) + await db.delete(organization).where(eq(organization.id, owner.organizationId)) + await db.delete(user).where(inArray(user.id, [owner.aliceId, owner.bobId])) + }) + afterAll(() => db.$client.end()) + + const insert = (overrides: Partial = {}) => + db.insert(document).values({ + id: generateId(), + knowledgeBaseId: owner.knowledgeBaseId, + connectorId: owner.connectorId, + externalId: generateId(), + filename: 'Event.txt', + mimeType: 'text/plain', + fileUrl: '', + fileSize: 0, + acl: [`u:${email()}`], + aclVerifiedAt: new Date(), + ...overrides, + }) + + it('finds a live, included document granted to the user with fresh permission evidence', async () => { + await insert() + expect(await hasVisibleUserDocuments(owner.connectorId, email())).toBe(true) + }) + + it('matches a mixed-case, padded directory email to the normalized ACL token', async () => { + await insert() + expect(await hasVisibleUserDocuments(owner.connectorId, ` ${email().toUpperCase()} `)).toBe( + true + ) + }) + + it.each([ + { + label: 'permission evidence older than the freshness limit', + overrides: () => ({ aclVerifiedAt: new Date(Date.now() - DAY_MS - 60_000) }), + }, + { label: 'no permission evidence', overrides: () => ({ aclVerifiedAt: null }) }, + { + label: 'a grant to a different user', + overrides: () => ({ acl: [`u:${owner.bobId}@fixture.test`] }), + }, + { + label: 'a document in another connector', + overrides: () => ({ connectorId: otherConnectorId }), + }, + { label: 'a user-excluded document', overrides: () => ({ userExcluded: true }) }, + { label: 'an archived document', overrides: () => ({ archivedAt: new Date() }) }, + { label: 'a deleted document', overrides: () => ({ deletedAt: new Date() }) }, + ])('ignores $label', async ({ overrides }) => { + await insert(overrides()) + expect(await hasVisibleUserDocuments(owner.connectorId, email())).toBe(false) + }) + + it('refuses an address that cannot be a user token', async () => { + await insert() + expect(await hasVisibleUserDocuments(owner.connectorId, ' ')).toBe(false) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/google-company-scheduler.test.ts b/apps/sim/lib/knowledge/connectors/google-company-scheduler.test.ts index 822beb3c60d..ae8a64894f6 100644 --- a/apps/sim/lib/knowledge/connectors/google-company-scheduler.test.ts +++ b/apps/sim/lib/knowledge/connectors/google-company-scheduler.test.ts @@ -15,7 +15,10 @@ import type { } from '@/lib/knowledge/connectors/partition-work' import { GoogleDriveApiError } from '@/connectors/google-drive/google-drive-errors' import { GoogleApiError } from '@/connectors/google-workspace/api-errors' -import { listGoogleWorkspaceDocuments } from '@/connectors/google-workspace/company-crawl' +import { + GoogleWorkspaceMailboxNotSetup, + listGoogleWorkspaceDocuments, +} from '@/connectors/google-workspace/company-crawl' import { googleCompanyUserContextSchema } from '@/connectors/google-workspace/company-work' import type { GoogleWorkspaceUser } from '@/connectors/google-workspace/users' import { ConnectorSourceError } from '@/connectors/source-error' @@ -36,8 +39,8 @@ const document: ExternalDocument = { contentHash: 'hash', mimeType: 'text/plain', } -function user(id: string): GoogleWorkspaceUser { - return { id, email: `${id}@fixture.test`, customerId: 'customer', active: true } +function user(id: string, extra: Partial = {}): GoogleWorkspaceUser { + return { id, email: `${id}@fixture.test`, customerId: 'customer', active: true, ...extra } } interface FakeWork extends ConnectorPartitionWorkItem { complete: boolean @@ -142,10 +145,12 @@ function fixture(provider = 'google_calendar', syncIntervalMinutes = 60) { documents: [document], hasMore: false, })) + const visible = vi.fn(async (_user: GoogleWorkspaceUser) => false) let scheduler = createGoogleCompanyScheduler({ provider, store, listDocuments: list, + hasVisibleDocuments: visible, syncIntervalMinutes, isListingCursorInvalidError: (error) => error === expiredCursor, now: () => clock, @@ -182,6 +187,7 @@ function fixture(provider = 'google_calendar', syncIntervalMinutes = 60) { return { rows, list, + visible, process, step, saved: () => saved, @@ -196,6 +202,7 @@ function fixture(provider = 'google_calendar', syncIntervalMinutes = 60) { provider, store, listDocuments: list, + hasVisibleDocuments: visible, syncIntervalMinutes, isListingCursorInvalidError: (error) => error === expiredCursor, now: () => clock, @@ -227,18 +234,19 @@ describe('durable Google company user scheduling', () => { }) it.each([ - ['google_calendar', []], - ['google_calendar', ['notACalendarUser']], - ['google_drive', []], + ['google_calendar', [], false], + ['google_calendar', ['forbidden'], true], + ['google_calendar', ['notACalendarUser'], false], + ['google_drive', [], false], ] as const)( - 'retains a failed %s user (%j) and continues other users', - async (provider, reasons) => { + 'retains a failed %s user (%j, reasons complete: %s) and continues other users', + async (provider, reasons, reasonsComplete) => { mocks.directory.mockResolvedValue({ users: [user('a'), user('z')] }) const f = fixture(provider) f.list.mockRejectedValueOnce( provider === 'google_drive' ? new GoogleDriveApiError(403, [], 'drive.files.list', false) - : new GoogleApiError('calendar.events.list', 403, reasons, reasons.length > 0) + : new GoogleApiError('calendar.events.list', 403, reasons, reasonsComplete) ) await f.step(4) expect(f.rows.get('a:content')).toMatchObject({ @@ -273,9 +281,9 @@ describe('durable Google company user scheduling', () => { } ) - it('continues past unavailable Calendar users without the unresolved-error pause and retries them later', async () => { - mocks.directory.mockResolvedValue({ users: ['a', 'b', 'c', 'z'].map(user) }) - const f = fixture() + it('completes Calendar users without the service so the listing stays reconcilable, re-probing them at the Directory refresh', async () => { + mocks.directory.mockResolvedValue({ users: ['a', 'b', 'c', 'z'].map((id) => user(id)) }) + const f = fixture('google_calendar', 15) const listUserDocuments = vi.fn( async (_token, _config, _cursor, ctx) => ({ documents: [{ ...document, externalId: memberDocumentId('event', ctx) }], @@ -302,27 +310,160 @@ describe('durable Google company user scheduling', () => { }) ) - await f.step(5) + await f.step(10) - expect(f.rows.get('z:content')?.complete).toBe(true) - expect(f.saved()).toMatchObject({ complete: false, unsafe: true, resumeAt: null }) for (const id of ['a', 'b', 'c']) { expect(f.rows.get(`${id}:content`)).toMatchObject({ - complete: false, - attempts: 1, + complete: true, + attempts: 0, retryAt: new Date('2026-09-17T01:00:00Z'), - failure: { status: 403, reasons: ['notACalendarUser'] }, }) + expect(f.rows.get(`${id}:content`)?.failure).toBeUndefined() } + expect(f.rows.get('z:content')).toMatchObject({ + complete: true, + retryAt: new Date('2026-09-17T00:15:00Z'), + }) + expect(f.saved()).toMatchObject({ complete: true, unsafe: false, listingFailures: null }) + expect(listUserDocuments).toHaveBeenCalledTimes(4) + }) + + it('refreshes permissions for a user without the Calendar service at the permission cadence', async () => { + mocks.directory.mockResolvedValue({ users: [user('a'), user('z')] }) + const f = fixture() + /** Skipping a permission pass hides nothing a retained failure would keep visible. */ + f.visible.mockResolvedValue(true) + f.list.mockResolvedValue({ documents: [document], hasMore: true, nextCursor: 'next-page' }) + await f.step(2) + f.advance(13 * 60 * 60 * 1000) + f.list.mockRejectedValueOnce( + new GoogleApiError('calendar.events.list', 403, ['notACalendarUser']) + ) + await f.step(2) + expect(mocks.directory).toHaveBeenCalledTimes(2) + expect(f.rows.get('a:permissions')).toMatchObject({ + attempts: 0, + retryAt: new Date('2026-09-18T01:00:00Z'), + }) + expect(f.rows.get('a:permissions')?.failure).toBeUndefined() + expect(f.rows.get('a:permissions')?.cursor).toBeUndefined() + expect(f.saved()).toMatchObject({ unsafe: false, listingFailures: null }) + }) + + it.each([ + ['gmail', false, false], + ['gmail', true, true], + ['google_calendar', false, true], + ] as const)( + 'for %s with visible documents %s, schedules a Directory user without a mailbox: %s', + async (provider, hasVisibleDocuments, scheduled) => { + mocks.directory.mockResolvedValue({ + users: [user('a', { isMailboxSetup: false }), user('z')], + }) + const f = fixture(provider) + f.visible.mockResolvedValue(hasVisibleDocuments) + await f.step(4) + expect(f.rows.has('a:content')).toBe(scheduled) + expect(f.rows.get('z:content')?.complete).toBe(true) + expect(f.saved()).toMatchObject({ unsafe: false, listingFailures: null }) + } + ) + + it.each([ + { + provider: 'google_calendar', + error: () => new GoogleApiError('calendar.events.list', 403, ['notACalendarUser']), + reason: { operation: 'calendar.events.list', status: 403, reasons: ['notACalendarUser'] }, + }, + { + provider: 'gmail', + error: () => new GoogleWorkspaceMailboxNotSetup(), + reason: { operation: 'directory.users.get', reasons: ['mailboxNotSetup'] }, + }, + ])( + 'retains a $provider user whose documents readers still see until the condition outlasts propagation', + async ({ provider, error, reason }) => { + mocks.directory.mockResolvedValue({ users: [user('a')] }) + const f = fixture(provider, 15) + f.visible.mockResolvedValue(true) + f.list.mockImplementation(async () => { + throw error() + }) + await f.step(5) + const retained = { + complete: false, + failure: { scope: 'a@fixture.test', ...reason, since: '2026-09-17T00:00:00.000Z' }, + } + expect(f.rows.get('a:content')).toMatchObject({ ...retained, attempts: 1 }) + expect(f.visible).toHaveBeenCalledWith(expect.objectContaining({ id: 'a' })) + expect(f.saved()).toMatchObject({ + complete: false, + unsafe: true, + listingFailures: { count: 1 }, + }) + + f.advance(23 * 60 * 60 * 1000) + f.restart() + await f.step(5) + expect(f.rows.get('a:content')).toMatchObject({ ...retained, attempts: 2 }) + expect(f.visible).toHaveBeenCalledOnce() + expect(f.saved().complete).toBe(false) + + f.advance(60 * 60 * 1000) + f.restart() + await f.step(5) + expect(f.rows.get('a:content')).toMatchObject({ complete: true, attempts: 0 }) + expect(f.rows.get('a:content')?.failure).toBeUndefined() + expect(f.saved()).toMatchObject({ complete: true, listingFailures: null }) + } + ) + + it('retains a service-not-enabled answer after the first page and restarts the user once it persists', async () => { + mocks.directory.mockResolvedValue({ users: [user('a')] }) + const f = fixture() + f.list + .mockResolvedValueOnce({ documents: [document], hasMore: true, nextCursor: 'page-2' }) + .mockRejectedValue(new GoogleApiError('calendar.events.list', 403, ['notACalendarUser'])) + await f.step(5) + expect(f.list.mock.calls.at(-1)?.[2]).toBe('page-2') + expect(f.rows.get('a:content')).toMatchObject({ + complete: false, + cursor: 'page-2', + failure: { reasons: ['notACalendarUser'], since: '2026-09-17T00:00:00.000Z' }, + }) + expect(f.visible).not.toHaveBeenCalled() + + f.advance(25 * 60 * 60 * 1000) + f.restart() + await f.step(5) + expect(f.rows.get('a:content')).toMatchObject({ complete: false }) + expect(f.rows.get('a:content')?.cursor).toContain('google-workspace:v1:') f.advance(60 * 60 * 1000) f.restart() - await f.step(4) + await f.step(5) + expect(f.rows.get('a:content')).toMatchObject({ complete: true }) + expect(f.rows.get('a:content')?.failure).toBeUndefined() + }) - for (const id of ['a', 'b', 'c']) { - expect(f.rows.get(`${id}:content`)).toMatchObject({ complete: true, attempts: 0 }) - expect(f.rows.get(`${id}:content`)?.failure).toBeUndefined() - } + it('picks up a Gmail user on the Directory refresh after their mailbox is provisioned', async () => { + mocks.directory + .mockResolvedValueOnce({ users: [user('a', { isMailboxSetup: false }), user('z')] }) + .mockResolvedValue({ users: [user('a', { isMailboxSetup: true }), user('z')] }) + const f = fixture('gmail') + let page = 0 + f.list.mockImplementation(async () => ({ + documents: [document], + hasMore: true, + nextCursor: `page-${++page}`, + })) + await f.step(3) + expect(f.rows.has('a:content')).toBe(false) + f.advance(61 * 60_000) + f.restart() + await f.step(2) + expect(mocks.directory).toHaveBeenCalledTimes(2) + expect(f.rows.has('a:content')).toBe(true) }) it('bounds a run of unresolved user errors rather than marking the tenant complete', async () => { @@ -457,8 +598,9 @@ describe('durable Google company user scheduling', () => { samples: [ { scope: 'a@fixture.test', - operation: 'directory.users.get', - reasons: ['mailboxNotSetup'], + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], }, ], }, diff --git a/apps/sim/lib/knowledge/connectors/google-company-scheduler.ts b/apps/sim/lib/knowledge/connectors/google-company-scheduler.ts index 8acdd0b9cfe..d99fc82cb35 100644 --- a/apps/sim/lib/knowledge/connectors/google-company-scheduler.ts +++ b/apps/sim/lib/knowledge/connectors/google-company-scheduler.ts @@ -6,10 +6,14 @@ import type { import { googleDriveCompanyCursorAdapter } from '@/connectors/google-drive/company-crawl' import { GoogleDriveApiError } from '@/connectors/google-drive/google-drive-errors' import { GoogleApiError } from '@/connectors/google-workspace/api-errors' -import { googleWorkspaceCompanyCursorAdapter } from '@/connectors/google-workspace/company-crawl' -import type { - GoogleCompanyCursorAdapter, - GoogleCompanyUserWork, +import { + googleWorkspaceCompanyCursorAdapter, + serviceNotEnabledFailure, +} from '@/connectors/google-workspace/company-crawl' +import { + type GoogleCompanyCursorAdapter, + type GoogleCompanyUserWork, + googleCompanyUserContextSchema, } from '@/connectors/google-workspace/company-work' import { listGoogleWorkspaceUsers, @@ -27,6 +31,8 @@ const CURSOR_PREFIX = 'google-company-work:v2:' export const GOOGLE_COMPANY_PERMISSION_REFRESH_MS = 12 * 60 * 60 * 1000 const DIRECTORY_REFRESH_MS = 60 * 60 * 1000 const MAX_UNRESOLVED_FAILURES_PER_PASS = 3 +/** Google applies a change to a user's services or organizational unit within 24 hours. */ +const SERVICE_CHANGE_PROPAGATION_MS = 24 * 60 * 60 * 1000 const cursorSchema = z.object({ directoryComplete: z.boolean(), directoryCursor: z.string().max(8192).optional(), @@ -114,6 +120,8 @@ export function createGoogleCompanyScheduler(input: { store: ConnectorPartitionWorkStore listDocuments: ConnectorConfig['listDocuments'] isListingCursorInvalidError?: ConnectorConfig['isListingCursorInvalidError'] + /** Whether this user still has documents readers can see, which a skip would remove. */ + hasVisibleDocuments: (user: GoogleCompanyUserWork['user']) => Promise syncIntervalMinutes: number now?: () => Date }) { @@ -183,9 +191,18 @@ export function createGoogleCompanyScheduler(input: { nextCursor: nextCursor({ ...state, directoryCursor: undefined }, {}), } } - const users = page.users.filter( - (user) => user.active && (!selected.length || selected.includes(user.email)) - ) + const users: typeof page.users = [] + for (const user of page.users) { + if (!user.active || (selected.length && !selected.includes(user.email))) continue + /** A user without a mailbox is out of scope like an inactive one, unless readers still see their mail. */ + if ( + input.provider === 'gmail' && + user.isMailboxSetup === false && + !(await input.hasVisibleDocuments(user)) + ) + continue + users.push(user) + } return { documents: [], currentCursor: writeCursor(state), @@ -249,6 +266,8 @@ export function createGoogleCompanyScheduler(input: { active: { userId: work.partitionKey, kind: work.kind }, } const currentCursor = writeCursor(active) + /** The user's first provider page, from identity alone so extra persisted fields never alter it. */ + const seed = adapter.seed(googleCompanyUserContextSchema.parse(work.context)) const next = { ...state, revision: active.revision, @@ -278,7 +297,7 @@ export function createGoogleCompanyScheduler(input: { update: { partitionKey: work.partitionKey, kind: work.kind, - cursor: resetCursor ? adapter.seed(work.context) : (work.cursor ?? null), + cursor: resetCursor ? seed : (work.cursor ?? null), completed: false, retryAt, attempts: work.attempts + 1, @@ -296,14 +315,58 @@ export function createGoogleCompanyScheduler(input: { : {}), } } + /** A user without the service completes cleanly and is re-probed no sooner than the Directory refresh. */ + const skipped = (): ExternalDocumentList => ({ + documents: [], + currentCursor, + hasMore: true, + nextCursor: nextCursor(next, { + update: { + partitionKey: work.partitionKey, + kind: work.kind, + cursor: null, + completed: true, + attempts: 0, + failure: null, + retryAt: new Date( + now().getTime() + + (work.kind === 'permissions' + ? GOOGLE_COMPANY_PERMISSION_REFRESH_MS + : Math.max(DIRECTORY_REFRESH_MS, input.syncIntervalMinutes * 60_000)) + ), + ...(work.kind === 'permissions' ? { permissionStartedAt: null } : {}), + }, + }), + }) + /** + * Only a user's first page proves the whole account lacks the service, and a user whose + * documents readers still see keeps them until the condition outlasts Google's propagation + * window. Until then it is a retained failure, which holds absence reconciliation. + */ + const unavailable = async ( + reason: Omit + ): Promise => { + const since = work.failure?.since + const persisted = + since !== undefined && + now().getTime() - new Date(since).getTime() >= SERVICE_CHANGE_PROPAGATION_MS + const firstPage = !work.cursor || work.cursor === seed + if ( + firstPage && + (persisted || + work.kind === 'permissions' || + (!since && !(await input.hasVisibleDocuments(work.context)))) + ) + return skipped() + return failed( + { scope: work.context.email, ...reason, since: since ?? now().toISOString() }, + false, + persisted + ) + } let page: ExternalDocumentList try { - page = await input.listDocuments( - accessToken, - sourceConfig, - work.cursor ?? adapter.seed(work.context), - syncContext - ) + page = await input.listDocuments(accessToken, sourceConfig, work.cursor ?? seed, syncContext) } catch (error) { signal?.throwIfAborted() if (input.isListingCursorInvalidError?.(error)) @@ -312,6 +375,8 @@ export function createGoogleCompanyScheduler(input: { false, true ) + const serviceNotEnabled = serviceNotEnabledFailure(error) + if (serviceNotEnabled) return unavailable(serviceNotEnabled) const failure = deferredUserFailure(error, work.context) if (!failure) throw error return failed(failure, true) diff --git a/apps/sim/lib/knowledge/connectors/partition-store.ts b/apps/sim/lib/knowledge/connectors/partition-store.ts index 02009652d1c..45fae9b2d53 100644 --- a/apps/sim/lib/knowledge/connectors/partition-store.ts +++ b/apps/sim/lib/knowledge/connectors/partition-store.ts @@ -8,7 +8,11 @@ import type { ConnectorPartitionWorkKind, ConnectorPartitionWorkStore, } from '@/lib/knowledge/connectors/partition-work' -import { listingFailuresSchema, MAX_LISTING_FAILURE_SAMPLES } from '@/connectors/listing-failures' +import { + listingFailureSampleSchema, + listingFailuresSchema, + MAX_LISTING_FAILURE_SAMPLES, +} from '@/connectors/listing-failures' type Row = typeof knowledgeConnectorPartition.$inferSelect const work = knowledgeConnectorPartition @@ -25,6 +29,9 @@ function project( cursor: (kind === 'content' ? row.cursor : row.permissionCursor) ?? undefined, attempts: kind === 'content' ? row.attempts : row.permissionAttempts, hasFailure: row.failure !== null || row.permissionFailure !== null, + failure: listingFailureSampleSchema.safeParse( + kind === 'content' ? row.failure : row.permissionFailure + ).data, permissionStartedAt: row.permissionStartedAt ?? undefined, } } diff --git a/apps/sim/lib/knowledge/connectors/partition-work.ts b/apps/sim/lib/knowledge/connectors/partition-work.ts index a0bb80fb711..2ad1c27869e 100644 --- a/apps/sim/lib/knowledge/connectors/partition-work.ts +++ b/apps/sim/lib/knowledge/connectors/partition-work.ts @@ -11,6 +11,8 @@ export interface ConnectorPartitionWorkItem { cursor?: string attempts: number hasFailure: boolean + /** This kind's retained failure, whose `since` carries a standing condition's first observation. */ + failure?: ListingFailure permissionStartedAt?: Date } diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts index 950fa21a115..150c3635638 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts @@ -45,6 +45,7 @@ import { resolveReconciliationDeleteCap, storedHashIsCurrent, } from '@/lib/knowledge/connectors/sync-primitives' +import { hasVisibleUserDocuments } from '@/lib/knowledge/connectors/user-document-visibility' import { hardDeleteDocuments } from '@/lib/knowledge/documents/service' import { SIM_SEARCH_SYNC_INTERVAL_MINUTES } from '@/lib/sim-search/constants' import { googleCompanyUserContextSchema } from '@/connectors/google-workspace/company-work' @@ -139,6 +140,7 @@ export async function runConnectorContentPass(input: ContentPassInput) { provider: input.connector.connectorType, listDocuments: input.connectorConfig.listDocuments, isListingCursorInvalidError: input.connectorConfig.isListingCursorInvalidError, + hasVisibleDocuments: (user) => hasVisibleUserDocuments(input.connectorId, user.email), syncIntervalMinutes, store: { get: (...args) => companyStore().get(...args), diff --git a/apps/sim/lib/knowledge/connectors/user-document-visibility.ts b/apps/sim/lib/knowledge/connectors/user-document-visibility.ts new file mode 100644 index 00000000000..f2034965306 --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/user-document-visibility.ts @@ -0,0 +1,34 @@ +import { db } from '@sim/db' +import { document } from '@sim/db/schema' +import { sql } from 'drizzle-orm' +import { SOURCE_ACL_MAX_AGE_MS } from '@/lib/knowledge/access/freshness' +import { userToken } from '@/lib/knowledge/access/tokens' + +/** + * Whether readers can still see any of this connector's documents granted to one user: a live, + * included document carrying their token with permission evidence inside the freshness limit. + * The user's grants are materialized from `doc_acl_gin_idx` first, so the probe is bounded by + * that user's grants instead of the connector's size. Planned inline, `LIMIT 1` makes a + * sequential scan look cheaper than the index, because PostgreSQL cannot estimate array overlap. + */ +export async function hasVisibleUserDocuments( + connectorId: string, + email: string +): Promise { + const token = userToken(email) + if (!token) return false + const rows = await db.execute(sql` + WITH granted AS MATERIALIZED ( + SELECT ${document.connectorId}, ${document.userExcluded}, ${document.archivedAt}, ${document.aclVerifiedAt} + FROM ${document} + WHERE ${document.deletedAt} IS NULL AND ${document.acl} && ARRAY[${token}]::text[] + ) + SELECT 1 FROM granted AS ${document} + WHERE ${document.connectorId} = ${connectorId} + AND ${document.userExcluded} = false + AND ${document.archivedAt} IS NULL + AND ${document.aclVerifiedAt} > statement_timestamp() - (${SOURCE_ACL_MAX_AGE_MS} * interval '1 millisecond') + LIMIT 1 + `) + return rows.length > 0 +}