fix(webapp): address coderabbit review on per-org basin migration

- reconfigure admin route: guard `JSON.parse` with try/catch + empty-body
  check so a malformed POST returns 400 instead of an unhandled 500
  (mirror of the backfill route).
- session-streams.wait race-check: select `streamBasinName` on the run
  and pass `{ run, session }` to `getRealtimeStreamInstance` so the
  resolver picks up the run's stamped basin when the session row is
  unavailable.
- streamBasinProvisioner: 10s `AbortSignal.timeout()` on both `s2CreateBasin`
  and `s2ReconfigureBasin` so the synchronous org-create path can't hang
  signup forever on a slow/unresponsive S2.
- commonWorker basin handlers: throw when `getCurrentPlan` returns
  undefined (billing API failure) so redis-worker retries instead of
  silently defaulting to "free" tier — a reconfigure landing during a
  transient billing outage would otherwise clip a pro org's retention
  from 365d to 7d.

Author: ericallam

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

@@ -38,8 +38,14 @@ export async function action({ request }: ActionFunctionArgs) {
     );
   }
 
-  const text = await request.text();
-  const parsed = BodySchema.safeParse(JSON.parse(text));
+  let parsed: ReturnType<typeof BodySchema.safeParse>;
+  try {
+    const text = await request.text();
+    const raw = text.length > 0 ? JSON.parse(text) : {};
+    parsed = BodySchema.safeParse(raw);
+  } catch {
+    return json({ ok: false, error: "Invalid JSON body" }, { status: 400 });
+  }
   if (!parsed.success) {
     return json({ ok: false, error: parsed.error.flatten() }, { status: 400 });
   }

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -47,6 +47,7 @@ const { action, loader } = createActionApiRoute(
           id: true,
           friendlyId: true,
           realtimeStreamsVersion: true,
+          streamBasinName: true,
         },
       });
 
@@ -129,6 +130,7 @@ const { action, loader } = createActionApiRoute(
           // Don't fall through to the run's own `realtimeStreamsVersion`,
           // which only describes the run's run-scoped streams.
           const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+            run,
             session: maybeSession,
           });
 

apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts

@@ -219,6 +219,11 @@ async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise<vo
   };
 
   const res = await fetch(url, {
+    // 10s upper bound so the synchronous org-create call site can't
+    // hang signup forever if S2 is slow / unreachable. Soft-fail at the
+    // caller swallows the resulting `TimeoutError`; the backfill worker
+    // retries the unprovisioned org later.
+    signal: AbortSignal.timeout(10_000),
     method: "POST",
     headers: {
       Authorization: `Bearer ${opts.accessToken}`,
@@ -250,6 +255,10 @@ async function s2ReconfigureBasin(name: string, opts: ReconfigureBasinOptions):
   };
 
   const res = await fetch(url, {
+    // Same 10s ceiling as create. The reconfigure path runs from the
+    // worker, so a timeout here just fails the job and lets redis-worker
+    // retry naturally.
+    signal: AbortSignal.timeout(10_000),
     method: "PATCH",
     headers: {
       Authorization: `Bearer ${opts.accessToken}`,

apps/webapp/app/v3/commonWorker.server.ts

@@ -324,6 +324,16 @@ function initializeWorker() {
         if (!org) return;
 
         const plan = await getCurrentPlan(payload.orgId);
+        // `plan === undefined` means the billing API call itself failed
+        // (or the client isn't configured). Throw so redis-worker retries
+        // — silently defaulting to free would risk a paid org getting
+        // provisioned with 7d retention if the backfill happened to land
+        // during a transient billing outage.
+        if (plan === undefined) {
+          throw new Error(
+            `[provisionStreamBasinForOrg] billing plan unavailable for org ${payload.orgId}; will retry`
+          );
+        }
         // `plan.code` carries the canonical plan id ("free", "v3_hobby_1",
         // "v3_pro_1", "enterprise"). `plan.type` is just the
         // billing-shape discriminator ("free" | "paid" | "enterprise")
@@ -334,10 +344,15 @@ function initializeWorker() {
       },
       "v3.reconfigureStreamBasinForOrg": async ({ payload }) => {
         const plan = await getCurrentPlan(payload.orgId);
-        // `plan.code` carries the canonical plan id ("free", "v3_hobby_1",
-        // "v3_pro_1", "enterprise"). `plan.type` is just the
-        // billing-shape discriminator ("free" | "paid" | "enterprise")
-        // and would lump hobby + pro into one bucket.
+        // Same guard as provision. A reconfigure that silently resolved
+        // to "free" would clip a pro org's retention from 365d to 7d
+        // and prematurely expire history — never acceptable. Throw and
+        // let the worker retry once billing recovers.
+        if (plan === undefined) {
+          throw new Error(
+            `[reconfigureStreamBasinForOrg] billing plan unavailable for org ${payload.orgId}; will retry`
+          );
+        }
         const tier = planTierFor(plan?.v3Subscription?.plan?.code);
         await reconfigureBasinForOrg(payload.orgId, tier);
       },