perf(db): serve workspace usage drill-downs index-only on usage_log (#9447)

* perf(db): serve workspace usage drill-downs index-only on usage_log

Every Workspaces drill-down read filters on the billing entity, one
workspace and a created_at window, and reads cost with user_id, source or
workflow_id. usage_log_workspace_created_at_idx held only (workspace_id,
created_at), so the planner matched the window there or on
usage_log_billing_entity_created_at_cost_idx and fetched a heap row per
match to check the rest: the workspace's rows on one, the whole
organization's on the other. 0441 dropped the compact entity/period index
that used to narrow the second through a BitmapAnd, so a large
organization's workflow ranking went from about 4.6s to 13.7s, reading
thousands of rows per row it returned.

usage_log_workspace_created_at_cost_idx replaces it: the same leading
(workspace_id, created_at), then the entity and every column the
drill-down reads. Each drill-down read is now an index-only scan bounded by
the workspace's rows in the window, and the readers that name only the
workspace (the foreign key's ON DELETE SET NULL, the ledger list filtered to
a workspace) keep the order they had. The new index is built concurrently
before the old one is dropped, so inserts maintain the same number of
indexes.

The plan integration test explains the exact SQL of the totals, series and
rankings for a workspace holding most and a small share of its
organization's usage, and fails on the old schema.

* perf(db): scope the workspace covering index's claim to created_at windows

Author: waleedlatif1

apps/sim/lib/billing/core/usage-analytics-plans.integration.ts

@@ -0,0 +1,136 @@
+/**
+ * The plans the Workspaces drill-down reads get in PostgreSQL, proven on the exact SQL the
+ * analytics queries send: every read narrowed to one workspace over a `created_at` window is an
+ * index-only scan of `usage_log_workspace_created_at_cost_idx`, bounded by that workspace's rows
+ * in the window, whether the workspace holds a sliver of its organization's usage or nearly all
+ * of it. Every other index leaves the workspace or the entity to a heap fetch per matched row.
+ */
+import { db } from '@sim/db'
+import { usageLog, user, workflow, 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, type SQL, sql } from 'drizzle-orm'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { buildUsageAnalyticsScope } from '@/lib/billing/core/usage-analytics'
+import {
+  readUsageBreakdown,
+  readUsageBreakdownOverTime,
+  readUsageTimeSeries,
+  readUsageTotals,
+} from '@/lib/billing/core/usage-analytics-queries'
+
+const COVERING_INDEX = 'usage_log_workspace_created_at_cost_idx'
+
+const member = generateId()
+const organization = { type: 'organization', id: `org-${generateId()}` } as const
+const otherOrganization = `org-${generateId()}`
+const workspaces = { dominant: generateId(), minor: generateId(), other: generateId() }
+const workflowIds = [generateId(), generateId(), generateId()]
+const anchor = new Date('2026-06-01T00:00:00Z')
+const window = {
+  kind: 'range',
+  from: new Date(anchor.getTime() - 30 * 24 * 60 * 60 * 1000),
+  to: anchor,
+} as const
+
+/**
+ * Ledger rows a minute apart over sixty days. The organization's rows go mostly to the dominant
+ * workspace and an eighth to the minor one, and an unrelated organization's workspace holds as
+ * many again, so a scan bounded by the organization or by the whole window is never the cheap
+ * one. The minor workspace also carries rows billed to its creator before it joined the
+ * organization, which the drill-down must skip.
+ */
+async function seedLedger() {
+  await db.execute(sql`
+    INSERT INTO ${usageLog} (id, user_id, category, source, description, cost,
+      billing_entity_type, billing_entity_id, billing_period_start, billing_period_end,
+      workspace_id, workflow_id, created_at)
+    SELECT 'ul_plan_' || n || '_' || ${member}, ${member}, 'fixed',
+      (ARRAY['workflow', 'copilot', 'wand']::usage_log_source[])[1 + n % 3], 'run', 0.01,
+      CASE WHEN n % 16 = 15 THEN 'user' ELSE 'organization' END::billing_entity_type,
+      CASE WHEN n % 16 = 15 THEN ${member} WHEN n % 2 = 0 THEN ${otherOrganization}
+        ELSE ${organization.id} END,
+      timestamp '2026-03-01', timestamp '2026-07-01',
+      CASE WHEN n % 2 = 0 THEN ${workspaces.other} WHEN n % 16 IN (7, 15) THEN ${workspaces.minor}
+        ELSE ${workspaces.dominant} END,
+      CASE WHEN n % 2 = 1 AND n % 5 < 3 THEN (ARRAY[${workflowIds[0]}, ${workflowIds[1]},
+        ${workflowIds[2]}])[1 + n % 3] END,
+      ${anchor.toISOString()}::timestamp - n * interval '1 minute'
+    FROM generate_series(1, 86400) AS n`)
+  await db.execute(sql`VACUUM ANALYZE ${usageLog}`)
+}
+
+const coveringScan = (nodes: PlanNode[]) =>
+  nodes.find((node) => node['Index Name'] === COVERING_INDEX)
+
+describe('Workspaces drill-down plans in PostgreSQL', () => {
+  beforeAll(async () => {
+    assertDisposableTestDatabaseUrl(process.env.DATABASE_URL ?? '')
+    const now = new Date()
+    await db.insert(user).values({
+      id: member,
+      name: 'Plan fixture',
+      email: `${member}@fixture.test`,
+      emailVerified: true,
+      createdAt: now,
+      updatedAt: now,
+    })
+    await insertWorkspaceFixture(
+      db,
+      Object.values(workspaces).map((id) => ({
+        id,
+        name: 'Plan fixture',
+        ownerId: member,
+        billedAccountUserId: member,
+      }))
+    )
+    await db.insert(workflow).values(
+      workflowIds.map((id, index) => ({
+        id,
+        userId: member,
+        workspaceId: workspaces.dominant,
+        name: `Plan fixture ${index}`,
+        lastSynced: now,
+        createdAt: now,
+        updatedAt: now,
+      }))
+    )
+    await seedLedger()
+  })
+
+  afterAll(async () => {
+    await db.delete(usageLog).where(eq(usageLog.userId, member))
+    await deleteWorkspaceFixture(db, inArray(workspace.id, Object.values(workspaces)))
+    await db.delete(user).where(eq(user.id, member))
+  })
+
+  const reads: [string, (scope: SQL[]) => Promise<unknown>][] = [
+    ['total', (scope) => readUsageTotals(scope)],
+    ['time series', (scope) => readUsageTimeSeries(scope, 'day', 'America/Los_Angeles')],
+    [
+      'source series',
+      (scope) => readUsageBreakdownOverTime(scope, 'source', 'America/Los_Angeles', 'hour'),
+    ],
+    [
+      'member series',
+      (scope) => readUsageBreakdownOverTime(scope, 'member', 'America/Los_Angeles', 'hour'),
+    ],
+    ['workflow ranking', (scope) => readUsageBreakdown(scope, 'workflow')],
+  ]
+
+  it.each(
+    (['dominant', 'minor'] as const).flatMap((share) =>
+      reads.map(([label, read]) => ({ label, share, read }))
+    )
+  )(
+    'reads the $label of the $share workspace index-only from its rows in the window',
+    async ({ share, read }) => {
+      const scope = buildUsageAnalyticsScope(organization, window, workspaces[share])
+      const scan = coveringScan(await explainCaptured(db, () => read(scope)))
+      expect(scan?.['Node Type']).toBe('Index Only Scan')
+      expect(scan?.['Index Cond']).toMatch(/^\(\(workspace_id = .+\) AND \(created_at >= .+\)/)
+    }
+  )
+})

apps/sim/lib/billing/core/usage-analytics-queries.ts

@@ -27,7 +27,8 @@ import type { DbClient } from '@/lib/db/types'
  * and the member/source breakdowns are index-only. `workspace_id`, `workflow_id`,
  * `description`, `metadata`, and `execution_id` are NOT in it, so the remaining
  * reads heap-fetch per row — which is why they are issued only for the dimension
- * actually being viewed.
+ * actually being viewed. A read narrowed to one workspace runs on
+ * `usage_log_workspace_created_at_cost_idx` instead, which also covers `workflow_id`.
  */
 
 export interface UsageTimeSeriesRow {

apps/sim/lib/billing/core/usage-analytics.ts

@@ -87,9 +87,10 @@ export function buildUsageAnalyticsScope(
    * entity predicates above are — so a workspace belonging to another organization
    * narrows to nothing rather than disclosing anything.
    *
-   * `workspace_id` is not in `usage_log_billing_entity_created_at_cost_idx`, so a
-   * narrowed read heap-fetches per row in the window. That is the cost the Workspaces
-   * tab already pays to rank its list, and it is only ever paid inside a drill-down.
+   * A narrowed read over a `created_at` window is index-only on
+   * `usage_log_workspace_created_at_cost_idx`, which carries the entity and every
+   * column the series and the member, source and workflow rankings read, so it scans
+   * the workspace's rows in the window rather than the organization's.
    */
   workspaceId?: string
 ): SQL[] {

packages/db/migrations/0446_usage_log_workspace_created_at_cost_idx.sql

@@ -0,0 +1,44 @@
+-- Replaces usage_log_workspace_created_at_idx with a covering index under the same two leading
+-- columns.
+--
+-- A Workspaces drill-down over a created_at window (a range, or a reporting period) filters on the
+-- billing entity, the workspace and that window, and reads cost with user_id, source or
+-- workflow_id. The old index held only
+-- (workspace_id, created_at), so a drill-down matched the window there or on
+-- usage_log_billing_entity_created_at_cost_idx and fetched every matched row from the heap to
+-- check the rest: the workspace's rows on one, the whole organization's on the other. Dropping
+-- usage_log_billing_entity_period_idx in 0441 took away the bitmap that used to narrow the second.
+-- With the entity and every column those reads return in the key, each one is an index-only scan
+-- bounded by the workspace's rows in the window. A drill-down over a Stripe period matches the
+-- period stamps instead, which this index does not carry, so it keeps the plans it had.
+--
+-- The new key extends the old one, so it serves every read the old index served in the same
+-- order: the workspace_id foreign key's ON DELETE SET NULL, and the ledger list filtered to a
+-- workspace. The old index is dropped once the new one is built, so ledger inserts maintain the
+-- same number of indexes.
+--
+-- Replay-safety: usage_log is a hot append-only ledger, so the build and the drop are
+-- 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 as is.
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('usage_log_workspace_created_at_cost_idx_failed')
+      AND (indisvalid OR indrelid <> to_regclass('usage_log'))
+  ) THEN
+    RAISE EXCEPTION 'Refusing to drop unexpected recovery index usage_log_workspace_created_at_cost_idx_failed';
+  END IF;
+  IF EXISTS (
+    SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('usage_log_workspace_created_at_cost_idx')
+      AND indrelid = to_regclass('usage_log') AND NOT indisvalid
+  ) THEN
+    ALTER INDEX "usage_log_workspace_created_at_cost_idx" RENAME TO "usage_log_workspace_created_at_cost_idx_failed";
+  END IF;
+END $$;--> statement-breakpoint
+COMMIT;--> statement-breakpoint
+SET lock_timeout = 0;--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "usage_log_workspace_created_at_cost_idx_failed";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "usage_log_workspace_created_at_cost_idx" ON "usage_log" USING btree ("workspace_id","created_at","billing_entity_type","billing_entity_id","user_id","source","cost","workflow_id");--> statement-breakpoint
+DROP INDEX CONCURRENTLY IF EXISTS "usage_log_workspace_created_at_idx";--> statement-breakpoint
+SET lock_timeout = '5s';