fix(data-drains): handle concurrent delete and DB outage in failure path

Two robustness fixes:
- PUT /data-drains/[drainId]: return 404 if returning() yields no row,
  i.e. a concurrent DELETE landed between loadDrain and the UPDATE.
- runDrain catch block: wrap the failed-status transaction so a DB
  outage during the status write doesn't mask the original delivery
  error. The reaper will eventually rewrite the row to failed.

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

Author: waleedlatif1

apps/sim/app/api/organizations/[id]/data-drains/[drainId]/route.ts

@@ -105,6 +105,11 @@ export const PUT = withRouteHandler(async (request: NextRequest, context: RouteC
     .where(eq(dataDrains.id, drainId))
     .returning()
 
+  if (!updated) {
+    // Concurrent DELETE landed between loadDrain() and this UPDATE.
+    return NextResponse.json({ error: 'Data drain not found' }, { status: 404 })
+  }
+
   logger.info('Data drain updated', { drainId, organizationId })
 
   recordAudit({

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

@@ -160,24 +160,36 @@ export async function runDrain(
   } catch (error) {
     const finishedAt = new Date()
     const message = toError(error).message
-    await db.transaction(async (tx) => {
-      await tx
-        .update(dataDrains)
-        .set({ lastRunAt: finishedAt, updatedAt: finishedAt })
-        .where(eq(dataDrains.id, drainId))
-      await tx
-        .update(dataDrainRuns)
-        .set({
-          status: 'failed',
-          finishedAt,
-          rowsExported,
-          bytesWritten,
-          cursorAfter: cursorBefore, // cursor not advanced on failure
-          locators,
-          error: message.slice(0, 4000),
-        })
-        .where(eq(dataDrainRuns.id, runId))
-    })
+    try {
+      await db.transaction(async (tx) => {
+        await tx
+          .update(dataDrains)
+          .set({ lastRunAt: finishedAt, updatedAt: finishedAt })
+          .where(eq(dataDrains.id, drainId))
+        await tx
+          .update(dataDrainRuns)
+          .set({
+            status: 'failed',
+            finishedAt,
+            rowsExported,
+            bytesWritten,
+            cursorAfter: cursorBefore, // cursor not advanced on failure
+            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.
+      logger.error('Failed to record data drain failure status', {
+        drainId,
+        runId,
+        deliveryError: message,
+        statusError: toError(statusError).message,
+      })
+    }
 
     logger.error('Data drain run failed', {
       drainId,