Function runDrain

apps/sim/lib/data-drains/service.ts:34–227

(
  drainId: string,
  trigger: RunTrigger,
  options: { signal?: AbortSignal } = {}
)

 * dedupe on the per-row `id` field.
 */
export async function runDrain(
  drainId: string,
  trigger: RunTrigger,
  options: { signal?: AbortSignal } = {}
): Promise<RunDrainResult> {
  const signal = options.signal ?? new AbortController().signal
  const [drain] = await db.select().from(dataDrains).where(eq(dataDrains.id, drainId)).limit(1)
  if (!drain) {
    throw new Error(`Data drain not found: ${drainId}`)
  }
  if (!drain.enabled) {
    return {
      drainId,
      runId: '',
      status: 'skipped',
      rowsExported: 0,
      bytesWritten: 0,
      cursorBefore: drain.cursor,
      cursorAfter: drain.cursor,
      locators: [],
    }
  }

  const source = getSource(drain.source)
  const destination = getDestination(drain.destinationType)

  const runId = generateId()
  const startedAt = new Date()
  await db.insert(dataDrainRuns).values({
    id: runId,
    drainId,
    status: 'running',
    trigger,
    startedAt,
    cursorBefore: drain.cursor,
  })

  const cursorBefore = drain.cursor
  let cursor: Cursor = drain.cursor
  let rowsExported = 0
  let bytesWritten = 0
  let sequence = 0
  const locators: string[] = []

  /**
   * 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 })