fix(data-drains): validate JSON client-side for Snowflake before binding

Switch Snowflake to parseNdjsonObjects so malformed rows are caught locally
with 1-indexed line numbers instead of failing the whole INSERT server-side.
Re-stringify each parsed object before binding to PARSE_JSON(?).
Drop the now-unused parseNdjsonLines helper.

Author: waleedlatif1

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

@@ -6,7 +6,7 @@ import { importPKCS8, SignJWT } from 'jose'
 import { z } from 'zod'
 import {
   backoffWithJitter,
-  parseNdjsonLines,
+  parseNdjsonObjects,
   parseRetryAfter,
   sleepUntilAborted,
 } from '@/lib/data-drains/destinations/utils'
@@ -411,7 +411,8 @@ export const snowflakeDestination: DrainDestination<
     }
     return {
       async deliver({ body, metadata, signal }) {
-        const rows = parseNdjsonLines(body)
+        const parsed = parseNdjsonObjects(body)
+        const rows = parsed.map((row) => JSON.stringify(row))
         if (rows.length === 0) {
           return {
             locator: `snowflake://${config.account}/${config.database}.${config.schema}.${config.table}#${metadata.runId}-${metadata.sequence}`,

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

@@ -154,20 +154,6 @@ export function parseServiceAccount(json: string): ParsedServiceAccount {
  * is parseable and carries `client_email` + `private_key`. Used by both
  * `gcsCredentialsSchema` and `bigqueryCredentialsSchema`.
  */
-/**
- * Splits an NDJSON buffer into its non-empty lines (UTF-8). Used by destinations
- * that ship raw JSON strings (e.g. Snowflake binds each line as a TEXT value).
- */
-export function parseNdjsonLines(body: Buffer): string[] {
-  const text = body.toString('utf8')
-  const out: string[] = []
-  for (const line of text.split(/\r?\n/)) {
-    if (line.length === 0) continue
-    out.push(line)
-  }
-  return out
-}
-
 export interface ParseNdjsonObjectsOptions {
   /** When true, throw if a parsed value is not a plain object. */
   requireObject?: boolean