fix(data-drains): address PR review comments

Author: waleedlatif1

apps/sim/ee/data-drains/destinations/registry.tsx

@@ -211,11 +211,12 @@ const DATADOG_SITE_OPTIONS = [
   { value: 'us5', label: 'US5 (us5.datadoghq.com)' },
   { value: 'eu1', label: 'EU1 (datadoghq.eu)' },
   { value: 'ap1', label: 'AP1 (ap1.datadoghq.com)' },
+  { value: 'ap2', label: 'AP2 (ap2.datadoghq.com)' },
   { value: 'gov', label: 'Gov (ddog-gov.com)' },
 ]
 
 interface DatadogState {
-  site: 'us1' | 'us3' | 'us5' | 'eu1' | 'ap1' | 'gov'
+  site: 'us1' | 'us3' | 'us5' | 'eu1' | 'ap1' | 'ap2' | 'gov'
   service: string
   tags: string
   apiKey: string

apps/sim/lib/api/contracts/data-drains.ts

@@ -52,7 +52,7 @@ const azureBlobCredentialsBodySchema = z.object({
 })
 
 const datadogConfigBodySchema = z.object({
-  site: z.enum(['us1', 'us3', 'us5', 'eu1', 'ap1', 'gov']),
+  site: z.enum(['us1', 'us3', 'us5', 'eu1', 'ap1', 'ap2', 'gov']),
   service: z.string().min(1).max(100).optional(),
   tags: z.string().max(1024).optional(),
 })

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

@@ -1,6 +1,5 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
 import { JWT } from 'google-auth-library'
 import { z } from 'zod'
 import type { DeliveryMetadata, DrainDestination } from '@/lib/data-drains/types'
@@ -216,6 +215,21 @@ const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504])
 const MAX_RETRY_ATTEMPTS = 3
 const BASE_RETRY_DELAY_MS = 250
 
+function sleepUntilAborted(ms: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve()
+  return new Promise((resolve) => {
+    const onAbort = () => {
+      clearTimeout(timeoutId)
+      resolve()
+    }
+    const timeoutId = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    signal.addEventListener('abort', onAbort, { once: true })
+  })
+}
+
 function parseRetryAfter(header: string | null): number | null {
   if (!header) return null
   const seconds = Number(header)
@@ -286,7 +300,8 @@ async function insertAll(input: InsertAllInput): Promise<void> {
     })
     // Drain the body so the connection can be reused.
     await response.text().catch(() => '')
-    await sleep(retryAfterMs)
+    await sleepUntilAborted(retryAfterMs, input.signal)
+    if (input.signal.aborted) throw input.signal.reason ?? new Error('Aborted')
   }
   if (!response.ok) {
     const text = await response.text().catch(() => '')

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

@@ -98,8 +98,25 @@ interface ParsedServiceAccount {
 }
 
 function parseServiceAccount(json: string): ParsedServiceAccount {
-  const parsed = JSON.parse(json) as { client_email: string; private_key: string }
-  return { clientEmail: parsed.client_email, privateKey: parsed.private_key }
+  let parsed: unknown
+  try {
+    parsed = JSON.parse(json)
+  } catch {
+    throw new Error('serviceAccountJson must be valid JSON')
+  }
+  if (typeof parsed !== 'object' || parsed === null) {
+    throw new Error('serviceAccountJson must be a JSON object')
+  }
+  const obj = parsed as Record<string, unknown>
+  const clientEmail = obj.client_email
+  const privateKey = obj.private_key
+  if (typeof clientEmail !== 'string' || clientEmail.length === 0) {
+    throw new Error('serviceAccountJson is missing client_email')
+  }
+  if (typeof privateKey !== 'string' || privateKey.length === 0) {
+    throw new Error('serviceAccountJson is missing private_key')
+  }
+  return { clientEmail, privateKey }
 }
 
 function normalizePrefix(raw: string | undefined): string {