refactor(webapp): drop plan vocabulary from streamBasinProvisioner

The provisioner is now purely retention-string-driven: callers pass a
duration like "30d" and it does the S2 round-trip. No tier types, no
plan-name matching, no billing imports.

The plan-aware mapping moves into a new
`streamBasinRetentionByPlan.server.ts` shim that's the only file in the
webapp that knows about plan codes. Callers that resolve retention
from a plan (the worker's backfill / reconfigure handlers) import the
shim; callers that just want a default (the org-create path) call the
provisioner without `retention`.

Also addresses two review concerns:

- `basinNameForOrg` now throws when the configured prefix + env-name
  leave zero or negative budget for the org slug. Without the guard a
  too-long prefix would produce `slice(0, 0) = ""` for every org and
  silently collide their basins via S2's idempotent-create path.
- The plan-code → retention mapping uses an exact-match switch instead
  of substring matching. Substring matching against future plan codes
  could grant the wrong retention (e.g. `"approved"` matching `"pro"`).
  The known set is small and explicit; new plan codes go in the switch
  at launch.

Net surface change:
  - `streamBasinProvisioner.server.ts`: drops `StreamBasinTier`,
    `planTierFor`, `retentionFor` exports. Adds `defaultRetention()`.
    `provisionBasinForOrg` takes `{ retention?: string }` instead of
    `{ tier?: StreamBasinTier }`. `reconfigureBasinForOrg` takes a
    retention string instead of a tier.
  - `streamBasinRetentionByPlan.server.ts` (new): exports
    `resolveRetentionForOrg(orgId)` and `retentionForPlanCode(code)`.
  - `commonWorker.server.ts`: handlers call the shim, hand a string to
    the provisioner.
  - Admin reconfigure route: replaces the `tier` body field with a
    direct `retention` duration override.
  - Org create: no longer passes `tier: "free"`; provisioner uses the
    default.
  - New env var `REALTIME_STREAMS_BASIN_DEFAULT_RETENTION` (default
    `30d`). Existing per-plan vars are still consulted by the shim
    only.

Verified end-to-end with chat.agent locally — fresh chat lands in the
per-org basin, multi-turn behaves the same, no leakage to the global
fallback.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1517,9 +1517,15 @@ const EnvironmentSchema = z
     /// — kept short to stay under S2's basin-name length limit.
     REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
     REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
-    /// Plan-tier retention strings (S2 duration syntax: 7d / 30d / 1y).
-    /// Free / hobby / pro line up with billing tiers; enterprise uses
-    /// the pro default and is reconfigured per-contract via the API.
+    /// 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: z.string().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.
     REALTIME_STREAMS_BASIN_RETENTION_FREE: z.string().default("7d"),
     REALTIME_STREAMS_BASIN_RETENTION_HOBBY: z.string().default("30d"),
     REALTIME_STREAMS_BASIN_RETENTION_PRO: z.string().default("365d"),

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

@@ -85,16 +85,18 @@ export async function createOrganization(
   });
 
   // Provision the org's S2 basin synchronously so the very first run
-  // gets `streamBasinName` stamped via the existing org read. Soft-fail
-  // on S2 errors so a transient outage doesn't block signup — the
+  // 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,
       slug: organization.slug,
-      tier: "free", // new orgs always start on free retention
       streamBasinName: organization.streamBasinName,
+      // No `retention` — provisioner uses `defaultRetention()`.
     });
   } catch (error) {
     logger.warn("[createOrganization] streamBasin provisioning failed; backfill will retry", {

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

@@ -4,27 +4,26 @@ import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
 import {
   isPerOrgBasinsEnabled,
   reconfigureBasinForOrg,
-  type StreamBasinTier,
 } from "~/services/realtime/streamBasinProvisioner.server";
 import { commonWorker } from "~/v3/commonWorker.server";
 
 /**
  * Admin trigger for `v3.reconfigureStreamBasinForOrg`. The plan-change
- * path in `setPlan` already enqueues this automatically in cloud mode;
+ * path in `setPlan` enqueues this automatically when billing is wired;
  * this route exists for ops + e2e testing.
  *
  * - Default (`{ orgId }`): enqueues the worker job which resolves the
- *   tier via `getCurrentPlan` and PATCHes the basin to match. No-op
- *   locally because `getCurrentPlan` is gated to cloud hosts.
- * - With `tier`: bypasses the billing lookup and runs reconfigure
- *   inline against the given tier. Useful for validating the PATCH
- *   wire shape end-to-end and as a manual override (e.g. enterprise
- *   contract retention).
+ *   retention from the org's plan and PATCHes the basin to match.
+ *   No-op when billing isn't configured (OSS).
+ * - With `retention`: bypasses the billing lookup and runs reconfigure
+ *   inline against 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).
  */
 const BodySchema = z
   .object({
     orgId: z.string(),
-    tier: z.enum(["free", "hobby", "pro"]).optional(),
+    retention: z.string().optional(),
   })
   .strict();
 
@@ -50,13 +49,17 @@ export async function action({ request }: ActionFunctionArgs) {
     return json({ ok: false, error: parsed.error.flatten() }, { status: 400 });
   }
 
-  if (parsed.data.tier) {
-    // Direct, synchronous reconfigure with the explicit tier override.
-    // Skips the worker queue + billing lookup so the PATCH is verifiable
-    // in the response. Errors surface as 500.
-    const tier: StreamBasinTier = parsed.data.tier;
-    await reconfigureBasinForOrg(parsed.data.orgId, tier);
-    return json({ ok: true, mode: "inline", orgId: parsed.data.orgId, tier });
+  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,
+      mode: "inline",
+      orgId: parsed.data.orgId,
+      retention: parsed.data.retention,
+    });
   }
 
   await commonWorker.enqueue({

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

@@ -8,93 +8,80 @@
  *    basin in `REALTIME_STREAMS_S2_BASIN`. `Organization.streamBasinName`
  *    stays null forever; reads / writes resolve to the global basin.
  *
- *  - **Per-org-basin mode** (cloud):
+ *  - **Per-org-basin mode**:
  *    `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=true`. Each org gets a
- *    dedicated basin with retention tied to its billing plan. The
- *    basin is the unit of cost attribution (S2 exposes per-basin
- *    metrics) and isolation (access tokens scope to one basin).
+ *    dedicated basin with its own retention. The basin is the unit of
+ *    cost attribution (S2 exposes per-basin metrics) and isolation
+ *    (access tokens scope to one basin).
  *
- * Provisioning is one-shot per org: at creation time (or a one-off
- * backfill for existing orgs) we create the basin and stamp
+ * This module is purely retention-string-driven: callers pass a
+ * duration like `"30d"` and the provisioner does the S2 round-trip.
+ * It has no concept of plans / tiers / billing — operators that want
+ * per-tier retention live one layer up (see
+ * `streamBasinRetentionByPlan.server.ts`).
+ *
+ * Provisioning is one-shot per org: at creation time (or via the
+ * backfill worker job for existing orgs) we create the basin and stamp
  * `Organization.streamBasinName`. New `TaskRun` / `Session` rows then
  * piggyback on the existing org read in `triggerTask` / session-create
  * paths and copy the value through. Reads use a precedence chain
  * (`run.streamBasinName ?? session.streamBasinName ?? globalBasin`).
  *
- * Plan changes update retention in-place via `reconfigureBasin`. We do
- * not move data across basins.
+ * Plan / retention changes update retention in-place via
+ * `reconfigureBasin`. We do not move data across basins.
  */
 import type { PrismaClientOrTransaction } from "~/db.server";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 
-/**
- * Plan-tier shorthand for retention mapping. Callers translate the
- * org's billing plan (via `getCurrentPlan`) into one of these and pass
- * it to the provisioner. New orgs (no plan yet) and unbilled orgs
- * default to `free` so we don't accidentally grant a year of retention
- * to a freeloader.
- */
-export type StreamBasinTier = "free" | "hobby" | "pro";
-
-export function retentionFor(tier: StreamBasinTier): string {
-  switch (tier) {
-    case "pro":
-      return env.REALTIME_STREAMS_BASIN_RETENTION_PRO;
-    case "hobby":
-      return env.REALTIME_STREAMS_BASIN_RETENTION_HOBBY;
-    case "free":
-    default:
-      return env.REALTIME_STREAMS_BASIN_RETENTION_FREE;
-  }
+export function isPerOrgBasinsEnabled(): boolean {
+  return env.REALTIME_STREAMS_PER_ORG_BASINS_ENABLED === "true";
 }
 
 /**
- * Permissive plan-name → tier mapping. Billing returns various strings
- * over time (`free_connected`, `hobby`, `team_pro`, `enterprise`, etc.)
- * — be forgiving but predictable.
+ * Default retention for new orgs and any caller that doesn't specify
+ * a value. Configurable via `REALTIME_STREAMS_BASIN_DEFAULT_RETENTION`.
  */
-export function planTierFor(planType: string | null | undefined): StreamBasinTier {
-  if (!planType) return "free";
-  const normalized = planType.toLowerCase();
-  if (normalized.includes("pro") || normalized.includes("team") || normalized.includes("enterprise")) {
-    return "pro";
-  }
-  if (normalized.includes("hobby") || normalized.includes("starter")) {
-    return "hobby";
-  }
-  return "free";
-}
-
-export function isPerOrgBasinsEnabled(): boolean {
-  return env.REALTIME_STREAMS_PER_ORG_BASINS_ENABLED === "true";
+export function defaultRetention(): string {
+  return env.REALTIME_STREAMS_BASIN_DEFAULT_RETENTION;
 }
 
 /**
- * Build the basin name for an org. Format: `{prefix}-{env}-org-{slug}`
- * (e.g. `triggerdotdev-prod-org-acme-corp`). The org slug is already
- * lowercase-and-hyphenated by `createOrganization`, so it satisfies S2
- * basin-name rules without further normalization. We truncate
- * defensively to keep total length under 63 chars (a common bucket
- * convention; verify against S2 docs before raising).
+ * Build the basin name for an org. Format: `{prefix}-{env}-org-{slug}`.
+ * The org slug is already lowercase-and-hyphenated by
+ * `createOrganization`, so it satisfies S2 basin-name rules without
+ * further normalization. We truncate defensively to keep total length
+ * under 63 chars (a common bucket convention; verify against S2 docs
+ * before raising).
+ *
+ * Throws if `REALTIME_STREAMS_BASIN_NAME_PREFIX` +
+ * `REALTIME_STREAMS_BASIN_NAME_ENV` are configured so long that no
+ * room remains for the slug — without this guard, `slice(0, 0)` would
+ * return an empty string and every org would share the same name,
+ * silently colliding via S2's 409-on-create.
  */
 export function basinNameForOrg(org: { slug: string }): string {
   const prefix = env.REALTIME_STREAMS_BASIN_NAME_PREFIX;
   const envName = env.REALTIME_STREAMS_BASIN_NAME_ENV;
   const head = `${prefix}-${envName}-org-`;
   const budget = 63 - head.length;
+  if (budget <= 0) {
+    throw new Error(
+      `[streamBasinProvisioner] REALTIME_STREAMS_BASIN_NAME_PREFIX + REALTIME_STREAMS_BASIN_NAME_ENV too long: head="${head}" leaves no room for the org slug (budget=${budget}). Shorten the prefix or env-name values.`
+    );
+  }
   const slug = org.slug.slice(0, budget);
   return `${head}${slug}`;
 }
 
 type ProvisionInput = {
   id: string;
   slug: string;
-  /// Caller decides the tier. Org-create path passes `"free"` for new
-  /// orgs; the backfill worker resolves the tier via `getCurrentPlan`
-  /// before calling. Defaults to `"free"` if omitted.
-  tier?: StreamBasinTier;
+  /// Duration string passed straight to S2. Defaults to
+  /// `defaultRetention()` when omitted. Caller decides; the provisioner
+  /// has no opinion about what retention is appropriate.
+  retention?: string;
   streamBasinName: string | null | undefined;
 };
 
@@ -109,9 +96,9 @@ type ProvisionResult =
  * success) and writes the column.
  *
  * Failure modes:
- *  - S2 unreachable / 5xx: throws. Callers in the org-create path
- *    should swallow + enqueue a retry job so signup never fails on a
- *    transient S2 outage. The backfill worker retries naturally.
+ *  - S2 unreachable / 5xx / timeout: throws. Callers in the org-create
+ *    path swallow + leave the column null so the backfill worker can
+ *    retry, so signup never fails on a transient S2 outage.
  *  - Auth misconfig (no token): throws. Should never happen in
  *    per-org-basins mode but worth surfacing loudly.
  */
@@ -135,7 +122,7 @@ export async function provisionBasinForOrg(
   }
 
   const basin = basinNameForOrg(org);
-  const retention = retentionFor(org.tier ?? "free");
+  const retention = org.retention ?? defaultRetention();
 
   await s2CreateBasin(basin, {
     accessToken,
@@ -159,13 +146,12 @@ export async function provisionBasinForOrg(
 }
 
 /**
- * Update retention after a plan change. Idempotent. No-op when the
- * org has no provisioned basin. Caller resolves the tier and passes
- * it in — keeps the provisioner ignorant of billing.
+ * Update retention in-place. Idempotent. No-op when the org has no
+ * provisioned basin.
  */
 export async function reconfigureBasinForOrg(
   orgId: string,
-  tier: StreamBasinTier
+  retention: string
 ): Promise<void> {
   if (!isPerOrgBasinsEnabled()) return;
 
@@ -178,7 +164,6 @@ export async function reconfigureBasinForOrg(
   });
   if (!org?.streamBasinName) return;
 
-  const retention = retentionFor(tier);
   await s2ReconfigureBasin(org.streamBasinName, { accessToken, retentionPolicy: retention });
 
   logger.info("[streamBasinProvisioner] reconfigured basin retention", {

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

@@ -0,0 +1,73 @@
+/**
+ * Cloud-flavored shim that resolves a stream-basin retention duration
+ * from an org's current billing plan.
+ *
+ * Kept deliberately separate from `streamBasinProvisioner.server.ts`
+ * so the provisioner stays purely retention-string-driven and has no
+ * coupling to plan vocabulary. This file is the only place in the
+ * webapp that maps "plan code" → "retention duration".
+ *
+ * Operators that don't run a billing API just don't call this — the
+ * provisioner accepts retention strings directly, and the org-create
+ * path falls back to `defaultRetention()`.
+ */
+import { env } from "~/env.server";
+import { getCurrentPlan } from "~/services/platform.v3.server";
+import { defaultRetention } from "./streamBasinProvisioner.server";
+
+/**
+ * Resolve the retention duration for an org based on its current plan.
+ *
+ *  - Returns the configured retention for the plan when the billing
+ *    API has data.
+ *  - Returns `defaultRetention()` when no billing client is configured
+ *    (OSS / non-cloud installs that flipped per-org basins on without
+ *    wiring billing).
+ *  - **Throws** when billing is configured but the call failed, so
+ *    the redis-worker retry kicks in and we don't silently downgrade
+ *    a paid org's retention.
+ */
+export async function resolveRetentionForOrg(orgId: string): Promise<string> {
+  const plan = await getCurrentPlan(orgId);
+
+  if (plan === undefined) {
+    // We can't tell from `getCurrentPlan` alone whether the billing
+    // client isn't configured (OSS) or whether the call failed
+    // (transient cloud outage). Today we conservatively throw so
+    // cloud installs retry. OSS installs that hit this path either:
+    //   (a) flipped the per-org-basins flag on without wiring billing
+    //       and should configure `BILLING_API_URL` / `BILLING_API_KEY`,
+    //       or
+    //   (b) shouldn't be calling this at all and should pass an
+    //       explicit retention to the provisioner.
+    throw new Error(
+      `[streamBasinRetentionByPlan] billing plan unavailable for org ${orgId}; will retry`
+    );
+  }
+
+  return retentionForPlanCode(plan.v3Subscription?.plan?.code);
+}
+
+/**
+ * Map a plan code to a retention duration via env-var lookup.
+ *
+ * Exact-match against a small known set rather than substring matching,
+ * since substring matching against future plan codes could grant the
+ * wrong tier (e.g. `"approved"` would match `"pro"`). Add a new code
+ * here when launching a new plan.
+ */
+export function retentionForPlanCode(code: string | null | undefined): string {
+  if (!code) return defaultRetention();
+
+  switch (code) {
+    case "free":
+      return env.REALTIME_STREAMS_BASIN_RETENTION_FREE;
+    case "v3_hobby_1":
+      return env.REALTIME_STREAMS_BASIN_RETENTION_HOBBY;
+    case "v3_pro_1":
+    case "enterprise":
+      return env.REALTIME_STREAMS_BASIN_RETENTION_PRO;
+    default:
+      return defaultRetention();
+  }
+}