feat(webapp): MOLLIFIER_DRAINER_ENABLED for per-service drainer control

The drainer's polling loop has been gated on WORKER_ENABLED, which
couples it to the legacy ZodWorker role. To split the drainer onto a
dedicated worker service in cloud (and keep all other replicas as
producer-only), introduce its own switch.

Semantics:
  - Unset                              → inherits MOLLIFIER_ENABLED.
    Single-container self-hosters with MOLLIFIER_ENABLED=1 get the
    drainer for free, no second flag to remember.
  - Explicit MOLLIFIER_DRAINER_ENABLED=0 → drainer off on this replica.
    Cloud sets this everywhere except the dedicated drainer service.
  - Explicit MOLLIFIER_DRAINER_ENABLED=1 → drainer on, subject to
    MOLLIFIER_ENABLED still being the master kill switch (a drainer
    can't construct without the gate-side buffer singleton).

The bootstrap in mollifierDrainerWorker.server.ts now gates on the new
flag instead of WORKER_ENABLED, so the drainer's lifecycle is no longer
coupled to the legacy worker role.

Author: d-cs

apps/webapp/app/env.server.ts

@@ -1052,7 +1052,46 @@ const EnvironmentSchema = z
       .optional()
       .transform((v) => v ?? process.env.REDIS_PASSWORD),
     COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
-    COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+    MOLLIFIER_ENABLED: z.string().default("0"),
+    // Separate switch for the drainer (consumer side) so it can be split
+    // off onto a dedicated worker service. Unset → inherits
+    // MOLLIFIER_ENABLED, so single-container self-hosters don't have to
+    // flip two switches. In multi-replica deployments, set this to "0"
+    // explicitly on every replica except the one dedicated drainer
+    // service — otherwise every replica's polling loop races for the
+    // same buffer entries. `MOLLIFIER_ENABLED` is still the master kill
+    // switch; setting this to "1" while `MOLLIFIER_ENABLED` is "0" is a
+    // no-op because the gate-side singleton refuses to construct a
+    // buffer when the system is off.
+    MOLLIFIER_DRAINER_ENABLED: z.string().default(process.env.MOLLIFIER_ENABLED ?? "0"),
+    MOLLIFIER_SHADOW_MODE: z.string().default("0"),
+    MOLLIFIER_REDIS_HOST: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_HOST),
+    MOLLIFIER_REDIS_PORT: z.coerce
+      .number()
+      .optional()
+      .transform(
+        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined),
+      ),
+    MOLLIFIER_REDIS_USERNAME: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_USERNAME),
+    MOLLIFIER_REDIS_PASSWORD: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_PASSWORD),
+    MOLLIFIER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+    MOLLIFIER_TRIP_WINDOW_MS: z.coerce.number().int().positive().default(200),
+    MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().default(100),
+    MOLLIFIER_HOLD_MS: z.coerce.number().int().positive().default(500),
+    MOLLIFIER_DRAIN_CONCURRENCY: z.coerce.number().int().positive().default(50),
+    MOLLIFIER_ENTRY_TTL_S: z.coerce.number().int().positive().default(600),
+    MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
+    MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(30_000),
+    MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK: z.coerce.number().int().positive().default(500),
 
     BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
       .number()

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

@@ -29,15 +29,18 @@ declare global {
  * `batchTriggerWorker`).
  *
  * Gating order:
- *   - `WORKER_ENABLED !== "true"`  → early return (API-only replicas
- *     still produce into the buffer via the trigger hot path; only worker
- *     replicas drain it, otherwise every replica races for the same
- *     entries).
+ *   - `MOLLIFIER_DRAINER_ENABLED !== "1"`  → early return. Unset defaults
+ *     to `MOLLIFIER_ENABLED`, so single-container self-hosters still get
+ *     the drainer for free with one flag. In multi-replica deployments,
+ *     set this to "0" explicitly on every replica except the dedicated
+ *     drainer service so the polling loop doesn't race across replicas.
  *   - `MOLLIFIER_ENABLED !== "1"`  → `getMollifierDrainer()` returns null
- *     and the bootstrap is a no-op.
+ *     and the bootstrap is a no-op. `MOLLIFIER_ENABLED` remains the
+ *     master kill switch; the new flag only controls WHICH replicas
+ *     run the drainer when the system is on.
  */
 export function initMollifierDrainerWorker(): void {
-  if (env.WORKER_ENABLED !== "true") {
+  if (env.MOLLIFIER_DRAINER_ENABLED !== "1") {
     return;
   }