refactor(data-drains): trim extraneous comments and defensive code

Audit cleanup before merge:
- service: drop chunk-empty defensive skip (sources already handle it),
  trim WHAT-comments
- dispatcher: tighten claim-race / rollback / enterprise-cache rationale
  to a single WHY each
- access: collapse the duplicated module-top + inline comments into one
  TSDoc on the gate function
- s3: fix orphaned doc block over assertEndpointIsPublic, soften the
  forcePathStyle TSDoc to match the actual default
- webhook: drop empty close() comment
- docs: clarify that drain reads also require owner/admin, drop the
  "on the dispatcher tick" implementation detail

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: waleedlatif1

apps/docs/content/docs/en/enterprise/data-drains.mdx

@@ -5,7 +5,7 @@ description: Continuously export workflow logs, audit logs, and Mothership data
 
 import { FAQ } from '@/components/ui/faq'
 
-Data Drains let organization owners and admins on Enterprise plans continuously export Sim data to a destination they control — a customer-owned S3 bucket or an HTTPS webhook. A drain runs on a schedule, picks up only new rows since its last successful run, and writes them as NDJSON to the destination.
+Data Drains let organization owners and admins on Enterprise plans continuously export Sim data to a destination they control — a customer-owned S3 bucket or an HTTPS webhook. A drain runs on a schedule, picks up only new rows since its last successful run, and writes them as NDJSON to the destination. Viewing drain configuration and run history is restricted to owners and admins as well, since destinations expose internal bucket names and webhook URLs.
 
 Drains pair naturally with [Data Retention](/enterprise/data-retention): drain into long-term storage first, then let retention safely delete from Sim.
 
@@ -32,7 +32,7 @@ A drain exports exactly one source. To export multiple sources, create multiple
 |---|---|
 | **Workflow logs** | Workflow execution records (one row per execution, only after the run reaches a terminal state). |
 | **Job logs** | Background job records (deployed APIs, schedules, webhooks). Only terminal-state rows are exported. |
-| **Audit logs** | Organization and workspace audit events — logins, permission changes, resource creation/deletion, drain configuration changes. |
+| **Audit logs** | Organization- and workspace-scoped audit events — logins, permission changes, resource creation/deletion, drain configuration changes. |
 | **Copilot chats** | Mothership chat history. |
 | **Copilot runs** | Mothership run records (terminal state only). |
 
@@ -98,7 +98,7 @@ Failed deliveries retry up to 3 times with exponential backoff (500ms, 1s, 2s wi
 
 | Cadence | Drain runs |
 |---|---|
-| **Hourly** | Once per hour, on the dispatcher tick. |
+| **Hourly** | Once per hour. |
 | **Daily** | Once per day. |
 
 You can also disable a drain with the **Enabled** toggle (it stops running but is preserved), or trigger an out-of-schedule run with **Run now** on any drain row.
@@ -127,7 +127,7 @@ The **last 10 runs** for each drain are visible by expanding its row in the sett
 <FAQ items={[
   {
     question: "Who can configure data drains?",
-    answer: "Only organization owners and admins can create, edit, run, or delete drains. On Sim Cloud, the organization must be on an Enterprise plan."
+    answer: "Only organization owners and admins can view, create, edit, run, or delete drains. On Sim Cloud, the organization must be on an Enterprise plan."
   },
   {
     question: "Will drained data be duplicated if a run fails?",

apps/sim/lib/data-drains/access.ts

@@ -6,13 +6,6 @@ import { getSession } from '@/lib/auth'
 import { isOrganizationOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { isBillingEnabled, isDataDrainsEnabled } from '@/lib/core/config/feature-flags'
 
-/**
- * On Sim Cloud (billing enabled), enterprise plan is the gate. On self-hosted
- * deployments, owners opt in by setting `DATA_DRAINS_ENABLED=true` — and
- * routes 404 until they do, so enterprise customers don't accidentally expose
- * mutating endpoints just by deploying a newer image.
- */
-
 export interface DrainAccessSession {
   user: {
     id: string
@@ -29,10 +22,11 @@ export type DrainAccessResult =
   | { ok: false; response: NextResponse }
 
 /**
- * Centralizes the auth + membership + role + enterprise-plan gates that all
- * data-drain routes share. Owner/admin role is required for both reads and
- * writes — drain configs expose customer-controlled bucket names and webhook
- * URLs, which a regular member shouldn't be able to enumerate.
+ * Auth + membership + role + enterprise-plan gate shared by every data-drain
+ * route. Owner/admin role is required for reads as well as writes since drain
+ * configs expose customer bucket names and webhook URLs. On Sim Cloud the
+ * gate is the Enterprise plan; on self-hosted it's `DATA_DRAINS_ENABLED`,
+ * which 404s when unset so a newer image doesn't silently expose drains.
  */
 export async function authorizeDrainAccess(
   organizationId: string,
@@ -59,10 +53,6 @@ export async function authorizeDrainAccess(
     }
   }
 
-  // Feature-flag and enterprise-plan gates apply to reads as well as writes —
-  // drain configs (bucket names, webhook URLs) are sensitive enough that an
-  // org member shouldn't be able to enumerate them on a deployment that
-  // hasn't opted in or after a downgrade off Enterprise.
   if (!isBillingEnabled && !isDataDrainsEnabled) {
     return {
       ok: false,

apps/sim/lib/data-drains/destinations/s3.ts

@@ -32,9 +32,8 @@ const s3ConfigSchema = z.object({
     })
     .optional(),
   /**
-   * Force path-style addressing. MinIO and Ceph RGW need `true`; AWS S3 and
-   * Cloudflare R2 need `false`. No auto-default — explicit choice avoids
-   * silent breakage when a user points at a non-MinIO endpoint.
+   * Force path-style addressing. Set `true` for MinIO / Ceph RGW; defaults
+   * to `false` for AWS S3 and Cloudflare R2.
    */
   forcePathStyle: z.boolean().optional(),
 })
@@ -97,18 +96,12 @@ function isS3ServiceException(error: unknown): error is S3ServiceException {
   )
 }
 
-/**
- * Wraps an S3 SDK call so customers see actionable error codes
- * (`AccessDenied`, `NoSuchBucket`, `InvalidAccessKeyId`,
- * `SignatureDoesNotMatch`, ...) instead of opaque "UnknownError".
- */
 /**
  * Resolves the optional custom endpoint and confirms it does not point at a
  * private, loopback, or cloud-metadata address. The schema-level
  * `validateExternalUrl` only catches IP literals, so a hostname like
  * `evil.example.com` resolving to `169.254.169.254` would slip past it; the
- * AWS SDK then resolves the host itself, bypassing the SSRF guard. This is
- * the same DNS-aware check used by the webhook destination.
+ * AWS SDK then resolves the host itself, bypassing the SSRF guard.
  */
 async function assertEndpointIsPublic(endpoint: string | undefined): Promise<void> {
   if (!endpoint) return
@@ -118,6 +111,11 @@ async function assertEndpointIsPublic(endpoint: string | undefined): Promise<voi
   }
 }
 
+/**
+ * Surfaces actionable S3 SDK error codes (`AccessDenied`, `NoSuchBucket`,
+ * `InvalidAccessKeyId`, `SignatureDoesNotMatch`, ...) and preserves the
+ * original error as `cause` so callers can still branch on `code`/`$metadata`.
+ */
 async function withS3ErrorContext<T>(action: string, fn: () => Promise<T>): Promise<T> {
   try {
     return await fn()
@@ -147,9 +145,8 @@ export const s3Destination: DrainDestination<S3DestinationConfig, S3DestinationC
   async test({ config, credentials, signal }) {
     await assertEndpointIsPublic(config.endpoint)
     const client = buildClient(config, credentials)
-    // Probe with an actual write — HeadBucket only checks read/list and
-    // produces both false positives (read-only creds pass, deliver later
-    // fails) and false negatives (write-only IAM policies fail).
+    // Probe with a real write so read-only creds and write-only IAM policies
+    // surface here instead of at the first scheduled run.
     const probeKey = `${normalizePrefix(config.prefix)}.sim-drain-write-probe/${generateShortId(12)}`
     try {
       await withS3ErrorContext('test-put', () =>

apps/sim/lib/data-drains/destinations/webhook.ts

@@ -256,10 +256,7 @@ export const webhookDestination: DrainDestination<
           ? lastError
           : new Error('Webhook delivery failed after retries')
       },
-      async close() {
-        // No persistent state to release; the pinned-IP agent is created
-        // per-request inside secureFetchWithPinnedIP.
-      },
+      async close() {},
     }
   },
 }

apps/sim/lib/data-drains/dispatcher.ts

@@ -84,11 +84,9 @@ export async function dispatchDueDrains(now: Date = new Date()): Promise<{
     return { candidates: 0, dispatched: 0, skipped: 0, reaped }
   }
 
-  // Cache enterprise checks per-org since we may have several drains per org.
-  // Self-hosted (no billing) deployments have no subscription infrastructure,
-  // so the enterprise predicate would return false for every org and silently
-  // skip every drain — short-circuit to allow when billing is disabled, and
-  // let the cron route's `DATA_DRAINS_ENABLED` gate own the global on/off.
+  // Self-hosted deployments have no subscription infra; `DATA_DRAINS_ENABLED`
+  // is the global on/off there. Cache per-org so a multi-drain org pays one
+  // billing lookup.
   const enterpriseCache = new Map<string, boolean>()
   const isEnterprise = async (orgId: string): Promise<boolean> => {
     if (!isBillingEnabled) return true
@@ -131,27 +129,22 @@ export async function dispatchDueDrains(now: Date = new Date()): Promise<{
       .where(and(eq(dataDrains.id, candidate.id), duePredicate))
       .returning({ id: dataDrains.id })
 
-    if (claimed.length === 0) {
-      // Lost the race — another invocation already claimed this drain.
-      continue
-    }
+    if (claimed.length === 0) continue
 
     try {
-      // concurrencyKey serializes runs of the same drain end-to-end on the
-      // job queue, so even if a manual run-now races with a cron claim the
-      // queue backend won't execute both in parallel.
+      // concurrencyKey serializes runs of the same drain on the job queue, so
+      // a manual run-now racing a cron claim can never execute in parallel.
       await queue.enqueue(
         'run-data-drain',
         { drainId: candidate.id, trigger: 'cron' },
         { concurrencyKey: `data-drain:${candidate.id}` }
       )
       dispatched++
     } catch (error) {
-      // Roll back the claim — otherwise a transient queue outage silently
-      // delays this drain by a full cadence. We only revert if no other
-      // process has advanced lastRunAt past our claim timestamp. Guard the
-      // rollback itself so a transient DB error here doesn't abort the
-      // remaining candidates in the batch.
+      // Roll back the claim so a transient queue outage doesn't delay this
+      // drain by a full cadence. Scoped to our own claim timestamp so it
+      // can't trample a concurrent advance. The rollback itself is guarded
+      // so a DB error here doesn't abort the rest of the batch.
       try {
         await db
           .update(dataDrains)

apps/sim/lib/data-drains/service.ts

@@ -88,7 +88,6 @@ export async function runDrain(
       chunkSize: CHUNK_SIZE,
       signal,
     })) {
-      if (chunk.length === 0) continue
       const ndjson = `${chunk.map((row) => JSON.stringify(source.serialize(row))).join('\n')}\n`
       const body = Buffer.from(ndjson, 'utf8')
 
@@ -177,16 +176,15 @@ export async function runDrain(
             finishedAt,
             rowsExported,
             bytesWritten,
-            cursorAfter: cursorBefore, // cursor not advanced on failure
+            cursorAfter: cursorBefore,
             locators,
             error: message.slice(0, 4000),
           })
           .where(eq(dataDrainRuns.id, runId))
       })
     } catch (statusError) {
-      // Don't let a failed status-update mask the real delivery error —
-      // the reaper will eventually rewrite the row to `failed`. Log so we
-      // can spot DB outages that would otherwise hide behind delivery errors.
+      // Best-effort status write — the reaper repairs stuck rows. Log so DB
+      // outages don't hide behind the original delivery error.
       logger.error('Failed to record data drain failure status', {
         drainId,
         runId,