fix(data-drains): bugbot fixes for cursor precision, invisible failures, audit-log index

- cursor: compare in millisecond buckets so PG's microsecond-precision
  timestamps don't cause the cursor row to re-emit forever. The JS Date
  round-trip truncates 00:00:00.123456 to 00:00:00.123, which made
  gt(col, cursor) match the cursor row itself.
- service: insert the run row before parse/decrypt so encryption-key
  rotation or schema drift surface as a failed run in the UI instead of
  vanishing into background-job logs while lastRunAt advances.
- audit_log: add (workspace_id, created_at, id) composite index so the
  audit-logs source's tie-breaking ORDER BY is satisfied by the index
  without a heap fetch.

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

Author: waleedlatif1

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

@@ -56,10 +56,6 @@ export async function runDrain(
 
   const source = getSource(drain.source)
   const destination = getDestination(drain.destinationType)
-  const config = destination.configSchema.parse(drain.destinationConfig)
-  const credentials = destination.credentialsSchema.parse(
-    await decryptCredentials(drain.destinationCredentials)
-  )
 
   const runId = generateId()
   const startedAt = new Date()
@@ -79,9 +75,21 @@ export async function runDrain(
   let sequence = 0
   const locators: string[] = []
 
-  const session = destination.openSession({ config, credentials })
+  /**
+   * Schema-parse and decrypt happen *after* the run row is created so failures
+   * in either (e.g. encryption-key rotation, schema drift across versions)
+   * surface as a `failed` run row in the UI rather than vanishing into the
+   * background-job logs while `lastRunAt` quietly advances.
+   */
+  let session: ReturnType<typeof destination.openSession> | null = null
 
   try {
+    const config = destination.configSchema.parse(drain.destinationConfig)
+    const credentials = destination.credentialsSchema.parse(
+      await decryptCredentials(drain.destinationCredentials)
+    )
+    session = destination.openSession({ config, credentials })
+
     for await (const chunk of source.pages({
       organizationId: drain.organizationId,
       cursor,
@@ -203,14 +211,16 @@ export async function runDrain(
 
     throw error
   } finally {
-    try {
-      await session.close()
-    } catch (closeError) {
-      logger.warn('Data drain session close failed', {
-        drainId,
-        runId,
-        error: toError(closeError).message,
-      })
+    if (session) {
+      try {
+        await session.close()
+      } catch (closeError) {
+        logger.warn('Data drain session close failed', {
+          drainId,
+          runId,
+          error: toError(closeError).message,
+        })
+      }
     }
   }
 }

apps/sim/lib/data-drains/sources/cursor.ts

@@ -1,4 +1,4 @@
-import { and, eq, gt, or, type SQL } from 'drizzle-orm'
+import { and, gt, gte, lt, or, type SQL } from 'drizzle-orm'
 import type { PgColumn } from 'drizzle-orm/pg-core'
 import type { Cursor } from '@/lib/data-drains/types'
 
@@ -29,17 +29,29 @@ export function decodeTimeCursor(cursor: Cursor): TimeCursor | null {
 
 /**
  * Builds a strict-greater-than predicate over a `(timestampCol, idCol)` pair.
- * Equivalent to `(timestampCol, idCol) > (cursor.ts, cursor.id)` — rows on the
- * cursor itself are excluded; ties on `timestampCol` break by `idCol`.
+ * Semantically: `(timestampCol, idCol) > (cursor.ts, cursor.id)`.
+ *
+ * Postgres `timestamp` columns store microsecond precision but JS `Date`
+ * round-trips at millisecond precision, so a row with PG value
+ * `00:00:00.123456` survives the JS → cursor → JS round-trip as
+ * `00:00:00.123000`. A naive `eq(col, new Date(cursor.ts))` would never match
+ * such rows, and `gt(col, new Date(cursor.ts))` would match the cursor row
+ * itself — re-emitting the last row of every chunk forever.
+ *
+ * Fix: compare in millisecond buckets. Anything in a strictly later ms bucket
+ * is included; anything inside the cursor's own ms bucket is included only
+ * when `idCol > cursor.id`.
  */
 export function timeCursorPredicate(
   timestampCol: PgColumn,
   idCol: PgColumn,
   cursor: TimeCursor | null
 ): SQL | undefined {
   if (!cursor) return undefined
+  const cursorTs = new Date(cursor.ts)
+  const nextMs = new Date(cursorTs.getTime() + 1)
   return or(
-    gt(timestampCol, new Date(cursor.ts)),
-    and(eq(timestampCol, new Date(cursor.ts)), gt(idCol, cursor.id))
+    gte(timestampCol, nextMs),
+    and(gte(timestampCol, cursorTs), lt(timestampCol, nextMs), gt(idCol, cursor.id))
   )
 }

packages/db/migrations/0202_panoramic_dreaming_celestial.sql

@@ -46,4 +46,5 @@ CREATE UNIQUE INDEX "data_drains_org_name_unique" ON "data_drains" USING btree (
 CREATE INDEX "copilot_chats_workspace_created_at_id_idx" ON "copilot_chats" USING btree ("workspace_id","created_at","id");--> statement-breakpoint
 CREATE INDEX "copilot_runs_workspace_completed_at_id_idx" ON "copilot_runs" USING btree ("workspace_id","completed_at","id");--> statement-breakpoint
 CREATE INDEX "job_execution_logs_workspace_ended_at_id_idx" ON "job_execution_logs" USING btree ("workspace_id","ended_at","id");--> statement-breakpoint
-CREATE INDEX "workflow_execution_logs_workspace_ended_at_id_idx" ON "workflow_execution_logs" USING btree ("workspace_id","ended_at","id");
+CREATE INDEX "workflow_execution_logs_workspace_ended_at_id_idx" ON "workflow_execution_logs" USING btree ("workspace_id","ended_at","id");--> statement-breakpoint
+CREATE INDEX "audit_log_workspace_created_at_id_idx" ON "audit_log" USING btree ("workspace_id","created_at","id");

packages/db/migrations/meta/0202_snapshot.json

@@ -1259,6 +1259,33 @@
           "method": "btree",
           "with": {}
         },
+        "audit_log_workspace_created_at_id_idx": {
+          "name": "audit_log_workspace_created_at_id_idx",
+          "columns": [
+            {
+              "expression": "workspace_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "created_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "audit_log_actor_created_idx": {
           "name": "audit_log_actor_created_idx",
           "columns": [

packages/db/schema.ts

@@ -2416,6 +2416,11 @@ export const auditLog = pgTable(
       table.workspaceId,
       table.createdAt
     ),
+    workspaceCreatedIdIdx: index('audit_log_workspace_created_at_id_idx').on(
+      table.workspaceId,
+      table.createdAt,
+      table.id
+    ),
     actorCreatedIdx: index('audit_log_actor_created_idx').on(table.actorId, table.createdAt),
     resourceIdx: index('audit_log_resource_idx').on(table.resourceType, table.resourceId),
     actionIdx: index('audit_log_action_idx').on(table.action),