perf(audit-logs): index organization-level rows and read them as two keyset parts

The organization listing reads its organization-level rows as one keyset
branch: workspace_id IS NULL and (metadata organizationId = org, or the
organization resource itself). No index served that branch, so the planner
seeked workspace_id IS NULL and filtered every workspace-less row in the
table, spread across the heap. An organization without a page of such rows
near the top paid for all of them on every request.

A partial expression index on (metadata->>'organizationId', the millisecond
created_at, id) over workspace-less rows that carry the key now serves the
metadata match, and the branch is split into two disjoint parts so that
part can be read in keyset order and stop after a page: the metadata match,
and the organization-resource match whose metadata does not already name the
organization (a few rows, on audit_log_resource_idx). Disjointness keeps the
union exact; buildOrgScopeCondition keeps its predicate.

The index alone is not enough: under the single OR the planner can only
BitmapOr the two indexes and sort, which reads every organization-level row
of the organization.

The existing list-scope suite compared results only, which the slow plan
returns correctly; a plan suite now asserts the metadata part is read
through the new index, never by a workspace_id IS NULL seek, and that the
index can stream it backward in keyset order. The scope suite gains rows
matching both parts, only the resource part, and another organization.

Author: waleedlatif1

apps/sim/lib/audit-logs/list-plans.integration.ts

@@ -0,0 +1,120 @@
+/**
+ * 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<string, string>
+) => {
+  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/)
+    }
+  )
+})

apps/sim/lib/audit-logs/list-scope.integration.ts

@@ -21,29 +21,44 @@ const foreignWorkspace = generateId()
 const workspaceIds = [workspaceA, workspaceB, foreignWorkspace]
 
 type Placement = { workspaceId: string | null; resourceId: string; metadata: object }
+/**
+ * Organization-level rows reach the listing through two disjoint parts, so the placements cover
+ * each part alone, a row both parts would match (it must be read once), a resource row whose
+ * metadata names another organization (only the resource part matches it), and rows of another
+ * organization that neither part may read.
+ */
 const PLACEMENTS: Placement[] = [
   { workspaceId: workspaceA, resourceId: generateId(), metadata: {} },
   { workspaceId: workspaceB, resourceId: generateId(), metadata: {} },
   { workspaceId: foreignWorkspace, resourceId: generateId(), metadata: {} },
   { workspaceId: null, resourceId: generateId(), metadata: { organizationId } },
   { workspaceId: null, resourceId: organizationId, metadata: {} },
+  { workspaceId: null, resourceId: organizationId, metadata: { organizationId } },
+  {
+    workspaceId: null,
+    resourceId: organizationId,
+    metadata: { organizationId: otherOrganizationId },
+  },
   {
     workspaceId: null,
     resourceId: generateId(),
     metadata: { organizationId: otherOrganizationId },
   },
+  { workspaceId: null, resourceId: otherOrganizationId, metadata: {} },
 ]
 const ACTORS = [memberId, departedId, null]
 
-const rows = Array.from({ length: 72 }, (_, index) => {
+const rows = Array.from({ length: 81 }, (_, index) => {
   const placement = PLACEMENTS[index % PLACEMENTS.length]
   return {
     id: `${generateId()}-${String(index).padStart(2, '0')}`,
     index,
     workspaceId: placement.workspaceId,
     actorId: ACTORS[Math.floor(index / PLACEMENTS.length) % ACTORS.length],
     action: index % 4 === 0 ? 'workflow.deployed' : 'workflow.updated',
-    resourceType: placement.resourceId === organizationId ? 'organization' : 'workflow',
+    resourceType: [organizationId, otherOrganizationId].includes(placement.resourceId)
+      ? 'organization'
+      : 'workflow',
     resourceId: placement.resourceId,
     metadata: { ...placement.metadata, fixture: 'x'.repeat(200) },
   }
@@ -162,6 +177,11 @@ describe('queryAuditLogs scope against real Postgres', () => {
     expect(
       new Set(rows.filter((row) => ids.includes(row.id)).map((row) => row.workspaceId))
     ).toEqual(new Set([null]))
+    expect(
+      new Set(
+        rows.filter((row) => ids.includes(row.id)).map((row) => row.index % PLACEMENTS.length)
+      )
+    ).toEqual(new Set([3, 4, 5, 6]))
   })
 
   it('narrows to one workspace when the filter names it', async () => {

apps/sim/lib/audit-logs/query.ts

@@ -106,19 +106,47 @@ 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<unknown> {
+  return sql`${auditLog.metadata}->>'organizationId' = ${organizationId}`
+}
+
+function orgResourceMatch(organizationId: string): SQL<unknown>[] {
+  return [
+    eq(auditLog.resourceType, AuditResourceType.ORGANIZATION),
+    eq(auditLog.resourceId, organizationId),
+  ]
+}
+
 function orgLevelCondition(organizationId: string): SQL<unknown> {
   return and(
     isNull(auditLog.workspaceId),
-    or(
-      sql`${auditLog.metadata}->>'organizationId' = ${organizationId}`,
-      and(
-        eq(auditLog.resourceType, AuditResourceType.ORGANIZATION),
-        eq(auditLog.resourceId, organizationId)
-      )
-    )
+    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<unknown>[] {
+  return [
+    conjunction([isNull(auditLog.workspaceId), orgMetadataMatch(organizationId)]),
+    conjunction([
+      isNull(auditLog.workspaceId),
+      ...orgResourceMatch(organizationId),
+      sql`(${orgMetadataMatch(organizationId)}) is not true`,
+    ]),
+  ]
+}
+
 function currentActorCondition(orgMemberIds: string[]): SQL<unknown> {
   return orgMemberIds.length > 0
     ? or(inArray(auditLog.actorId, orgMemberIds), isNull(auditLog.actorId))!
@@ -170,15 +198,15 @@ function conjunction(conditions: SQL<unknown>[]): SQL<unknown> {
 }
 
 /**
- * The ids of the first `take` rows in scope, newest first. The scope is a
- * disjoint union — a row either belongs to one of the organization's workspaces
- * or has no workspace and names the organization — so each part is read
- * separately: one keyset scan of `audit_log_workspace_created_at_id_idx` per
- * workspace, plus the organization-level rows, each stopping after `take`
- * matches. The first `take` rows of the union are exactly the first `take` of
- * the whole scope. Planned as one `OR`, Postgres can follow no index in order
- * and sorts every row in the scope (for a busy organization, a full scan of the
- * table) to return one page.
+ * 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<unknown>[], take: number): SQL {
   const shared = scope.includeDeparted
@@ -187,7 +215,7 @@ function pageIdsQuery(scope: AuditLogListScope, conditions: SQL<unknown>[], take
   const branch = (part: SQL<unknown>) =>
     sql`(select ${auditLog.id}, ${auditLog.createdAt} from ${auditLog} where ${conjunction([part, ...shared])} order by ${auditLogTimestamp} desc, ${auditLog.id} desc limit ${take})`
 
-  const parts = [branch(orgLevelCondition(scope.organizationId))]
+  const parts = orgLevelParts(scope.organizationId).map(branch)
   const workspaceIds = [...new Set(scope.orgWorkspaceIds)]
   if (workspaceIds.length > 0) {
     parts.unshift(

packages/db/migrations/0444_audit_log_org_level_created_at_id_idx.sql

@@ -0,0 +1,35 @@
+-- The organization-level rows of an organization's audit log, newest first. A listing reads them
+-- as one keyset branch (workspace_id IS NULL, metadata organizationId equal to the organization,
+-- ordered by the millisecond-truncated created_at and id), and no index served that order: the
+-- planner fetched every workspace-less row, which are spread across the whole heap, and filtered
+-- them for the organization, so an organization with few such rows near the top paid for all of
+-- them on every page.
+--
+-- The partial predicate is what that branch states: equality on the metadata key is strict, so it
+-- implies the IS NOT NULL, and the branch spells workspace_id IS NULL literally. Rows attributed
+-- only by resource (resource_type 'organization') stay on audit_log_resource_idx; there are few.
+--
+-- Replay-safety: audit_log takes writes continuously, so the build is CONCURRENTLY, below an
+-- embedded COMMIT, and a failure there replays the whole file from the top. A failed concurrent
+-- build leaves an INVALID index that IF NOT EXISTS would keep: the replay moves only that one aside
+-- to be dropped, so a valid index built by an earlier attempt is kept rather than rebuilt.
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('audit_log_org_level_created_at_id_idx_failed')
+      AND (indisvalid OR indrelid <> to_regclass('audit_log'))
+  ) THEN
+    RAISE EXCEPTION 'Refusing to drop unexpected recovery index audit_log_org_level_created_at_id_idx_failed';
+  END IF;
+  IF EXISTS (
+    SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('audit_log_org_level_created_at_id_idx')
+      AND indrelid = to_regclass('audit_log') AND NOT indisvalid
+  ) THEN
+    ALTER INDEX "audit_log_org_level_created_at_id_idx" RENAME TO "audit_log_org_level_created_at_id_idx_failed";
+  END IF;
+END $$;--> statement-breakpoint
+COMMIT;--> statement-breakpoint
+SET lock_timeout = 0;--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "audit_log_org_level_created_at_id_idx_failed";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "audit_log_org_level_created_at_id_idx" ON "audit_log" USING btree (("metadata"->>'organizationId'),date_trunc('milliseconds', "created_at"),"id") WHERE "audit_log"."workspace_id" IS NULL AND ("audit_log"."metadata"->>'organizationId') IS NOT NULL;--> statement-breakpoint
+SET lock_timeout = '5s';