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

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';