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..e082a7aa38d --- /dev/null +++ b/apps/sim/lib/audit-logs/list-plans.integration.ts @@ -0,0 +1,148 @@ +/** + * 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, inArray, 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 departedId = 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( + [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 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 % 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', + 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(inArray(user.id, [memberId, departedId])) + }) + + 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