diff --git a/apps/sim/app/api/v1/audit-logs/route.ts b/apps/sim/app/api/v1/audit-logs/route.ts index 15ccf719952..26498d298a8 100644 --- a/apps/sim/app/api/v1/audit-logs/route.ts +++ b/apps/sim/app/api/v1/audit-logs/route.ts @@ -25,7 +25,12 @@ 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, getOrgWorkspaceIds, queryAuditLogs } from '@/lib/audit-logs/query' +import { + buildFilterConditions, + buildOrgScopeCondition, + 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' @@ -94,12 +99,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => { ) } - const scope = { + const scopeCondition = buildOrgScopeCondition({ organizationId, orgWorkspaceIds, orgMemberIds, includeDeparted: params.includeDeparted, - } + }) const filterConditions = buildFilterConditions({ action: params.action, resourceType: params.resourceType, @@ -111,8 +116,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { }) const { data, nextCursor } = await queryAuditLogs( - scope, - filterConditions, + [scopeCondition, ...filterConditions], 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 cc659b94e49..ce0507f2989 100644 --- a/apps/sim/lib/audit-logs/application/export-audit-logs.ts +++ b/apps/sim/lib/audit-logs/application/export-audit-logs.ts @@ -5,6 +5,7 @@ import { AUDIT_LOG_CSV_HEADER, toAuditLogCsvRow } from '@/lib/audit-logs/csv' import { type AuditLogFilterParams, buildFilterConditions, + buildOrgScopeCondition, getOrgWorkspaceIds, queryAuditLogs, } from '@/lib/audit-logs/query' @@ -36,13 +37,13 @@ export const exportAuditLogs = defineAuthorizedAuditLogUseCase({ ) { throw new OrchestrationError('validation', 'actorId is not a member of your organization') } - const scope = { + const scope = buildOrgScopeCondition({ organizationId: context.organizationId, orgWorkspaceIds: workspaceIds, orgMemberIds: context.orgMemberIds, includeDeparted: input.includeDeparted, - } - const conditions = buildFilterConditions(input.filters) + }) + const conditions = [scope, ...buildFilterConditions(input.filters)] const lines = [AUDIT_LOG_CSV_HEADER] let rowCount = 0 let cursor: string | undefined @@ -51,7 +52,6 @@ export const exportAuditLogs = defineAuthorizedAuditLogUseCase({ pages: while (rowCount < EXPORT_MAX_ROWS && bytes < EXPORT_MAX_BYTES) { request?.signal?.throwIfAborted() const page = await queryAuditLogs( - scope, conditions, Math.min(EXPORT_PAGE_ROWS, EXPORT_MAX_ROWS - rowCount), cursor, 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 65164c6f833..97805b2b503 100644 --- a/apps/sim/lib/audit-logs/application/list-audit-logs.ts +++ b/apps/sim/lib/audit-logs/application/list-audit-logs.ts @@ -3,6 +3,7 @@ import { auditLogOperations } from '@/lib/audit-logs/application/operations' import { type AuditLogFilterParams, buildFilterConditions, + buildOrgScopeCondition, decodeAuditLogCursor, getOrgWorkspaceIds, queryAuditLogs, @@ -31,12 +32,12 @@ 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 scope = { + const scopeCondition = buildOrgScopeCondition({ organizationId: context.organizationId, orgWorkspaceIds, orgMemberIds: context.orgMemberIds, includeDeparted: input.includeDeparted, - } + }) if ( context.workspaceId && input.filters.workspaceId && @@ -47,6 +48,6 @@ export const listAuditLogs = defineAuthorizedAuditLogUseCase({ ...input.filters, ...(context.workspaceId ? { workspaceId: context.workspaceId } : {}), }) - return queryAuditLogs(scope, filterConditions, input.limit, input.cursor) + return queryAuditLogs([scopeCondition, ...filterConditions], input.limit, input.cursor) }, }) diff --git a/apps/sim/lib/audit-logs/list-plans.integration.ts b/apps/sim/lib/audit-logs/list-plans.integration.ts deleted file mode 100644 index 9124a2e3251..00000000000 --- a/apps/sim/lib/audit-logs/list-plans.integration.ts +++ /dev/null @@ -1,120 +0,0 @@ -/** - * The plan an organization's audit-log listing must get, proven on the exact SQL - * `queryAuditLogs` sends against a table with many workspace-less rows of other organizations: - * the organization-level metadata part is read through `audit_log_org_level_created_at_id_idx`, - * which can stream it in keyset order and stop after a page, and never by seeking - * `workspace_id IS NULL` and filtering every workspace-less row in the table. - */ -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 AuditLogListScope, queryAuditLogs } from '@/lib/audit-logs/query' - -const organizationId = generateId() -const memberId = generateId() -const workspaceId = generateId() -const idPrefix = `audit_plan_${generateId()}_` -const orgLevelIndex = 'audit_log_org_level_created_at_id_idx' - -const cursor = Buffer.from( - JSON.stringify({ createdAt: '2026-01-02T00:00:00.000Z', id: `${idPrefix}~` }) -).toString('base64') - -const listingPlan = ( - includeDeparted: boolean, - pageCursor: string | undefined, - gucs?: Record -) => { - const scope: AuditLogListScope = { - organizationId, - orgWorkspaceIds: [workspaceId], - orgMemberIds: [memberId], - includeDeparted, - } - return explainCaptured(db, () => queryAuditLogs(scope, [], 50, pageCursor), { gucs }) -} - -const scansOf = (nodes: PlanNode[], index: string) => - nodes.filter((node) => node['Index Name'] === index) - -const CASES = [ - ['current members', false, undefined], - ['departed members included', true, undefined], - ['current members, after a cursor', false, cursor], - ['departed members included, after a cursor', true, cursor], -] as const - -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, - }) - // Workspace-less rows of many other organizations across the whole history, and a few - // hundred of this organization's, so a scan that filters workspace-less rows for the - // organization costs far more than one that reads only the organization's rows. - 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 % 10 = 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(CASES)( - 'reads organization-level metadata rows through their index, never every workspace-less row (%s)', - async (_label, includeDeparted, pageCursor) => { - const nodes = await listingPlan(includeDeparted, pageCursor) - const orgLevel = scansOf(nodes, orgLevelIndex) - expect(orgLevel).toHaveLength(1) - expect(orgLevel[0]['Index Cond']).toMatch(/'organizationId'/) - expect(nodes.filter((node) => /workspace_id IS NULL/.test(node['Index Cond'] ?? ''))).toEqual( - [] - ) - } - ) - - /** - * Whether the planner streams the part or collects and sorts it is a cost choice: the - * expression has no statistics (a partial index's are not used), so the estimate grows with - * the table. What must hold is that the index can serve the part in keyset order, so that on a - * large table the scan stops after a page. - */ - it.each(CASES)( - 'can stream organization-level metadata rows newest first from their index (%s)', - async (_label, includeDeparted, pageCursor) => { - const nodes = await listingPlan(includeDeparted, pageCursor, { enable_bitmapscan: 'off' }) - const [orgLevel] = scansOf(nodes, orgLevelIndex) - expect(orgLevel).toMatchObject({ 'Node Type': 'Index Scan', 'Scan Direction': 'Backward' }) - if (pageCursor) expect(orgLevel['Index Cond']).toMatch(/ROW\(date_trunc/) - } - ) -}) diff --git a/apps/sim/lib/audit-logs/list-scope.integration.ts b/apps/sim/lib/audit-logs/list-scope.integration.ts index 5bfc4edebfe..2de211a2e4c 100644 --- a/apps/sim/lib/audit-logs/list-scope.integration.ts +++ b/apps/sim/lib/audit-logs/list-scope.integration.ts @@ -4,11 +4,7 @@ import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/ import { generateId } from '@sim/utils/id' import { and, desc, eq, inArray, type SQL, sql } from 'drizzle-orm' import { afterAll, beforeAll, describe, expect, it } from 'vitest' -import { - type AuditLogListScope, - buildOrgScopeCondition, - queryAuditLogs, -} from '@/lib/audit-logs/query' +import { buildOrgScopeCondition, type OrgScopeParams, queryAuditLogs } from '@/lib/audit-logs/query' const organizationId = generateId() const otherOrganizationId = generateId() @@ -68,7 +64,7 @@ const fixtureIds = rows.map((row) => row.id) 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: AuditLogListScope, filters: SQL[]) { +async function referenceOrder(scope: OrgScopeParams, filters: SQL[]) { const matching = await db .select({ id: auditLog.id }) .from(auditLog) @@ -78,7 +74,7 @@ async function referenceOrder(scope: AuditLogListScope, filters: SQL[]) } async function paginate( - scope: AuditLogListScope, + scope: OrgScopeParams, filters: SQL[], limit: number, maxBytes?: number @@ -86,7 +82,12 @@ async function paginate( const seen: string[] = [] let cursor: string | undefined for (let page = 0; page <= rows.length; page++) { - const result = await queryAuditLogs(scope, filters, limit, cursor, maxBytes) + const result = await queryAuditLogs( + [buildOrgScopeCondition(scope), ...filters], + limit, + cursor, + maxBytes + ) expect(result.truncated ?? false).toBe(false) seen.push(...result.data.map((row) => row.id)) if (!result.nextCursor) return seen @@ -146,7 +147,7 @@ describe('queryAuditLogs scope against real Postgres', () => { ])( 'pages through exactly the single-predicate result (%o)', async ({ includeDeparted, action }) => { - const scope: AuditLogListScope = { + const scope: OrgScopeParams = { organizationId, orgWorkspaceIds: [workspaceA, workspaceB, workspaceA], orgMemberIds: [memberId], @@ -164,7 +165,7 @@ describe('queryAuditLogs scope against real Postgres', () => { ) it('reads only organization-level rows when the organization has no workspaces', async () => { - const scope: AuditLogListScope = { + const scope: OrgScopeParams = { organizationId, orgWorkspaceIds: [], orgMemberIds: [], @@ -185,7 +186,7 @@ describe('queryAuditLogs scope against real Postgres', () => { }) it('narrows to one workspace when the filter names it', async () => { - const scope: AuditLogListScope = { + const scope: OrgScopeParams = { organizationId, orgWorkspaceIds: [workspaceA, workspaceB], orgMemberIds: [memberId], diff --git a/apps/sim/lib/audit-logs/query.integration.ts b/apps/sim/lib/audit-logs/query.integration.ts index f149114a2f6..4e995665bbd 100644 --- a/apps/sim/lib/audit-logs/query.integration.ts +++ b/apps/sim/lib/audit-logs/query.integration.ts @@ -53,13 +53,12 @@ describe('Audit pagination and drain tenant coverage in PostgreSQL', () => { { limit: 1, maxBytes: undefined, boundary: 'row limit' }, { limit: 10, maxBytes: 1000, boundary: 'byte limit' }, ])( - 'returns every organization event once across a $boundary within one PostgreSQL millisecond', + 'returns every event once across a $boundary within one PostgreSQL millisecond', async ({ limit, maxBytes }) => { const seen: string[] = [] let cursor: string | undefined for (let pageIndex = 0; pageIndex < ids.length + 1; pageIndex++) { const page = await runtime.query.queryAuditLogs( - { organizationId, orgWorkspaceIds: [], orgMemberIds: [], includeDeparted: true }, [runtime.inArray(runtime.auditLog.id, ids)], limit, cursor, @@ -73,7 +72,8 @@ describe('Audit pagination and drain tenant coverage in PostgreSQL', () => { if (!page.nextCursor) break cursor = page.nextCursor } - expect(seen).toEqual(ids.slice(0, 3).reverse()) + expect(seen).toHaveLength(ids.length) + expect(new Set(seen)).toEqual(new Set(ids)) } ) diff --git a/apps/sim/lib/audit-logs/query.ts b/apps/sim/lib/audit-logs/query.ts index bb0833a6e65..3e5ed133b99 100644 --- a/apps/sim/lib/audit-logs/query.ts +++ b/apps/sim/lib/audit-logs/query.ts @@ -3,9 +3,8 @@ import { db, dbReplica } from '@sim/db' import { auditLog, workspace } from '@sim/db/schema' import { escapeLikePattern } from '@sim/utils/string' import type { InferSelectModel } from 'drizzle-orm' -import { and, desc, eq, gte, ilike, inArray, isNull, lte, or, type SQL, sql } 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 @@ -106,138 +105,66 @@ export interface OrgScopeParams { includeDeparted: boolean } -/** - * An organization-level row has no workspace and names the organization in its metadata, or is - * the organization resource itself: {@link orgLevelCondition}, read by a listing as - * {@link orgLevelParts}. - */ -function orgMetadataMatch(organizationId: string): SQL { - return sql`${auditLog.metadata}->>'organizationId' = ${organizationId}` -} - -function orgResourceMatch(organizationId: string): SQL[] { - return [ - eq(auditLog.resourceType, AuditResourceType.ORGANIZATION), - eq(auditLog.resourceId, organizationId), - ] -} - -function orgLevelCondition(organizationId: string): SQL { - return and( - isNull(auditLog.workspaceId), - or(orgMetadataMatch(organizationId), and(...orgResourceMatch(organizationId))) - )! -} - -/** - * {@link orgLevelCondition} as two disjoint parts, each with its own index. The metadata match, - * which can hold thousands of rows, is read through `audit_log_org_level_created_at_id_idx`, which - * can stream it in keyset order and stop after a page: it spells `workspace_id IS NULL` and a strict equality on the key, which proves that - * partial index's predicate. The resource match whose metadata does not already name the - * organization is a few rows, sought on `audit_log_resource_idx` and sorted. - */ -function orgLevelParts(organizationId: string): SQL[] { - return [ - conjunction([isNull(auditLog.workspaceId), orgMetadataMatch(organizationId)]), - conjunction([ - isNull(auditLog.workspaceId), - ...orgResourceMatch(organizationId), - sql`(${orgMetadataMatch(organizationId)}) is not true`, - ]), - ] -} - -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 orgLevel = orgLevelCondition(organizationId) + const orgLevelCondition = and( + isNull(auditLog.workspaceId), + or( + sql`${auditLog.metadata}->>'organizationId' = ${organizationId}`, + and( + eq(auditLog.resourceType, AuditResourceType.ORGANIZATION), + eq(auditLog.resourceId, organizationId) + ) + ) + )! + const workspaceCondition = orgWorkspaceIds === undefined ? sql`exists (select 1 from ${workspace} where ${workspace.id} = ${auditLog.workspaceId} and ${workspace.organizationId} = ${organizationId})` : orgWorkspaceIds.length > 0 ? inArray(auditLog.workspaceId, orgWorkspaceIds) : undefined - const orgScope = workspaceCondition ? or(workspaceCondition, orgLevel)! : orgLevel + const orgScope = workspaceCondition + ? or(workspaceCondition, orgLevelCondition)! + : orgLevelCondition if (includeDeparted) return orgScope - return and(orgScope, currentActorCondition(orgMemberIds))! -} -/** The organization scope a listing reads: its workspace ids resolved up front. */ -export interface AuditLogListScope extends OrgScopeParams { - orgWorkspaceIds: string[] -} + const currentActorCondition = + orgMemberIds.length > 0 + ? or(inArray(auditLog.actorId, orgMemberIds), isNull(auditLog.actorId))! + : isNull(auditLog.actorId) -const auditLogTimestamp = sql`date_trunc('milliseconds', ${auditLog.createdAt})` -const auditLogOrder = [desc(auditLogTimestamp), desc(auditLog.id)] + return and(orgScope, currentActorCondition)! +} -/** - * Keyset position after the cursor. A row comparison is the same order as the - * `ORDER BY` below, and it lets an index scan seek straight to the cursor - * rather than walk every newer row and discard it. - */ function buildCursorCondition(cursor: string): SQL | null { const cursorData = decodeAuditLogCursor(cursor) if (!cursorData) return null - const cursorDate = sql.param(new Date(cursorData.createdAt), auditLog.createdAt) - return sql`(${auditLogTimestamp}, ${auditLog.id}) < (${cursorDate}::timestamp, ${cursorData.id})` -} - -function conjunction(conditions: SQL[]): SQL { - return sql.join( - conditions.map((condition) => sql`(${condition})`), - sql` and ` + const cursorDate = new Date(cursorData.createdAt) + const cursorTimestamp = sql`date_trunc('milliseconds', ${auditLog.createdAt})` + + return ( + or( + lt(cursorTimestamp, sql.param(cursorDate, auditLog.createdAt)), + and( + eq(cursorTimestamp, sql.param(cursorDate, auditLog.createdAt)), + lt(auditLog.id, cursorData.id) + ) + ) ?? null ) } -/** - * The ids of the first `take` rows in scope, newest first. The scope is a disjoint union, read - * part by part, each part limited to `take`: one keyset scan of - * `audit_log_workspace_created_at_id_idx` per organization workspace, and the two - * {@link orgLevelParts}. 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 part is cheap only when an - * index serves its filter in order, so the scan stops after a page, or its filter matches few - * rows; otherwise it reads and sorts every row it matches. Planned as one `OR`, that is the whole - * scope (for a busy organization, a full scan of the table); with only `workspace_id IS NULL` - * indexed, it is every workspace-less row in the table, spread across the heap. - */ -function pageIdsQuery(scope: AuditLogListScope, conditions: SQL[], take: number): SQL { - const shared = scope.includeDeparted - ? conditions - : [...conditions, currentActorCondition(scope.orgMemberIds)] - const branch = (part: SQL) => - sql`(select ${auditLog.id}, ${auditLog.createdAt} from ${auditLog} where ${conjunction([part, ...shared])} order by ${auditLogTimestamp} desc, ${auditLog.id} desc limit ${take})` - - const parts = orgLevelParts(scope.organizationId).map(branch) - 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 ${branch(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})` -} - interface CursorPaginatedResult { data: DbAuditLog[] nextCursor?: string truncated?: boolean } -/** - * One page of the organization's audit log matching `conditions`, newest first. - * An optional byte budget bounds each database page before hydration and stops - * at oversized rows. - */ +/** An optional byte budget bounds each database page before hydration and stops at oversized rows. */ export async function queryAuditLogs( - scope: AuditLogListScope, conditions: SQL[], limit: number, cursor?: string, @@ -248,8 +175,9 @@ export async function queryAuditLogs( const cursorCondition = buildCursorCondition(cursor) if (cursorCondition) allConditions.push(cursorCondition) } - const inPage = inArray(auditLog.id, pageIdsQuery(scope, allConditions, limit + 1)) + const condition = allConditions.length > 0 ? and(...allConditions) : undefined + const orderBy = [desc(sql`date_trunc('milliseconds', ${auditLog.createdAt})`), desc(auditLog.id)] if (maxBytes !== undefined) { return dbReplica.transaction( async (tx) => { @@ -259,8 +187,8 @@ export async function queryAuditLogs( bytes: sql`octet_length(row_to_json(${auditLog})::text)`.mapWith(Number), }) .from(auditLog) - .where(inPage) - .orderBy(...auditLogOrder) + .where(condition) + .orderBy(...orderBy) .limit(limit + 1) const ids: string[] = [] let bytes = 0 @@ -280,7 +208,7 @@ export async function queryAuditLogs( .select() .from(auditLog) .where(inArray(auditLog.id, ids)) - .orderBy(...auditLogOrder) + .orderBy(...orderBy) .limit(limit) : [] const last = data.at(-1) @@ -300,8 +228,8 @@ export async function queryAuditLogs( const rows = await dbReplica .select() .from(auditLog) - .where(inPage) - .orderBy(...auditLogOrder) + .where(condition) + .orderBy(...orderBy) .limit(limit + 1) const hasMore = rows.length > limit