chore(webapp): trim per-org-basin comments

Pass over the basin-related code to drop running commentary and
cloud-product-specific phrasing. No behaviour change.

- streamBasinProvisioner.server.ts: shorter module + helper docblocks;
  drop stale "synchronous org-create call site" rationale that no
  longer applies after the paid-only refactor.
- streamBasinRetentionByPlan.server.ts: tighter module doc; collapse
  the reconcile-transitions narrative into a short table; drop the
  "cloud-flavored" framing.
- v1StreamsGlobal.server.ts: short doc on resolveStreamBasin; drop
  references to specific operational state.
- env.server.ts: terse one-liner per env var; drop sample basin name
  example.
- platform.v3.server.ts, commonWorker.server.ts, runEngine/types.ts +
  index.ts, triggerTask.server.ts, api.v1.sessions.ts, the two admin
  routes and the two row-optional session-channel handlers: drop
  inline rationale paragraphs that re-explained the reconciler /
  read-precedence chain at every call site.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -5,12 +5,8 @@ import { isValidDatabaseUrl } from "./utils/db";
 import { isValidRegex } from "./utils/regex";
 import { isValidDuration } from "./services/realtime/duration.server";
 
-/**
- * `z.string()` constrained to a duration string parseable by
- * `parseDuration` (e.g. `7d`, `30d`, `365d`, `1h`). Validated at boot
- * so a typo'd retention env var fails fast at startup rather than
- * lurking until the first basin operation.
- */
+// `z.string()` constrained to a `parseDuration`-parseable string (e.g.
+// `7d`, `1h`). Validated at boot so a typo'd duration fails fast.
 function durationString() {
   return z
     .string()
@@ -1519,33 +1515,20 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
     REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
-    /// Per-org basin migration. When "true", the webapp provisions a
-    /// dedicated S2 basin per org with plan-tied retention and stamps
-    /// `streamBasinName` on new TaskRun / Session rows. OSS / s2-lite
-    /// installs leave this off and keep using the single basin defined
-    /// by `REALTIME_STREAMS_S2_BASIN`.
+    // When "true", provision a dedicated S2 basin per org and stamp
+    // `streamBasinName` on new rows. Off keeps everything on the single
+    // basin defined by `REALTIME_STREAMS_S2_BASIN`.
     REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: z.enum(["true", "false"]).default("false"),
-    /// Naming pattern for per-org basins: `{prefix}-{env}-org-{slug}`
-    /// e.g. `triggerdotdev-prod-org-acme-corp`. Cluster + tier shorthand
-    /// — kept short to stay under S2's basin-name length limit.
+    // Per-org basin name = `{prefix}-{env}-org-{orgId}`.
     REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
     REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
-    /// Default retention for new basins (S2 duration syntax: 7d / 30d / 1y).
-    /// Used at org-create and as the fallback when no plan-specific
-    /// retention is resolved. Operators that don't run a billing API
-    /// only need this one.
     REALTIME_STREAMS_BASIN_DEFAULT_RETENTION: durationString().default("30d"),
-    /// Plan-specific retention overrides — only consulted by the
-    /// optional `streamBasinRetentionByPlan` shim. Operators that
-    /// don't map plans to retention (OSS, self-hosted) can ignore
-    /// these and rely on the default above.
+    // 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"),
-    /// Storage class applied to per-org basins at create time.
     REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
-    /// `delete_on_empty_min_age` applied to per-org basins. Streams
-    /// that go empty for this long are reaped automatically.
     REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("1h"),
     REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
     WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),

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

@@ -7,23 +7,9 @@ import { commonWorker } from "~/v3/commonWorker.server";
 import { logger } from "~/services/logger.server";
 
 /**
- * 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
- *    operations against a misconfigured backend.
- *  - `dryRun=true` (default false) returns the count without enqueueing.
- *  - `limit` (default 1000, max 10000) caps a single invocation. To
- *    page through more orgs than `limit`, pass `afterOrgId` from the
- *    previous response's `nextAfterOrgId`.
- *  - Each job is keyed `reconcileStreamBasin:<orgId>` so concurrent
- *    calls converge to one job per org.
+ * Backfill: enqueue `v3.reconcileStreamBasinForOrg` for every
+ * non-deleted org. Idempotent. Page through `>limit` orgs by passing
+ * `afterOrgId` from the previous response's `nextAfterOrgId`.
  */
 
 const BodySchema = z
@@ -73,11 +59,8 @@ export async function action({ request }: ActionFunctionArgs) {
 
   const { dryRun, limit, afterOrgId } = parsed;
 
-  // 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. Cursor on `id` (cuid is sortable) gives stable paging
-  // across calls; `createdAt` ties get broken by the cursor.
+  // Reconcile is fast for the no-op case, so we enqueue for all orgs
+  // rather than filter on plan here.
   const candidates = await prisma.organization.findMany({
     where: { deletedAt: null },
     orderBy: { id: "asc" },
@@ -89,9 +72,6 @@ export async function action({ request }: ActionFunctionArgs) {
   const lastReturnedId = candidates[candidates.length - 1]?.id;
   const nextAfterOrgId = candidates.length === limit && lastReturnedId ? lastReturnedId : null;
 
-  // Orgs still beyond the cursor we just returned. On the final page,
-  // `lastReturnedId` is undefined (empty result) or the response is short
-  // of `limit`, so this is 0 — exactly what the caller needs to stop.
   const remaining = lastReturnedId
     ? await prisma.organization.count({
         where: { deletedAt: null, id: { gt: lastReturnedId } },
@@ -128,10 +108,6 @@ export async function action({ request }: ActionFunctionArgs) {
     }
   }
 
-  // `remaining` counts orgs strictly past the cursor returned to the
-  // caller. Enqueue failures don't change this — re-running with the
-  // same `afterOrgId` would page through the same window and the
-  // per-org idempotency key keeps it safe.
   const response: BackfillResponse = {
     ok: true,
     dryRun: false,
@@ -151,8 +127,7 @@ export async function action({ request }: ActionFunctionArgs) {
   return json(response);
 }
 
-// GET returns the current state without doing anything — useful for
-// monitoring "is the backfill done yet?" from a dashboard / curl.
+// GET: read-only progress — orgs with vs without a basin stamped.
 export async function loader({ request }: ActionFunctionArgs) {
   await requireAdminApiRequest(request);
 

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

@@ -9,21 +9,14 @@ import {
 import { commonWorker } from "~/v3/commonWorker.server";
 
 /**
- * Admin trigger for stream-basin reconfiguration. The plan-change path
- * in `setPlan` enqueues the same reconcile job automatically when
- * billing is wired; this route exists for ops + e2e testing.
+ * Admin route for forcing a basin reconfigure for an org. Two modes:
  *
- * - Default (`{ orgId }`): enqueues `v3.reconcileStreamBasinForOrg`,
- *   the full reconciler. It resolves retention from the org's current
- *   plan and either provisions, reconfigures, or deprovisions the basin
- *   to match — including nulling `streamBasinName` if the org is now on
- *   a free plan. No-op when billing isn't configured (OSS) or when
- *   `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=false`.
- * - With `retention`: skips the worker queue and the reconciler entirely.
- *   Calls `reconfigureBasinForOrg` inline with the given duration string
- *   (e.g. `"7d"`, `"30d"`, `"365d"`, `"1y"`). Useful for validating the
- *   PATCH wire shape end-to-end and as a manual override (e.g.
- *   enterprise contracts) — does NOT touch the column or check the plan.
+ * - `{ orgId }`: enqueues `v3.reconcileStreamBasinForOrg` (the full
+ *   reconciler). May provision, reconfigure, or deprovision based on
+ *   the org's current plan.
+ * - `{ orgId, retention }`: bypasses the reconciler and PATCHes the
+ *   basin retention inline against the given duration. Doesn't touch
+ *   the column or check the plan.
  */
 const BodySchema = z
   .object({
@@ -58,9 +51,6 @@ export async function action({ request }: ActionFunctionArgs) {
   }
 
   if (parsed.data.retention) {
-    // Direct, synchronous reconfigure with the explicit retention.
-    // Skips the worker queue + billing lookup so the PATCH is
-    // verifiable in the response. Errors surface as 500.
     await reconfigureBasinForOrg(parsed.data.orgId, parsed.data.retention);
     return json({
       ok: true,

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

@@ -123,19 +123,10 @@ const { action, loader } = createActionApiRoute(
       // and remove the pending registration.
       if (!result.isCached) {
         try {
-          // Session streams are always v2 (S2) — the writer in
-          // `appendPartToSessionStream` and the SSE subscribe both
-          // hardcode "v2", so the race-check reader has to match.
-          // Don't fall through to the run's own `realtimeStreamsVersion`,
-          // which only describes the run's run-scoped streams.
-          //
-          // Resolve basin from `session` only (not `run`). The append-side
-          // writer in `realtime.v1.sessions.$session.$io.append.ts` passes
-          // only `{ session }`, and `resolveStreamBasin` prefers `run` over
-          // `session` when both are present. During the per-org-basin
-          // migration window, `run.streamBasinName` and
-          // `session.streamBasinName` can differ — the writes land in the
-          // session's basin, so the race-check has to read from the same.
+          // Session streams are hardcoded v2 by the append-side writer
+          // and SSE subscribe, so the race-check reader matches. Basin
+          // comes from `session` only — the writer side passes the same
+          // and we have to read from the same basin to find the record.
           const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
             session: maybeSession,
           });

apps/webapp/app/routes/api.v1.sessions.ts

@@ -167,9 +167,6 @@ const { action } = createActionApiRoute(
             runtimeEnvironmentId: authentication.environment.id,
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
-            // Stamp the org's S2 basin so realtime reads on this
-            // session's `.in/.out` channels resolve without joining
-            // Organization. Null until per-org basins are provisioned.
             streamBasinName: authentication.environment.organization.streamBasinName,
           },
           update: { triggerConfig: triggerConfigJson },
@@ -190,9 +187,6 @@ const { action } = createActionApiRoute(
             runtimeEnvironmentId: authentication.environment.id,
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
-            // Stamp the org's S2 basin so realtime reads on this
-            // session's `.in/.out` channels resolve without joining
-            // Organization. Null until per-org basins are provisioned.
             streamBasinName: authentication.environment.organization.streamBasinName,
           },
         });

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.ts

@@ -59,15 +59,9 @@ const { action } = createActionApiRoute(
       });
     }
 
-    // When the row is missing (externalId form, row not yet upserted),
-    // default to the org's current basin instead of falling through to
-    // the legacy global. A fresh session row would be stamped with the
-    // org's basin at creation time, so subsequent appends/subscribes
-    // would resolve to the same place — without this, PUT-returned
-    // headers point at legacy and the actual writes go to per-org once
-    // the row exists. Pre-migration rows (row exists, column null) keep
-    // their existing legacy behaviour because we only fall back to org
-    // when there's no row at all.
+    // No-row form: resolve via the org so the stream initialised here
+    // matches what later appends/subscribes will land on once the row
+    // is created.
     const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
       session: maybeSession,
       organization: maybeSession ? null : authentication.environment.organization,
@@ -134,10 +128,7 @@ const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, authentication, resource }) => {
-    // Same row-optional reasoning as the PUT handler above: if no row
-    // exists yet, resolve via the org's current basin so the SSE
-    // subscribe lands in the same place that subsequent appends will
-    // (once the row gets created and stamped).
+    // Same no-row fallback as PUT above.
     const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
       session: resource.row,
       organization: resource.row ? null : authentication.environment.organization,

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -395,10 +395,6 @@ export class RunEngineTriggerTaskService {
                 bulkActionId: body.options?.bulkActionId,
                 planType,
                 realtimeStreamsVersion: options.realtimeStreamsVersion,
-                // Stamp the org's S2 basin onto the new TaskRun so
-                // realtime read paths can resolve the basin without
-                // joining `Organization`. Null in OSS / pre-backfill;
-                // reads then fall back to the global basin env var.
                 streamBasinName: environment.organization.streamBasinName,
                 debounce: body.options?.debounce,
                 annotations,

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

@@ -434,32 +434,15 @@ export async function setPlan(
   }
 }
 
-/**
- * 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.
- */
+// 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 },
-      // Per-org dedupe key — concurrent plan changes collapse to one
-      // pending reconcile job. The job re-reads the current plan when
-      // it executes, so the latest tier wins.
       id: `reconcileStreamBasin:${orgId}`,
     });
   } catch (error) {