refactor(webapp): per-org basins for paid orgs only

Free orgs share the global stream basin (the existing legacy fallback
path); paid orgs get a dedicated per-org basin with retention tied to
their tier. Cleaner story, much smaller S2 footprint, and basin
existence becomes a real tier benefit rather than a default for
everyone.

A single `v3.reconcileStreamBasinForOrg` worker job handles every
plan transition idempotently:

  free → paid:  provision a new basin, stamp `Organization.streamBasinName`.
  paid → paid:  reconfigure retention (tier-change). S2 retention only
                takes effect on new streams, but that's fine — old
                streams age out on their original retention.
  paid → free:  null `Organization.streamBasinName`. Future runs/sessions
                for this org route through the shared global basin via
                the existing read-precedence fallback. The per-org basin
                lingers; existing streams there respect their original
                retention until they age out.
  free → free:  no-op.

Replaces the previous `provisionStreamBasinForOrg` /
`reconfigureStreamBasinForOrg` job pair so callers don't have to
choose the right job for the transition. `setPlan` enqueues
`reconcile` from all three plan-changed branches; the admin backfill
route enqueues `reconcile` for every non-deleted org (idempotent — the
worker decides per-org what to do).

Org create no longer provisions synchronously — new orgs start free
and use the shared basin until their first paid upgrade.

Verified locally: backfill correctly deprovisioned 4 free orgs (column
nulled, basins left intact) and kept the 1 hobby-tier org's basin.
A fresh chat for a free org streams into the shared basin under the
legacy prefix `org/{orgId}/env/.../sessions/{chatId}/{io}` with no
new streams in the old per-org basin.

Author: ericallam

apps/webapp/app/models/organization.server.ts

@@ -14,8 +14,6 @@ import { env } from "~/env.server";
 import { featuresForUrl } from "~/features.server";
 import { createApiKeyForEnv, createPkApiKeyForEnv, envSlug } from "./api-key.server";
 import { getDefaultEnvironmentConcurrencyLimit } from "~/services/platform.v3.server";
-import { logger } from "~/services/logger.server";
-import { provisionBasinForOrg } from "~/services/realtime/streamBasinProvisioner.server";
 export type { Organization };
 
 const nanoid = customAlphabet("1234567890abcdef", 4);
@@ -84,25 +82,13 @@ export async function createOrganization(
     },
   });
 
-  // Provision the org's S2 basin synchronously so the very first run
-  // gets `streamBasinName` stamped via the existing org read. New orgs
-  // get the default retention; the plan-change path updates retention
-  // later if the operator runs a billing-aware install. Soft-fail on
-  // S2 errors so a transient outage doesn't block signup — the
-  // backfill reconciler picks up any org left with `streamBasinName: null`.
-  // No-op when `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=false` (OSS mode).
-  try {
-    await provisionBasinForOrg({
-      id: organization.id,
-      streamBasinName: organization.streamBasinName,
-      // No `retention` — provisioner uses `defaultRetention()`.
-    });
-  } catch (error) {
-    logger.warn("[createOrganization] streamBasin provisioning failed; backfill will retry", {
-      orgId: organization.id,
-      error: error instanceof Error ? error.message : String(error),
-    });
-  }
+  // No basin work at signup. New orgs start on free and share the
+  // global stream basin via the read-precedence fallback
+  // (`run.streamBasinName ?? session.streamBasinName ?? legacy basin`).
+  // When the org upgrades to a paid plan, `setPlan` enqueues
+  // `v3.reconcileStreamBasinForOrg` which provisions a dedicated
+  // basin and stamps `streamBasinName`. Free → paid → free
+  // round-trips are reconciled by the same job.
 
   return { ...organization };
 }

apps/webapp/app/routes/admin.api.v1.stream-basins.backfill.ts

@@ -7,26 +7,22 @@ import { commonWorker } from "~/v3/commonWorker.server";
 import { logger } from "~/services/logger.server";
 
 /**
- * One-shot backfill that enqueues `v3.provisionStreamBasinForOrg` for
- * every org with `streamBasinName: null`. Idempotent — re-running picks
- * up only the orgs that haven't been provisioned yet, and the worker
- * job itself is also idempotent (the provisioner short-circuits if the
- * org column is already set).
+ * One-shot backfill that enqueues `v3.reconcileStreamBasinForOrg` for
+ * every non-deleted org. The reconciler decides per-org what to do:
+ * provision a basin for paid orgs that don't have one, reconfigure
+ * retention for paid orgs whose tier changed, deprovision (null the
+ * column) for free orgs that were mistakenly provisioned. Idempotent
+ * — re-running converges to the desired state.
  *
  *  - Admin auth via `requireAdminApiRequest` (PAT in `Authorization`).
  *  - Refuses to run when `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=false`
  *    so OSS / s2-lite installs can't accidentally trigger basin
- *    creation against a misconfigured backend.
+ *    operations against a misconfigured backend.
  *  - `dryRun=true` (default false) returns the count without enqueueing.
  *  - `limit` (default 1000, max 10000) caps a single invocation. Run
- *    again to process more — the column filter naturally walks the
- *    queue forward each call.
- *  - Each job is keyed `provisionStreamBasin:<orgId>` so concurrent
- *    backfill calls converge to one job per org instead of duplicating.
- *
- * Run from a shell:
- *   curl -X POST -H "Authorization: Bearer $PAT" \
- *     "https://api.trigger.dev/admin/api/v1/stream-basins/backfill?limit=200&dryRun=true"
+ *    again with the next batch.
+ *  - Each job is keyed `reconcileStreamBasin:<orgId>` so concurrent
+ *    calls converge to one job per org.
  */
 
 const BodySchema = z
@@ -59,8 +55,6 @@ export async function action({ request }: ActionFunctionArgs) {
     );
   }
 
-  // `application/json` POST body — empty body falls back to defaults so
-  // a parameterless POST does the right thing for the default backfill.
   let parsed: z.infer<typeof BodySchema>;
   try {
     const text = await request.text();
@@ -76,21 +70,19 @@ export async function action({ request }: ActionFunctionArgs) {
 
   const { dryRun, limit } = parsed;
 
-  // Page candidate orgs. Ordered by createdAt so re-runs walk the queue
-  // forward predictably; deletedAt filter avoids resurrecting orgs.
+  // Walk every non-deleted org. The reconcile worker is fast for the
+  // no-op case (free with null column) so enqueueing for all is fine
+  // — saves us from doing per-org billing lookups here just to filter
+  // candidates.
   const candidates = await prisma.organization.findMany({
-    where: {
-      streamBasinName: null,
-      deletedAt: null,
-    },
+    where: { deletedAt: null },
     orderBy: { createdAt: "asc" },
     take: limit,
     select: { id: true },
   });
 
-  // Total count of remaining nulls (for progress reporting).
-  const remainingTotal = await prisma.organization.count({
-    where: { streamBasinName: null, deletedAt: null },
+  const totalOrgs = await prisma.organization.count({
+    where: { deletedAt: null },
   });
 
   if (dryRun) {
@@ -99,22 +91,19 @@ export async function action({ request }: ActionFunctionArgs) {
       dryRun: true,
       enqueued: 0,
       pending: candidates.length,
-      remaining: Math.max(0, remainingTotal - candidates.length),
+      remaining: Math.max(0, totalOrgs - candidates.length),
       orgIds: candidates.map((o) => o.id),
     };
     return json(response);
   }
 
-  // Enqueue one job per org. Per-org dedupe key collapses concurrent
-  // backfill calls into a single pending job, and a job that's already
-  // run (basin set) is a no-op on the worker side.
   let enqueued = 0;
   for (const org of candidates) {
     try {
       await commonWorker.enqueue({
-        job: "v3.provisionStreamBasinForOrg",
+        job: "v3.reconcileStreamBasinForOrg",
         payload: { orgId: org.id },
-        id: `provisionStreamBasin:${org.id}`,
+        id: `reconcileStreamBasin:${org.id}`,
       });
       enqueued += 1;
     } catch (error) {
@@ -130,11 +119,11 @@ export async function action({ request }: ActionFunctionArgs) {
     dryRun: false,
     enqueued,
     pending: candidates.length,
-    remaining: Math.max(0, remainingTotal - enqueued),
+    remaining: Math.max(0, totalOrgs - enqueued),
     orgIds: candidates.map((o) => o.id),
   };
 
-  logger.info("[stream-basins-backfill] enqueued provisioning jobs", {
+  logger.info("[stream-basins-backfill] enqueued reconcile jobs", {
     enqueued,
     candidates: candidates.length,
     remaining: response.remaining,
@@ -149,17 +138,15 @@ export async function loader({ request }: ActionFunctionArgs) {
   await requireAdminApiRequest(request);
 
   const totalOrgs = await prisma.organization.count({ where: { deletedAt: null } });
-  const provisioned = await prisma.organization.count({
+  const withBasin = await prisma.organization.count({
     where: { deletedAt: null, NOT: { streamBasinName: null } },
   });
-  const remaining = totalOrgs - provisioned;
 
   return json({
     ok: true,
     perOrgBasinsEnabled: isPerOrgBasinsEnabled(),
     totalOrgs,
-    provisioned,
-    remaining,
-    completion: totalOrgs === 0 ? 1 : provisioned / totalOrgs,
+    withBasin,
+    withoutBasin: totalOrgs - withBasin,
   });
 }

apps/webapp/app/routes/admin.api.v1.stream-basins.reconfigure.ts

@@ -67,9 +67,9 @@ export async function action({ request }: ActionFunctionArgs) {
   }
 
   await commonWorker.enqueue({
-    job: "v3.reconfigureStreamBasinForOrg",
+    job: "v3.reconcileStreamBasinForOrg",
     payload: { orgId: parsed.data.orgId },
-    id: `reconfigureStreamBasin:${parsed.data.orgId}`,
+    id: `reconcileStreamBasin:${parsed.data.orgId}`,
   });
 
   return json({ ok: true, mode: "queued", enqueued: parsed.data.orgId });

apps/webapp/app/services/platform.v3.server.ts

@@ -403,7 +403,7 @@ export async function setPlan(
         // Invalidate billing cache since plan changed
         opts?.invalidateBillingCache?.(organization.id);
         platformCache.entitlement.remove(organization.id).catch(() => {});
-        await enqueueStreamBasinReconfigure(organization.id);
+        await enqueueStreamBasinReconcile(organization.id);
         return redirect(newProjectPath(organization, "You're on the Free plan."));
       } else {
         return redirectWithErrorMessage(
@@ -421,40 +421,49 @@ export async function setPlan(
       // Invalidate billing cache since subscription changed
       opts?.invalidateBillingCache?.(organization.id);
       platformCache.entitlement.remove(organization.id).catch(() => {});
-      await enqueueStreamBasinReconfigure(organization.id);
+      await enqueueStreamBasinReconcile(organization.id);
       return redirectWithSuccessMessage(callerPath, request, "Subscription updated successfully.");
     }
     case "canceled_subscription": {
       // Invalidate billing cache since subscription was canceled
       opts?.invalidateBillingCache?.(organization.id);
       platformCache.entitlement.remove(organization.id).catch(() => {});
-      await enqueueStreamBasinReconfigure(organization.id);
+      await enqueueStreamBasinReconcile(organization.id);
       return redirectWithSuccessMessage(callerPath, request, "Subscription canceled.");
     }
   }
 }
 
 /**
- * Best-effort enqueue: when an org's plan changes we want the per-org
- * S2 basin's retention to follow (free=7d, hobby=30d, pro=365d). The
- * worker job is idempotent and a no-op when per-org basins are disabled
- * or the org has no basin yet (OSS / pre-backfill). Failures are
- * logged but never block the plan change itself — billing has already
- * accepted by the time we reach this code.
+ * Best-effort enqueue: when an org's plan changes we reconcile its
+ * stream-basin state. The reconciler handles every transition:
+ *
+ *   free → paid:  provision a dedicated basin with the plan's retention.
+ *   paid → paid:  reconfigure the existing basin's retention.
+ *   paid → free:  null `Organization.streamBasinName`. Future runs/sessions
+ *                 flow to the shared global basin; the per-org basin
+ *                 lingers until existing streams age out on their original
+ *                 retention.
+ *   free → free:  no-op.
+ *
+ * Idempotent and a no-op when per-org basins are disabled or billing
+ * isn't configured. Failures are logged but never block the plan
+ * change itself — billing has already accepted by the time we reach
+ * this code.
  */
-async function enqueueStreamBasinReconfigure(orgId: string) {
+async function enqueueStreamBasinReconcile(orgId: string) {
   try {
     const { commonWorker } = await import("~/v3/commonWorker.server");
     await commonWorker.enqueue({
-      job: "v3.reconfigureStreamBasinForOrg",
+      job: "v3.reconcileStreamBasinForOrg",
       payload: { orgId },
       // Per-org dedupe key — concurrent plan changes collapse to one
-      // pending reconfigure job. The job re-reads the current plan
-      // when it executes, so the latest tier wins.
-      id: `reconfigureStreamBasin:${orgId}`,
+      // pending reconcile job. The job re-reads the current plan when
+      // it executes, so the latest tier wins.
+      id: `reconcileStreamBasin:${orgId}`,
     });
   } catch (error) {
-    logger.warn("[setPlan] failed to enqueue stream basin reconfigure", {
+    logger.warn("[setPlan] failed to enqueue stream basin reconcile", {
       orgId,
       error: error instanceof Error ? error.message : String(error),
     });