From 2216fbabdeba3d4993e797f66fcbc9aa28148468 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 11:33:32 -0700 Subject: [PATCH 1/2] perf(audit-logs): read a listing narrowed only by time through per-workspace keyset scans An organization's audit listing is one OR predicate over the organization's workspaces and its organization-level rows. With no filter, or only a time range (the audit page's default view), PostgreSQL has no index that returns that OR in listing order, so it collects every row of the scope inside the range and sorts it for each page; for a busy organization that is a scan of a large part of the table. Such a listing now reads its page ids as a disjoint union: one keyset scan of audit_log_workspace_created_at_id_idx per organization workspace, the organization-level metadata rows through audit_log_org_level_created_at_id_idx, and the organization resource rows that do not name it in metadata. Each part stops after a page, and the first page of the union is exactly the first page of the scope. The time range is kept verbatim and also stated as implied bounds on the sort key, so each scan starts and stops at the range; the cursor is the same keyset position written as a row comparison so the scan seeks to it. A listing with any other filter (action, resource, actor, workspace, search) keeps the single predicate unchanged, byte for byte: per-workspace scans cannot stop early under such a filter, which is what made #9419 slow, while the single predicate lets the filter's own index drive. queryAuditLogs now takes the scope and the filters rather than prebuilt SQL so it can make that choice. list-scope.integration.ts pages every filter shape, with and without departed members, across page sizes and the byte budget, and requires exactly the single predicate's order, including sub-millisecond ties, bounds inside a millisecond and a workspace the organization no longer holds. list-plans.integration.ts pins the keyset plan for the unfiltered and time-bounded shapes (failing on the previous single-predicate listing) and the single predicate for every other one. --- apps/sim/app/api/v1/audit-logs/route.test.ts | 24 +-- apps/sim/app/api/v1/audit-logs/route.ts | 34 +--- .../application/export-audit-logs.ts | 11 +- .../audit-logs/application/list-audit-logs.ts | 27 +-- .../audit-logs/application/use-cases.test.ts | 10 +- .../lib/audit-logs/list-plans.integration.ts | 143 +++++++++++++ .../lib/audit-logs/list-scope.integration.ts | 118 ++++++++--- apps/sim/lib/audit-logs/query.integration.ts | 25 ++- apps/sim/lib/audit-logs/query.ts | 189 ++++++++++++++---- 9 files changed, 441 insertions(+), 140 deletions(-) create mode 100644 apps/sim/lib/audit-logs/list-plans.integration.ts diff --git a/apps/sim/app/api/v1/audit-logs/route.test.ts b/apps/sim/app/api/v1/audit-logs/route.test.ts index 0ee0cc08384..d7dab9fd7a3 100644 --- a/apps/sim/app/api/v1/audit-logs/route.test.ts +++ b/apps/sim/app/api/v1/audit-logs/route.test.ts @@ -7,19 +7,12 @@ import { v1LogsMetaMock, v1LogsMetaMockFns } from '@sim/testing/mocks/v1-logs-me import { v1MiddlewareMock, v1MiddlewareMockFns } from '@sim/testing/mocks/v1-middleware.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' -const { - mockValidateV1EnterpriseAuditAccess, - mockBuildOrgScopeCondition, - mockGetOrgWorkspaceIds, - mockQueryAuditLogs, - mockBuildFilterConditions, -} = vi.hoisted(() => ({ - mockValidateV1EnterpriseAuditAccess: vi.fn(), - mockBuildOrgScopeCondition: vi.fn(), - mockGetOrgWorkspaceIds: vi.fn(), - mockQueryAuditLogs: vi.fn(), - mockBuildFilterConditions: vi.fn(), -})) +const { mockValidateV1EnterpriseAuditAccess, mockGetOrgWorkspaceIds, mockQueryAuditLogs } = + vi.hoisted(() => ({ + mockValidateV1EnterpriseAuditAccess: vi.fn(), + mockGetOrgWorkspaceIds: vi.fn(), + mockQueryAuditLogs: vi.fn(), + })) vi.mock('@/app/api/v1/middleware', () => v1MiddlewareMock) @@ -28,8 +21,6 @@ vi.mock('@/app/api/v1/audit-logs/auth', () => ({ })) vi.mock('@/lib/audit-logs/query', () => ({ - buildFilterConditions: mockBuildFilterConditions, - buildOrgScopeCondition: mockBuildOrgScopeCondition, getOrgWorkspaceIds: mockGetOrgWorkspaceIds, queryAuditLogs: mockQueryAuditLogs, })) @@ -48,7 +39,6 @@ v1LogsMetaMockFns.mockCreateApiResponse.mockImplementation((body: unknown) => ({ const ORG_ID = 'org-1' const MEMBER_IDS = ['admin-1', 'member-1'] const ORG_WORKSPACE_IDS = ['ws-org-1', 'ws-org-2'] -const SCOPE_SENTINEL = { type: 'org-scope-sentinel' } function makeRequest(query: string) { return createMockRequest('GET', undefined, {}, `http://localhost:3000/api/v1/audit-logs${query}`) @@ -63,8 +53,6 @@ describe('GET /api/v1/audit-logs', () => { context: { organizationId: ORG_ID, orgMemberIds: MEMBER_IDS }, }) mockGetOrgWorkspaceIds.mockResolvedValue(ORG_WORKSPACE_IDS) - mockBuildOrgScopeCondition.mockReturnValue(SCOPE_SENTINEL) - mockBuildFilterConditions.mockReturnValue([]) mockQueryAuditLogs.mockResolvedValue({ data: [], nextCursor: undefined }) }) diff --git a/apps/sim/app/api/v1/audit-logs/route.ts b/apps/sim/app/api/v1/audit-logs/route.ts index 26498d298a8..187e3a83b1e 100644 --- a/apps/sim/app/api/v1/audit-logs/route.ts +++ b/apps/sim/app/api/v1/audit-logs/route.ts @@ -25,12 +25,7 @@ import { generateId } from '@sim/utils/id' import { type NextRequest, NextResponse } from 'next/server' import { v1ListAuditLogsContract } from '@/lib/api/contracts/v1/audit-logs' import { parseRequest } from '@/lib/api/server' -import { - buildFilterConditions, - buildOrgScopeCondition, - getOrgWorkspaceIds, - queryAuditLogs, -} from '@/lib/audit-logs/query' +import { getOrgWorkspaceIds, queryAuditLogs } from '@/lib/audit-logs/query' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { validateV1EnterpriseAuditAccess } from '@/app/api/v1/audit-logs/auth' import { formatAuditLogEntry } from '@/app/api/v1/audit-logs/format' @@ -99,24 +94,17 @@ export const GET = withRouteHandler(async (request: NextRequest) => { ) } - const scopeCondition = buildOrgScopeCondition({ - organizationId, - orgWorkspaceIds, - orgMemberIds, - includeDeparted: params.includeDeparted, - }) - const filterConditions = buildFilterConditions({ - action: params.action, - resourceType: params.resourceType, - resourceId: params.resourceId, - workspaceId: params.workspaceId, - actorId: params.actorId, - startDate: params.startDate, - endDate: params.endDate, - }) - const { data, nextCursor } = await queryAuditLogs( - [scopeCondition, ...filterConditions], + { organizationId, orgWorkspaceIds, orgMemberIds, includeDeparted: params.includeDeparted }, + { + action: params.action, + resourceType: params.resourceType, + resourceId: params.resourceId, + workspaceId: params.workspaceId, + actorId: params.actorId, + startDate: params.startDate, + endDate: params.endDate, + }, params.limit, params.cursor ) diff --git a/apps/sim/lib/audit-logs/application/export-audit-logs.ts b/apps/sim/lib/audit-logs/application/export-audit-logs.ts index ce0507f2989..10e630c2438 100644 --- a/apps/sim/lib/audit-logs/application/export-audit-logs.ts +++ b/apps/sim/lib/audit-logs/application/export-audit-logs.ts @@ -4,8 +4,7 @@ import { auditLogOperations } from '@/lib/audit-logs/application/operations' import { AUDIT_LOG_CSV_HEADER, toAuditLogCsvRow } from '@/lib/audit-logs/csv' import { type AuditLogFilterParams, - buildFilterConditions, - buildOrgScopeCondition, + type AuditLogListScope, getOrgWorkspaceIds, queryAuditLogs, } from '@/lib/audit-logs/query' @@ -37,13 +36,12 @@ export const exportAuditLogs = defineAuthorizedAuditLogUseCase({ ) { throw new OrchestrationError('validation', 'actorId is not a member of your organization') } - const scope = buildOrgScopeCondition({ + const scope: AuditLogListScope = { organizationId: context.organizationId, orgWorkspaceIds: workspaceIds, orgMemberIds: context.orgMemberIds, includeDeparted: input.includeDeparted, - }) - const conditions = [scope, ...buildFilterConditions(input.filters)] + } const lines = [AUDIT_LOG_CSV_HEADER] let rowCount = 0 let cursor: string | undefined @@ -52,7 +50,8 @@ export const exportAuditLogs = defineAuthorizedAuditLogUseCase({ pages: while (rowCount < EXPORT_MAX_ROWS && bytes < EXPORT_MAX_BYTES) { request?.signal?.throwIfAborted() const page = await queryAuditLogs( - conditions, + scope, + input.filters, Math.min(EXPORT_PAGE_ROWS, EXPORT_MAX_ROWS - rowCount), cursor, EXPORT_MAX_BYTES diff --git a/apps/sim/lib/audit-logs/application/list-audit-logs.ts b/apps/sim/lib/audit-logs/application/list-audit-logs.ts index 97805b2b503..46a4a4794d0 100644 --- a/apps/sim/lib/audit-logs/application/list-audit-logs.ts +++ b/apps/sim/lib/audit-logs/application/list-audit-logs.ts @@ -2,8 +2,6 @@ import { defineAuthorizedAuditLogUseCase } from '@/lib/audit-logs/application/au import { auditLogOperations } from '@/lib/audit-logs/application/operations' import { type AuditLogFilterParams, - buildFilterConditions, - buildOrgScopeCondition, decodeAuditLogCursor, getOrgWorkspaceIds, queryAuditLogs, @@ -32,22 +30,25 @@ export const listAuditLogs = defineAuthorizedAuditLogUseCase({ if (input.filters.workspaceId && !orgWorkspaceIds.includes(input.filters.workspaceId)) { throw new OrchestrationError('validation', 'workspaceId does not belong to your organization') } - const scopeCondition = buildOrgScopeCondition({ - organizationId: context.organizationId, - orgWorkspaceIds, - orgMemberIds: context.orgMemberIds, - includeDeparted: input.includeDeparted, - }) if ( context.workspaceId && input.filters.workspaceId && context.workspaceId !== input.filters.workspaceId ) throw new OrchestrationError('not_found', 'Audit logs not found in the selected workspace') - const filterConditions = buildFilterConditions({ - ...input.filters, - ...(context.workspaceId ? { workspaceId: context.workspaceId } : {}), - }) - return queryAuditLogs([scopeCondition, ...filterConditions], input.limit, input.cursor) + return queryAuditLogs( + { + organizationId: context.organizationId, + orgWorkspaceIds, + orgMemberIds: context.orgMemberIds, + includeDeparted: input.includeDeparted, + }, + { + ...input.filters, + ...(context.workspaceId ? { workspaceId: context.workspaceId } : {}), + }, + input.limit, + input.cursor + ) }, }) diff --git a/apps/sim/lib/audit-logs/application/use-cases.test.ts b/apps/sim/lib/audit-logs/application/use-cases.test.ts index d54ffc5f026..2fc046c11c3 100644 --- a/apps/sim/lib/audit-logs/application/use-cases.test.ts +++ b/apps/sim/lib/audit-logs/application/use-cases.test.ts @@ -29,7 +29,6 @@ const hoisted = vi.hoisted(() => ({ resolveDefaultOrganization: vi.fn(), getOrgWorkspaceIds: vi.fn(), buildOrgScopeCondition: vi.fn(), - buildFilterConditions: vi.fn(), decodeAuditLogCursor: vi.fn(), queryAuditLogs: vi.fn(), isCapabilityWithheldForUser: vi.fn(), @@ -50,7 +49,6 @@ vi.mock('@/lib/audit-logs/authorization', () => ({ vi.mock('@/lib/audit-logs/query', () => ({ getOrgWorkspaceIds: hoisted.getOrgWorkspaceIds, buildOrgScopeCondition: hoisted.buildOrgScopeCondition, - buildFilterConditions: hoisted.buildFilterConditions, decodeAuditLogCursor: hoisted.decodeAuditLogCursor, queryAuditLogs: hoisted.queryAuditLogs, })) @@ -96,7 +94,6 @@ describe('audit-log application use cases', () => { }) mocks.getOrgWorkspaceIds.mockResolvedValue(['workspace-1']) mocks.buildOrgScopeCondition.mockReturnValue({ type: 'scope' }) - mocks.buildFilterConditions.mockReturnValue([]) mocks.decodeAuditLogCursor.mockReturnValue({ createdAt: '2026-01-01T00:00:00.000Z', id: 'audit-1', @@ -111,7 +108,12 @@ describe('audit-log application use cases', () => { await listAuditLogs.execute({ principal, input: { ...listInput, organizationId: undefined } }) expect(mocks.resolveAccess).toHaveBeenCalledWith('admin-1', 'organization-1') expect(mocks.resolveDefaultOrganization).not.toHaveBeenCalled() - expect(mocks.buildFilterConditions).toHaveBeenCalledWith({ workspaceId: 'workspace-1' }) + expect(mocks.queryAuditLogs).toHaveBeenCalledWith( + expect.objectContaining({ organizationId: 'organization-1' }), + { workspaceId: 'workspace-1' }, + 50, + undefined + ) mocks.resolveAccess.mockResolvedValueOnce({ success: false, code: 'FORBIDDEN', diff --git a/apps/sim/lib/audit-logs/list-plans.integration.ts b/apps/sim/lib/audit-logs/list-plans.integration.ts new file mode 100644 index 00000000000..75cc09bd568 --- /dev/null +++ b/apps/sim/lib/audit-logs/list-plans.integration.ts @@ -0,0 +1,143 @@ +/** + * The plans an organization's audit-log listing must get, proven on the exact SQL `queryAuditLogs` + * sends. Narrowed by nothing but a time range, each workspace is read by a keyset scan of + * `audit_log_workspace_created_at_id_idx` that a page limit stops, never by collecting and sorting + * the whole scope. Any other filter keeps the single scope predicate, so its plan is the one the + * filter's own index drives. + */ +import { db } from '@sim/db' +import { auditLog, user, workspace } from '@sim/db/schema' +import { explainCaptured, type PlanNode } from '@sim/db/testing/explain' +import { assertDisposableTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { generateId } from '@sim/utils/id' +import { eq, like, sql } from 'drizzle-orm' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { + type AuditLogFilterParams, + type AuditLogListScope, + queryAuditLogs, +} from '@/lib/audit-logs/query' + +const organizationId = generateId() +const memberId = generateId() +const workspaceId = generateId() +const idPrefix = `audit_plan_${generateId()}_` +const workspaceIndex = 'audit_log_workspace_created_at_id_idx' +const orgLevelIndex = 'audit_log_org_level_created_at_id_idx' + +const cursor = Buffer.from( + JSON.stringify({ createdAt: '2026-01-01T03:00:00.000Z', id: `${idPrefix}~` }) +).toString('base64') + +const listingPlan = ( + filters: AuditLogFilterParams, + options: { includeDeparted?: boolean; cursor?: string } = {} +) => { + const scope: AuditLogListScope = { + organizationId, + orgWorkspaceIds: [workspaceId], + orgMemberIds: [memberId], + includeDeparted: options.includeDeparted ?? false, + } + return explainCaptured(db, () => queryAuditLogs(scope, filters, 50, options.cursor)) +} + +/** The workspace part's keyset scans, each directly under the limit that stops it after a page. */ +const limitedWorkspaceScans = (nodes: PlanNode[]) => + nodes + .filter((node) => node['Node Type'] === 'Limit') + .map((node) => node.Plans?.[0]) + .filter((child) => child?.['Index Name'] === workspaceIndex) + +const KEYSET_CASES: [ + string, + AuditLogFilterParams, + { includeDeparted?: boolean; cursor?: string }, +][] = [ + ['no filter, current members', {}, {}], + ['no filter, departed members included', {}, { includeDeparted: true }], + ['no filter, after a cursor', {}, { cursor }], + ['a start date', { startDate: '2026-01-01T01:00:00.000Z' }, {}], + [ + 'a date range, after a cursor', + { startDate: '2026-01-01T01:00:00.000Z', endDate: '2026-01-01T04:00:00.000Z' }, + { cursor }, + ], +] + +const NARROWED_CASES: [string, AuditLogFilterParams][] = [ + ['an action', { action: 'organization.updated' }], + [ + 'an action and a start date', + { action: 'organization.updated', startDate: '2026-01-01T01:00:00.000Z' }, + ], + ['a resource type', { resourceType: 'workflow' }], + ['a resource', { resourceType: 'workflow', resourceId: 'resource-7' }], + ['an actor', { actorId: memberId }], + ['a workspace', { workspaceId }], + ['a search', { search: 'fixture' }], +] + +describe('organization audit-log listing plans in PostgreSQL', () => { + beforeAll(async () => { + assertDisposableTestDatabaseUrl(process.env.DATABASE_URL ?? '') + await db.insert(user).values({ + id: memberId, + name: 'Audit plan fixture', + email: `${memberId}@audit-plan.test`, + emailVerified: true, + createdAt: new Date(), + updatedAt: new Date(), + }) + await insertWorkspaceFixture(db, { + id: workspaceId, + name: 'Audit plan fixture', + ownerId: memberId, + billedAccountUserId: memberId, + }) + // A workspace with a long history, and workspace-less rows of many other organizations, so + // collecting the scope and sorting it costs far more than a keyset scan that stops at a page. + await db.execute(sql` + INSERT INTO ${auditLog} (id, workspace_id, actor_id, action, resource_type, resource_id, + metadata, created_at) + SELECT ${idPrefix} || n, CASE WHEN n % 4 = 1 THEN ${workspaceId} END, + CASE WHEN n % 2 = 0 THEN ${memberId} END, 'organization.updated', + CASE WHEN n % 997 = 0 THEN 'organization' ELSE 'workflow' END, + CASE WHEN n % 997 = 0 THEN ${organizationId} ELSE 'resource-' || n END, + jsonb_build_object('organizationId', + CASE WHEN n % 50 = 0 THEN ${organizationId} ELSE 'other-organization-' || (n % 400) END), + timestamp '2026-01-01' + n * interval '1 second' + FROM generate_series(1, 20000) AS n`) + await db.execute(sql`VACUUM ANALYZE ${auditLog}`) + }) + + afterAll(async () => { + await db.delete(auditLog).where(like(auditLog.id, `${idPrefix}%`)) + await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) + await db.delete(user).where(eq(user.id, memberId)) + }) + + it.each(KEYSET_CASES)( + 'reads each workspace newest first through its keyset index and stops after a page (%s)', + async (_label, filters, options) => { + const nodes = await listingPlan(filters, options) + const [scan] = limitedWorkspaceScans(nodes) + expect(scan).toMatchObject({ 'Node Type': 'Index Scan', 'Scan Direction': 'Backward' }) + expect(scan?.['Index Cond']).toMatch(/workspace_id = org_workspace\.workspace_id/) + if (filters.startDate) expect(scan?.['Index Cond']).toMatch(/date_trunc\(.*\) >= /) + if (filters.endDate) expect(scan?.['Index Cond']).toMatch(/date_trunc\(.*\) <= /) + if (options.cursor) expect(scan?.['Index Cond']).toMatch(/ROW\(date_trunc/) + expect(nodes.filter((node) => node['Index Name'] === orgLevelIndex)).toHaveLength(1) + } + ) + + it.each(NARROWED_CASES)( + 'keeps the single scope predicate when the listing is narrowed by %s', + async (_label, filters) => { + const nodes = await listingPlan(filters) + expect(nodes.filter((node) => node['Node Type'] === 'Function Scan')).toEqual([]) + expect(nodes.filter((node) => /org_workspace/.test(node['Index Cond'] ?? ''))).toEqual([]) + } + ) +}) diff --git a/apps/sim/lib/audit-logs/list-scope.integration.ts b/apps/sim/lib/audit-logs/list-scope.integration.ts index 2de211a2e4c..cbc14df3596 100644 --- a/apps/sim/lib/audit-logs/list-scope.integration.ts +++ b/apps/sim/lib/audit-logs/list-scope.integration.ts @@ -2,9 +2,15 @@ import { db } from '@sim/db' import { auditLog, user, workspace } from '@sim/db/schema' import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' import { generateId } from '@sim/utils/id' -import { and, desc, eq, inArray, type SQL, sql } from 'drizzle-orm' +import { and, desc, eq, inArray, sql } from 'drizzle-orm' import { afterAll, beforeAll, describe, expect, it } from 'vitest' -import { buildOrgScopeCondition, type OrgScopeParams, queryAuditLogs } from '@/lib/audit-logs/query' +import { + type AuditLogFilterParams, + type AuditLogListScope, + buildFilterConditions, + buildOrgScopeCondition, + queryAuditLogs, +} from '@/lib/audit-logs/query' const organizationId = generateId() const otherOrganizationId = generateId() @@ -56,38 +62,38 @@ const rows = Array.from({ length: 81 }, (_, index) => { ? 'organization' : 'workflow', resourceId: placement.resourceId, + resourceName: index % 5 === 0 ? 'Quarterly report' : 'Fixture resource', metadata: { ...placement.metadata, fixture: 'x'.repeat(200) }, } }) const fixtureIds = rows.map((row) => row.id) +/** The fixture clock: row `index` is written at `millisecond(floor(index / 4))` plus `index % 4` × 100µs. */ +const millisecond = (offset: number) => + new Date(Date.UTC(2026, 0, 1, 12, 0, 0, offset)).toISOString() + const auditLogTimestamp = sql`date_trunc('milliseconds', ${auditLog.createdAt})` -/** The listing's contract as one predicate: every row in scope, newest first. */ -async function referenceOrder(scope: OrgScopeParams, filters: SQL[]) { +/** The listing's contract as the single scope predicate: every matching row in scope, newest first. */ +async function referenceOrder(scope: AuditLogListScope, filters: AuditLogFilterParams) { const matching = await db .select({ id: auditLog.id }) .from(auditLog) - .where(and(buildOrgScopeCondition(scope), ...filters)) + .where(and(buildOrgScopeCondition(scope), ...buildFilterConditions(filters))) .orderBy(desc(auditLogTimestamp), desc(auditLog.id)) return matching.map((row) => row.id) } async function paginate( - scope: OrgScopeParams, - filters: SQL[], + scope: AuditLogListScope, + filters: AuditLogFilterParams, limit: number, maxBytes?: number ) { const seen: string[] = [] let cursor: string | undefined for (let page = 0; page <= rows.length; page++) { - const result = await queryAuditLogs( - [buildOrgScopeCondition(scope), ...filters], - limit, - cursor, - maxBytes - ) + const result = await queryAuditLogs(scope, filters, limit, cursor, maxBytes) expect(result.truncated ?? false).toBe(false) seen.push(...result.data.map((row) => row.id)) if (!result.nextCursor) return seen @@ -136,26 +142,47 @@ afterAll(async () => { await db.$client.end() }) +/** + * A time range alone is read per workspace through the keyset index; any other filter keeps the + * single predicate. Bounds fall inside a millisecond holding four rows, so a bound on the sort key + * that admitted the whole millisecond would show up as extra rows. + */ +const FILTERS: [string, AuditLogFilterParams][] = [ + ['no filter', {}], + ['a start date', { startDate: millisecond(8) }], + ['an end date inside a millisecond', { endDate: millisecond(15) }], + ['a date range', { startDate: millisecond(3), endDate: millisecond(15) }], + ['a start date near the newest rows', { startDate: millisecond(19) }], + ['an action', { action: 'workflow.deployed' }], + ['an action and a start date', { action: 'workflow.deployed', startDate: millisecond(8) }], + ['a resource type', { resourceType: 'organization' }], + [ + 'resource types and an end date', + { resourceType: 'workflow, organization', endDate: millisecond(15) }, + ], + ['a resource', { resourceId: organizationId }], + ['an actor', { actorId: memberId }], + ['a workspace', { workspaceId: workspaceB }], + ['a search', { search: 'quarterly' }], +] + describe('queryAuditLogs scope against real Postgres', () => { - const fixtureOnly = inArray(auditLog.id, fixtureIds) - - it.each([ - { includeDeparted: false, action: undefined }, - { includeDeparted: true, action: undefined }, - { includeDeparted: false, action: 'workflow.deployed' }, - { includeDeparted: true, action: 'workflow.deployed' }, - ])( - 'pages through exactly the single-predicate result (%o)', - async ({ includeDeparted, action }) => { - const scope: OrgScopeParams = { + it.each( + FILTERS.flatMap(([label, filters]) => [ + { label, includeDeparted: false, filters }, + { label, includeDeparted: true, filters }, + ]) + )( + 'pages through exactly the single-predicate result ($label, departed included: $includeDeparted)', + async ({ includeDeparted, filters }) => { + const scope: AuditLogListScope = { organizationId, orgWorkspaceIds: [workspaceA, workspaceB, workspaceA], orgMemberIds: [memberId], includeDeparted, } - const filters = action ? [fixtureOnly, eq(auditLog.action, action)] : [fixtureOnly] const expected = await referenceOrder(scope, filters) - expect(expected.length).toBeGreaterThan(4) + expect(expected.length).toBeGreaterThan(0) for (const limit of [1, 3, 7, 100]) { expect(await paginate(scope, filters, limit)).toEqual(expected) @@ -164,15 +191,28 @@ describe('queryAuditLogs scope against real Postgres', () => { } ) + it('keeps an end bound to the instant, not its whole millisecond', async () => { + const scope: AuditLogListScope = { + organizationId, + orgWorkspaceIds: [workspaceA, workspaceB], + orgMemberIds: [], + includeDeparted: true, + } + const ids = await paginate(scope, { endDate: millisecond(15) }, 100) + const inScope = rows.filter((row) => ids.includes(row.id)) + expect(Math.max(...inScope.map((row) => row.index))).toBeLessThanOrEqual(60) + expect(inScope.some((row) => row.index === 60)).toBe(true) + }) + it('reads only organization-level rows when the organization has no workspaces', async () => { - const scope: OrgScopeParams = { + const scope: AuditLogListScope = { organizationId, orgWorkspaceIds: [], orgMemberIds: [], includeDeparted: true, } - const expected = await referenceOrder(scope, [fixtureOnly]) - const ids = await paginate(scope, [fixtureOnly], 5) + const expected = await referenceOrder(scope, {}) + const ids = await paginate(scope, {}, 5) expect(ids).toEqual(expected) expect( @@ -185,14 +225,30 @@ describe('queryAuditLogs scope against real Postgres', () => { ).toEqual(new Set([3, 4, 5, 6])) }) + it('leaves out a workspace the organization no longer holds', async () => { + const scope: AuditLogListScope = { + organizationId, + orgWorkspaceIds: [workspaceB], + orgMemberIds: [], + includeDeparted: true, + } + const ids = await paginate(scope, {}, 4) + + expect(ids).toEqual(await referenceOrder(scope, {})) + const workspaces = new Set( + rows.filter((row) => ids.includes(row.id)).map((row) => row.workspaceId) + ) + expect(workspaces).toEqual(new Set([workspaceB, null])) + }) + it('narrows to one workspace when the filter names it', async () => { - const scope: OrgScopeParams = { + const scope: AuditLogListScope = { organizationId, orgWorkspaceIds: [workspaceA, workspaceB], orgMemberIds: [memberId], includeDeparted: true, } - const filters = [fixtureOnly, eq(auditLog.workspaceId, workspaceB)] + const filters = { workspaceId: workspaceB } const expected = await referenceOrder(scope, filters) expect(await paginate(scope, filters, 4)).toEqual(expected) diff --git a/apps/sim/lib/audit-logs/query.integration.ts b/apps/sim/lib/audit-logs/query.integration.ts index 4e995665bbd..358b5c20524 100644 --- a/apps/sim/lib/audit-logs/query.integration.ts +++ b/apps/sim/lib/audit-logs/query.integration.ts @@ -50,16 +50,29 @@ describe('Audit pagination and drain tenant coverage in PostgreSQL', () => { }) it.each([ - { limit: 1, maxBytes: undefined, boundary: 'row limit' }, - { limit: 10, maxBytes: 1000, boundary: 'byte limit' }, + { limit: 1, maxBytes: undefined, boundary: 'row limit', filters: {} }, + { limit: 10, maxBytes: 1000, boundary: 'byte limit', filters: {} }, + { + limit: 1, + maxBytes: undefined, + boundary: 'row limit with a filter', + filters: { action: 'organization.updated' }, + }, + { + limit: 10, + maxBytes: 1000, + boundary: 'byte limit with a filter', + filters: { action: 'organization.updated' }, + }, ])( 'returns every event once across a $boundary within one PostgreSQL millisecond', - async ({ limit, maxBytes }) => { + async ({ limit, maxBytes, filters }) => { const seen: string[] = [] let cursor: string | undefined for (let pageIndex = 0; pageIndex < ids.length + 1; pageIndex++) { const page = await runtime.query.queryAuditLogs( - [runtime.inArray(runtime.auditLog.id, ids)], + { organizationId, orgWorkspaceIds: [], orgMemberIds: [], includeDeparted: true }, + filters, limit, cursor, maxBytes @@ -72,8 +85,8 @@ describe('Audit pagination and drain tenant coverage in PostgreSQL', () => { if (!page.nextCursor) break cursor = page.nextCursor } - expect(seen).toHaveLength(ids.length) - expect(new Set(seen)).toEqual(new Set(ids)) + expect(seen).toHaveLength(3) + expect(new Set(seen)).toEqual(new Set(ids.slice(0, 3))) } ) diff --git a/apps/sim/lib/audit-logs/query.ts b/apps/sim/lib/audit-logs/query.ts index 3e5ed133b99..9be8e52adec 100644 --- a/apps/sim/lib/audit-logs/query.ts +++ b/apps/sim/lib/audit-logs/query.ts @@ -5,6 +5,7 @@ import { escapeLikePattern } from '@sim/utils/string' import type { InferSelectModel } from 'drizzle-orm' import { and, desc, eq, gte, ilike, inArray, isNull, lt, lte, or, type SQL, sql } from 'drizzle-orm' import { parseUnorderedList } from '@/lib/api/cursor-binding' +import { textArrayLiteral } from '@/lib/db/arrays' type DbAuditLog = InferSelectModel @@ -49,14 +50,14 @@ export interface AuditLogFilterParams { } /** - * Compiles the caller-supplied filters into SQL conditions. + * The filters that narrow a listing to a subset of its scope: everything but the time range. * * `resourceType` is a comma-separated set, parsed through * {@link parseUnorderedList} — the same parse the v2 cursor scope fingerprints * through. Splitting it here independently is what let `file,workflow` and * `file, workflow` mean one thing to the query and another to the cursor. */ -export function buildFilterConditions(params: AuditLogFilterParams): SQL[] { +function narrowingConditions(params: AuditLogFilterParams): SQL[] { const conditions: SQL[] = [] if (params.action) conditions.push(eq(auditLog.action, params.action)) @@ -83,12 +84,21 @@ export function buildFilterConditions(params: AuditLogFilterParams): SQL[] { + const conditions: SQL[] = [] if (params.startDate) conditions.push(gte(auditLog.createdAt, new Date(params.startDate))) if (params.endDate) conditions.push(lte(auditLog.createdAt, new Date(params.endDate))) - return conditions } +/** Compiles the caller-supplied filters into SQL conditions. */ +export function buildFilterConditions(params: AuditLogFilterParams): SQL[] { + return [...narrowingConditions(params), ...timeRangeConditions(params)] +} + /** Returns the IDs of all workspaces attached to the organization. */ export async function getOrgWorkspaceIds(organizationId: string): Promise { const rows = await db @@ -105,19 +115,30 @@ export interface OrgScopeParams { includeDeparted: boolean } +function orgMetadataMatch(organizationId: string): SQL { + return sql`${auditLog.metadata}->>'organizationId' = ${organizationId}` +} + +function orgResourceMatch(organizationId: string): SQL | undefined { + return and( + eq(auditLog.resourceType, AuditResourceType.ORGANIZATION), + eq(auditLog.resourceId, organizationId) + ) +} + +function currentActorCondition(orgMemberIds: string[]): SQL { + return orgMemberIds.length > 0 + ? or(inArray(auditLog.actorId, orgMemberIds), isNull(auditLog.actorId))! + : isNull(auditLog.actorId) +} + /** Builds the tenant-boundary predicate for organization audit log access. */ export function buildOrgScopeCondition(params: OrgScopeParams): SQL { const { organizationId, orgWorkspaceIds, orgMemberIds, includeDeparted } = params const orgLevelCondition = and( isNull(auditLog.workspaceId), - or( - sql`${auditLog.metadata}->>'organizationId' = ${organizationId}`, - and( - eq(auditLog.resourceType, AuditResourceType.ORGANIZATION), - eq(auditLog.resourceId, organizationId) - ) - ) + or(orgMetadataMatch(organizationId), orgResourceMatch(organizationId)) )! const workspaceCondition = @@ -131,29 +152,121 @@ export function buildOrgScopeCondition(params: OrgScopeParams): SQL { : orgLevelCondition if (includeDeparted) return orgScope + return and(orgScope, currentActorCondition(orgMemberIds))! +} - const currentActorCondition = - orgMemberIds.length > 0 - ? or(inArray(auditLog.actorId, orgMemberIds), isNull(auditLog.actorId))! - : isNull(auditLog.actorId) - - return and(orgScope, currentActorCondition)! +/** The organization scope a listing reads, its workspaces resolved up front. */ +export interface AuditLogListScope extends OrgScopeParams { + orgWorkspaceIds: string[] } -function buildCursorCondition(cursor: string): SQL | null { - const cursorData = decodeAuditLogCursor(cursor) - if (!cursorData) return null +const auditLogTimestamp = sql`date_trunc('milliseconds', ${auditLog.createdAt})` +const auditLogOrder = [desc(auditLogTimestamp), desc(auditLog.id)] + +function buildCursorCondition(cursorData: CursorData): SQL | undefined { const cursorDate = new Date(cursorData.createdAt) - const cursorTimestamp = sql`date_trunc('milliseconds', ${auditLog.createdAt})` + return or( + lt(auditLogTimestamp, sql.param(cursorDate, auditLog.createdAt)), + and( + eq(auditLogTimestamp, sql.param(cursorDate, auditLog.createdAt)), + lt(auditLog.id, cursorData.id) + ) + ) +} + +/** + * {@link buildCursorCondition} as one row comparison, the same rows in the same order, spelled so + * an index on `(…, date_trunc('milliseconds', created_at), id)` can seek straight to the cursor. + */ +function buildCursorKeyCondition(cursorData: CursorData): SQL { + const cursorDate = sql.param(new Date(cursorData.createdAt), auditLog.createdAt) + return sql`(${auditLogTimestamp}, ${auditLog.id}) < (${cursorDate}::timestamp, ${cursorData.id})` +} + +/** + * The time range as bounds on the listing's sort key. `date_trunc` never decreases, so + * `created_at >= start` implies `date_trunc(created_at) >= date_trunc(start)` and likewise for the + * end: these bounds keep every row {@link timeRangeConditions} keeps, and let a keyset index scan + * start and stop at the range instead of filtering its way through a workspace's whole history. + */ +function timeRangeKeyBounds(params: AuditLogFilterParams): SQL[] { + const bound = (date: string) => + sql`date_trunc('milliseconds', ${sql.param(new Date(date), auditLog.createdAt)}::timestamp)` + const bounds: SQL[] = [] + if (params.startDate) bounds.push(sql`${auditLogTimestamp} >= ${bound(params.startDate)}`) + if (params.endDate) bounds.push(sql`${auditLogTimestamp} <= ${bound(params.endDate)}`) + return bounds +} + +/** + * The ids of the first `take` rows in scope, newest first, for a listing narrowed by nothing but + * a time range. The scope is a disjoint union: each organization workspace's rows, the + * organization-level rows that name the organization in their metadata, and the organization + * resource rows that do not. Each part is read newest first and limited to `take`; the first + * `take` rows of the union are exactly the first `take` of the whole scope, and disjointness keeps + * `union all` free of duplicates. + * + * A workspace's part is a keyset scan of `audit_log_workspace_created_at_id_idx` that stops after + * a page, and the metadata part one of `audit_log_org_level_created_at_id_idx`. Planned as one + * `OR`, the same listing reads and sorts every row of the scope inside the range. Only a part + * whose index serves its filter in order stops after a page, which is why a listing with any + * other filter keeps the single predicate: there the filter's own index finds the rows. + */ +function keysetPageIds( + scope: AuditLogListScope, + timeRange: AuditLogFilterParams, + cursorData: CursorData | null, + take: number +): SQL { + const shared = [ + ...(scope.includeDeparted ? [] : [currentActorCondition(scope.orgMemberIds)]), + ...timeRangeConditions(timeRange), + ...timeRangeKeyBounds(timeRange), + ...(cursorData ? [buildCursorKeyCondition(cursorData)] : []), + ] + const part = (scopePart: SQL | undefined) => + sql`(select ${auditLog.id}, ${auditLog.createdAt} from ${auditLog} where ${and(scopePart, ...shared)} order by ${auditLogTimestamp} desc, ${auditLog.id} desc limit ${take})` - return ( - or( - lt(cursorTimestamp, sql.param(cursorDate, auditLog.createdAt)), + const { organizationId } = scope + const parts = [ + part(and(isNull(auditLog.workspaceId), orgMetadataMatch(organizationId))), + part( and( - eq(cursorTimestamp, sql.param(cursorDate, auditLog.createdAt)), - lt(auditLog.id, cursorData.id) + isNull(auditLog.workspaceId), + orgResourceMatch(organizationId), + sql`(${orgMetadataMatch(organizationId)}) is not true` ) - ) ?? null + ), + ] + const workspaceIds = [...new Set(scope.orgWorkspaceIds)] + if (workspaceIds.length > 0) { + parts.unshift( + sql`(select page.id, page.created_at from unnest(${textArrayLiteral(workspaceIds)}) as org_workspace(workspace_id) cross join lateral ${part(sql`${auditLog.workspaceId} = org_workspace.workspace_id`)} as page)` + ) + } + return sql`(select candidate.id from (${sql.join(parts, sql` union all `)}) as candidate order by date_trunc('milliseconds', candidate.created_at) desc, candidate.id desc limit ${take})` +} + +/** + * Which rows a page may hold: the first `take` of the scope after the cursor. A listing narrowed + * only by time reads them part by part through {@link keysetPageIds}; any other filter keeps the + * single scope predicate, whose plan the filter's index drives. + */ +function pageCondition( + scope: AuditLogListScope, + filters: AuditLogFilterParams, + cursorData: CursorData | null, + take: number +): SQL | undefined { + const narrowing = narrowingConditions(filters) + if (narrowing.length === 0) { + return inArray(auditLog.id, keysetPageIds(scope, filters, cursorData, take)) + } + return and( + buildOrgScopeCondition(scope), + ...narrowing, + ...timeRangeConditions(filters), + cursorData ? buildCursorCondition(cursorData) : undefined ) } @@ -163,21 +276,19 @@ interface CursorPaginatedResult { truncated?: boolean } -/** An optional byte budget bounds each database page before hydration and stops at oversized rows. */ +/** + * One page of the organization's audit log matching `filters`, newest first. An optional byte + * budget bounds each database page before hydration and stops at oversized rows. + */ export async function queryAuditLogs( - conditions: SQL[], + scope: AuditLogListScope, + filters: AuditLogFilterParams, limit: number, cursor?: string, maxBytes?: number ): Promise { - const allConditions = [...conditions] - if (cursor) { - const cursorCondition = buildCursorCondition(cursor) - if (cursorCondition) allConditions.push(cursorCondition) - } - - const condition = allConditions.length > 0 ? and(...allConditions) : undefined - const orderBy = [desc(sql`date_trunc('milliseconds', ${auditLog.createdAt})`), desc(auditLog.id)] + const cursorData = cursor ? decodeAuditLogCursor(cursor) : null + const condition = pageCondition(scope, filters, cursorData, limit + 1) if (maxBytes !== undefined) { return dbReplica.transaction( async (tx) => { @@ -188,7 +299,7 @@ export async function queryAuditLogs( }) .from(auditLog) .where(condition) - .orderBy(...orderBy) + .orderBy(...auditLogOrder) .limit(limit + 1) const ids: string[] = [] let bytes = 0 @@ -208,7 +319,7 @@ export async function queryAuditLogs( .select() .from(auditLog) .where(inArray(auditLog.id, ids)) - .orderBy(...orderBy) + .orderBy(...auditLogOrder) .limit(limit) : [] const last = data.at(-1) @@ -229,7 +340,7 @@ export async function queryAuditLogs( .select() .from(auditLog) .where(condition) - .orderBy(...orderBy) + .orderBy(...auditLogOrder) .limit(limit + 1) const hasMore = rows.length > limit From 0c1f3e8865e50424fb741b6e548d6bacdb5e008e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 11:40:12 -0700 Subject: [PATCH 2/2] test(audit-logs): plan the keyset listing over a workspace written mostly by a departed member --- .../lib/audit-logs/list-plans.integration.ts | 31 +++++++++++-------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/apps/sim/lib/audit-logs/list-plans.integration.ts b/apps/sim/lib/audit-logs/list-plans.integration.ts index 75cc09bd568..e082a7aa38d 100644 --- a/apps/sim/lib/audit-logs/list-plans.integration.ts +++ b/apps/sim/lib/audit-logs/list-plans.integration.ts @@ -11,7 +11,7 @@ import { explainCaptured, type PlanNode } from '@sim/db/testing/explain' import { assertDisposableTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' import { generateId } from '@sim/utils/id' -import { eq, like, sql } from 'drizzle-orm' +import { eq, inArray, like, sql } from 'drizzle-orm' import { afterAll, beforeAll, describe, expect, it } from 'vitest' import { type AuditLogFilterParams, @@ -21,6 +21,7 @@ import { const organizationId = generateId() const memberId = generateId() +const departedId = generateId() const workspaceId = generateId() const idPrefix = `audit_plan_${generateId()}_` const workspaceIndex = 'audit_log_workspace_created_at_id_idx' @@ -82,27 +83,31 @@ const NARROWED_CASES: [string, AuditLogFilterParams][] = [ describe('organization audit-log listing plans in PostgreSQL', () => { beforeAll(async () => { assertDisposableTestDatabaseUrl(process.env.DATABASE_URL ?? '') - await db.insert(user).values({ - id: memberId, - name: 'Audit plan fixture', - email: `${memberId}@audit-plan.test`, - emailVerified: true, - createdAt: new Date(), - updatedAt: new Date(), - }) + await db.insert(user).values( + [memberId, departedId].map((id) => ({ + id, + name: 'Audit plan fixture', + email: `${id}@audit-plan.test`, + emailVerified: true, + createdAt: new Date(), + updatedAt: new Date(), + })) + ) await insertWorkspaceFixture(db, { id: workspaceId, name: 'Audit plan fixture', ownerId: memberId, billedAccountUserId: memberId, }) - // A workspace with a long history, and workspace-less rows of many other organizations, so - // collecting the scope and sorting it costs far more than a keyset scan that stops at a page. + // A workspace with a long history written mostly by a departed member, and workspace-less + // rows of many other organizations, so collecting the scope and sorting it costs far more than + // a keyset scan that stops at a page, even one that passes over departed rows to fill it. await db.execute(sql` INSERT INTO ${auditLog} (id, workspace_id, actor_id, action, resource_type, resource_id, metadata, created_at) SELECT ${idPrefix} || n, CASE WHEN n % 4 = 1 THEN ${workspaceId} END, - CASE WHEN n % 2 = 0 THEN ${memberId} END, 'organization.updated', + CASE WHEN n % 20 = 1 OR n % 2 = 0 THEN ${memberId} WHEN n % 4 = 1 THEN ${departedId} END, + 'organization.updated', CASE WHEN n % 997 = 0 THEN 'organization' ELSE 'workflow' END, CASE WHEN n % 997 = 0 THEN ${organizationId} ELSE 'resource-' || n END, jsonb_build_object('organizationId', @@ -115,7 +120,7 @@ describe('organization audit-log listing plans in PostgreSQL', () => { afterAll(async () => { await db.delete(auditLog).where(like(auditLog.id, `${idPrefix}%`)) await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) - await db.delete(user).where(eq(user.id, memberId)) + await db.delete(user).where(inArray(user.id, [memberId, departedId])) }) it.each(KEYSET_CASES)(