fix(data-drains): address PR review thread fixes

- service: throw on cancellation after pages loop so a run aborted mid-stream
  isn't recorded as success
- audit-logs: include org-scoped rows (workspace_id IS NULL with
  metadata->>organizationId match) alongside workspace rows
- access: require owner/admin for read routes too; drain configs leak bucket
  names and webhook URLs

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

Author: waleedlatif1

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

@@ -30,9 +30,9 @@ export type DrainAccessResult =
 
 /**
  * Centralizes the auth + membership + role + enterprise-plan gates that all
- * data-drain routes share. `requireMutating` flag enforces owner/admin role
- * and enterprise plan; read-only callers can pass `false` to allow any
- * organization member.
+ * 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.
  */
 export async function authorizeDrainAccess(
   organizationId: string,
@@ -84,15 +84,17 @@ export async function authorizeDrainAccess(
       }
     }
   }
-  if (options.requireMutating) {
-    if (memberEntry.role !== 'owner' && memberEntry.role !== 'admin') {
-      return {
-        ok: false,
-        response: NextResponse.json(
-          { error: 'Forbidden - Only organization owners and admins can manage data drains' },
-          { status: 403 }
-        ),
-      }
+  if (memberEntry.role !== 'owner' && memberEntry.role !== 'admin') {
+    return {
+      ok: false,
+      response: NextResponse.json(
+        {
+          error: options.requireMutating
+            ? 'Forbidden - Only organization owners and admins can manage data drains'
+            : 'Forbidden - Only organization owners and admins can view data drains',
+        },
+        { status: 403 }
+      ),
     }
   }
 

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

@@ -113,6 +113,10 @@ export async function runDrain(
       sequence++
     }
 
+    if (signal.aborted) {
+      throw new Error('Data drain run cancelled')
+    }
+
     const finishedAt = new Date()
     await db.transaction(async (tx) => {
       await tx

apps/sim/lib/data-drains/sources/audit-logs.ts

@@ -1,6 +1,6 @@
 import { db } from '@sim/db'
 import { auditLog } from '@sim/db/schema'
-import { and, asc, inArray } from 'drizzle-orm'
+import { and, asc, inArray, isNull, or, sql } from 'drizzle-orm'
 import {
   decodeTimeCursor,
   encodeTimeCursor,
@@ -12,15 +12,23 @@ import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/typ
 type AuditLogRow = typeof auditLog.$inferSelect
 
 /**
- * Drains workspace-scoped audit events for every workspace owned by the
- * organization. Org-scoped events (rows with `workspace_id IS NULL`, e.g.
- * org-level invitations) are intentionally out of scope — `audit_log` has no
- * `organization_id` column to filter on, so cleanly attributing them is not
- * possible without a schema change. Track separately if needed.
+ * Drains audit events scoped to the organization: rows from any of the org's
+ * workspaces, plus org-level rows (`workspace_id IS NULL`) where
+ * `metadata->>'organizationId'` matches. Audit-log writers consistently set
+ * `metadata.organizationId` for org-scoped actions even though the table has
+ * no dedicated FK column.
  */
 async function* pages(input: SourcePageInput): AsyncIterable<AuditLogRow[]> {
   const workspaceIds = await getOrganizationWorkspaceIds(input.organizationId)
-  if (workspaceIds.length === 0) return
+
+  const orgScopedClause = and(
+    isNull(auditLog.workspaceId),
+    sql`${auditLog.metadata}->>'organizationId' = ${input.organizationId}`
+  )
+  const scopeClause =
+    workspaceIds.length === 0
+      ? orgScopedClause
+      : or(inArray(auditLog.workspaceId, workspaceIds), orgScopedClause)
 
   let cursor = decodeTimeCursor(input.cursor)
   while (!input.signal.aborted) {
@@ -29,7 +37,7 @@ async function* pages(input: SourcePageInput): AsyncIterable<AuditLogRow[]> {
     const rows = await db
       .select()
       .from(auditLog)
-      .where(and(inArray(auditLog.workspaceId, workspaceIds), cursorClause))
+      .where(and(scopeClause, cursorClause))
       .orderBy(asc(auditLog.createdAt), asc(auditLog.id))
       .limit(input.chunkSize)