refactor(webapp): cloud-driven basin sync, drop reconcile worker

The cloud billing app now drives basin lifecycle operations via a
single webapp admin endpoint instead of the webapp enqueuing a
reconcile worker that called back into billing to resolve plan +
retention. Plan vocabulary now lives in cloud
(`Limits.streamBasinRetention`); S2 access stays in the webapp.

New: POST /admin/api/v1/orgs/:orgId/stream-basin
  - { action: "ensure", retention } — provision or PATCH retention
  - { action: "deprovision" }       — null the column

Helpers: ensureBasinForOrg / deprovisionBasinForOrg in the
provisioner.

Removed:
- streamBasinRetentionByPlan.server.ts (plan vocabulary)
- v3.reconcileStreamBasinForOrg worker job
- enqueueStreamBasinReconcile from the three setPlan branches
- admin.api.v1.stream-basins.{backfill,reconfigure}.ts (replaced by
  the per-org sync endpoint)
- REALTIME_STREAMS_BASIN_RETENTION_FREE/HOBBY/PRO env vars

The .server-changes entry stays accurate — feature behaviour is
unchanged from a customer's perspective.

Author: ericallam

.server-changes/per-org-stream-basins.md

@@ -3,4 +3,4 @@ area: webapp
 type: feature
 ---
 
-Per-org S2 stream basins with plan-tied retention (free 7d / hobby 30d / pro 365d), gated by `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED`. Stops basin retention from deleting streams out from under live chat sessions and unlocks per-org cost attribution via S2 basin metrics.
+Per-org S2 stream basins with retention tied to the org's billing plan, gated by `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED`. Stops basin retention from deleting streams out from under live chat sessions and unlocks per-org cost attribution.

apps/webapp/app/env.server.ts

@@ -1523,11 +1523,6 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
     REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
     REALTIME_STREAMS_BASIN_DEFAULT_RETENTION: durationString().default("30d"),
-    // Plan-specific retention overrides consulted by the
-    // streamBasinRetentionByPlan shim only.
-    REALTIME_STREAMS_BASIN_RETENTION_FREE: durationString().default("7d"),
-    REALTIME_STREAMS_BASIN_RETENTION_HOBBY: durationString().default("30d"),
-    REALTIME_STREAMS_BASIN_RETENTION_PRO: durationString().default("365d"),
     REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
     REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("1h"),
     REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),

apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.stream-basin.ts

@@ -0,0 +1,47 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { isValidDuration } from "~/services/realtime/duration.server";
+import {
+  deprovisionBasinForOrg,
+  ensureBasinForOrg,
+} from "~/services/realtime/streamBasinProvisioner.server";
+
+const ParamsSchema = z.object({ organizationId: z.string() });
+
+const BodySchema = z.discriminatedUnion("action", [
+  z.object({
+    action: z.literal("ensure"),
+    retention: z
+      .string()
+      .refine(isValidDuration, "retention must be a duration like 7d, 30d, 365d, 1h, 1y"),
+  }),
+  z.object({ action: z.literal("deprovision") }),
+]);
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  const { organizationId } = ParamsSchema.parse(params);
+
+  let parsed: z.infer<typeof BodySchema>;
+  try {
+    const text = await request.text();
+    const raw = text.length > 0 ? JSON.parse(text) : {};
+    const result = BodySchema.safeParse(raw);
+    if (!result.success) {
+      return json({ ok: false, error: result.error.flatten() }, { status: 400 });
+    }
+    parsed = result.data;
+  } catch {
+    return json({ ok: false, error: "Invalid JSON body" }, { status: 400 });
+  }
+
+  if (parsed.action === "ensure") {
+    const result = await ensureBasinForOrg(organizationId, parsed.retention);
+    return json({ ok: true, ...result });
+  }
+
+  const result = await deprovisionBasinForOrg(organizationId);
+  return json({ ok: true, ...result });
+}

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

@@ -403,7 +403,6 @@ export async function setPlan(
         // Invalidate billing cache since plan changed
         opts?.invalidateBillingCache?.(organization.id);
         platformCache.entitlement.remove(organization.id).catch(() => {});
-        await enqueueStreamBasinReconcile(organization.id);
         return redirect(newProjectPath(organization, "You're on the Free plan."));
       } else {
         return redirectWithErrorMessage(
@@ -421,38 +420,17 @@ export async function setPlan(
       // Invalidate billing cache since subscription changed
       opts?.invalidateBillingCache?.(organization.id);
       platformCache.entitlement.remove(organization.id).catch(() => {});
-      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 enqueueStreamBasinReconcile(organization.id);
       return redirectWithSuccessMessage(callerPath, request, "Subscription canceled.");
     }
   }
 }
 
-// Best-effort: failures are logged but never block the plan change.
-// The reconciler is idempotent and re-reads the plan when it runs, so
-// concurrent plan changes collapse to one pending job per org.
-async function enqueueStreamBasinReconcile(orgId: string) {
-  try {
-    const { commonWorker } = await import("~/v3/commonWorker.server");
-    await commonWorker.enqueue({
-      job: "v3.reconcileStreamBasinForOrg",
-      payload: { orgId },
-      id: `reconcileStreamBasin:${orgId}`,
-    });
-  } catch (error) {
-    logger.warn("[setPlan] failed to enqueue stream basin reconcile", {
-      orgId,
-      error: error instanceof Error ? error.message : String(error),
-    });
-  }
-}
-
 export async function setConcurrencyAddOn(organizationId: string, amount: number) {
   if (!client) return undefined;