perf(webapp,database): index RuntimeEnvironment.pauseSource for the billing-limit reconcile tick (#4590)

## What

The `billingLimit.reconcileTick` worker calls
`getOrgIdsWithBillingPauseSource()` on
`BILLING_LIMIT_RECONCILE_INTERVAL_MS` (~every 90s) to find which orgs
currently have billing-limit-paused environments. Two problems:

1. `RuntimeEnvironment.pauseSource` had no index, so `WHERE pauseSource
= 'BILLING_LIMIT'` was a **sequential scan of the whole table** on the
control-plane primary, every tick.
2. Prisma `distinct` dedups **after** fetching, so it read every paused
row (thousands) to produce a handful of distinct org ids.

This PR:

- Adds a **partial index** on `RuntimeEnvironment (pauseSource,
organizationId) WHERE pauseSource IS NOT NULL`. Nearly all rows have
`pauseSource = null`, so the index stays tiny. Second column lets the DB
satisfy the distinct-org lookup from the index. Defined in SQL (Prisma
can't express partial indexes), matching the existing partial-unique
indexes on this model.
- Switches the query from `findMany({ distinct })` to
`groupBy(["organizationId"])`, pushing DISTINCT into the DB so it
returns only the distinct orgs.

## Evidence

**Correctness** — colocated `postgresTest` (testcontainers, no mocks):
multiple `BILLING_LIMIT` envs in one org collapse to one org id,
`pauseSource = null` envs are excluded, each org id returned once. 5/5
tests in `billingLimitReconciliation.test.ts` pass.

**Plan change** — `EXPLAIN ANALYZE` on a synthetic table (200k rows,
5,250 `BILLING_LIMIT` across ~40 orgs, mirroring the test-side numbers
from the investigation):

| | Before (no index) | After (partial index) |
|---|---|---|
| Plan | Seq Scan (194,750 rows removed by filter) | Bitmap Index Scan
on partial index |
| Buffers | 1355 | 51 (index 6 + heap 45) |
| Exec time | 6.06 ms | 0.59 ms |

Index size 56 kB vs table 11 MB. The key win: cost now scales with the
paused-env count, not total table size, which matters most on prod where
the table is far larger.

## Rollout & rollback

- **Index**: `CREATE INDEX CONCURRENTLY IF NOT EXISTS`, in its own
migration file. Pre-apply the index manually on the control-plane
primary before deploying the migration (the migration is a no-op if the
index already exists).
- **Query change** is behavior-equivalent (same distinct org set), so no
flag needed.
- **Rollback**: revert the deploy and drop the index. No data migration
either direction.

## Notes / limitations

- The planner uses a Bitmap Heap Scan, so `organizationId` is still read
from the heap (45 blocks for the matched rows only, not the whole
table). A pure index-only scan isn't chosen for the bitmap path; the
second index column keeps that open for the index-scan path at
negligible cost.

refs TRI-13169

.server-changes/billing-limit-reconcile-index.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Reduced recurring background database load from the billing-limit recovery check, so paused environments are reconciled with less overhead.

apps/webapp/app/v3/services/billingLimit/billingLimitReconciliation.server.ts

@@ -1,4 +1,5 @@
 import { EnvironmentPauseSource } from "@trigger.dev/database";
+import type { PrismaClient } from "@trigger.dev/database";
 import pMap from "p-map";
 import { prisma } from "~/db.server";
 import type { BillingLimitResult } from "~/services/billingLimit.schemas";
@@ -47,15 +48,14 @@ export function resolveReconcileTargetFromBillingLimit(
   return resolveConvergeTargetFromBillingLimit(billingLimit);
 }
 
-export async function getOrgIdsWithBillingPauseSource(): Promise<string[]> {
-  const rows = await prisma.runtimeEnvironment.findMany({
+export async function getOrgIdsWithBillingPauseSource(
+  db: PrismaClient = prisma
+): Promise<string[]> {
+  const rows = await db.runtimeEnvironment.groupBy({
+    by: ["organizationId"],
     where: {
       pauseSource: EnvironmentPauseSource.BILLING_LIMIT,
     },
-    select: {
-      organizationId: true,
-    },
-    distinct: ["organizationId"],
   });
 
   return rows.map((row) => row.organizationId);

apps/webapp/test/billingLimitReconciliation.test.ts

@@ -1,7 +1,10 @@
+import { postgresTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
 import { describe, expect, it } from "vitest";
 import type { BillingLimitResult } from "~/services/billingLimit.schemas";
 import {
   collectOrgIdsNeedingBillingLimitLookup,
+  getOrgIdsWithBillingPauseSource,
   resolveConvergeTargetFromBillingLimit,
   resolveReconcileTargetFromBillingLimit,
   resolveReconcileTargetsForOrgLookups,
@@ -97,3 +100,70 @@ describe("billingLimitReconciliation", () => {
     expect(new Set(lookedUpOrgIds)).toEqual(new Set(["org_ok", "org_fail", "org_grace"]));
   });
 });
+
+let envSeedCounter = 0;
+
+async function seedEnvironment(
+  prisma: PrismaClient,
+  opts: { organizationId: string; projectId: string; pauseSource: "BILLING_LIMIT" | null }
+) {
+  const n = envSeedCounter++;
+  return prisma.runtimeEnvironment.create({
+    data: {
+      slug: `env-${n}`,
+      type: "PRODUCTION",
+      projectId: opts.projectId,
+      organizationId: opts.organizationId,
+      apiKey: `api-${n}`,
+      pkApiKey: `pk-${n}`,
+      shortcode: `sc-${n}`,
+      pauseSource: opts.pauseSource,
+    },
+  });
+}
+
+describe("getOrgIdsWithBillingPauseSource", () => {
+  postgresTest(
+    "returns each org once and ignores envs without the billing-limit pause source",
+    async ({ prisma }) => {
+      const seed: Record<string, Array<"BILLING_LIMIT" | null>> = {
+        org_a: ["BILLING_LIMIT", "BILLING_LIMIT"],
+        org_b: ["BILLING_LIMIT"],
+        org_c: [null],
+      };
+
+      const orgIdBySlug = new Map<string, string>();
+
+      for (const [slug, pauseSources] of Object.entries(seed)) {
+        const organization = await prisma.organization.create({
+          data: { title: slug, slug: `${slug}-${envSeedCounter}` },
+        });
+        const project = await prisma.project.create({
+          data: {
+            name: slug,
+            slug: `proj-${slug}-${envSeedCounter}`,
+            organizationId: organization.id,
+            externalRef: `ext-${slug}-${envSeedCounter}`,
+          },
+        });
+        orgIdBySlug.set(slug, organization.id);
+
+        for (const pauseSource of pauseSources) {
+          await seedEnvironment(prisma, {
+            organizationId: organization.id,
+            projectId: project.id,
+            pauseSource,
+          });
+        }
+      }
+
+      const orgIds = await getOrgIdsWithBillingPauseSource(prisma);
+
+      expect(orgIds.length).toBe(new Set(orgIds).size);
+      expect([...orgIds].sort()).toEqual(
+        [orgIdBySlug.get("org_a")!, orgIdBySlug.get("org_b")!].sort()
+      );
+    },
+    30_000
+  );
+});

internal-packages/database/prisma/migrations/20260812120000_add_runtime_environment_pause_source_index/migration.sql

@@ -0,0 +1,3 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "RuntimeEnvironment_pauseSource_organizationId_idx"
+ON "RuntimeEnvironment" ("pauseSource", "organizationId")
+WHERE "pauseSource" IS NOT NULL;