Function insertAll

apps/sim/lib/data-drains/destinations/bigquery.ts:162–276

* Streams a chunk of rows to `tabledata.insertAll`. * * Partial-success caveat: BigQuery may return HTTP 200 with a non-empty * `insertErrors` array. Rows not listed there are inserted and dedup-keyed by * `insertId` for ~60s. We throw on any `insertErrors`; retries within the * dedup window ar

(input: InsertAllInput)

 * dedup window are safe, but retries after it may duplicate succeeded rows.
 */
async function insertAll(input: InsertAllInput): Promise<void> {
  if (input.rows.length > MAX_ROWS_PER_REQUEST) {
    throw new Error(
      `BigQuery insertAll chunk has ${input.rows.length} rows, exceeds the ${MAX_ROWS_PER_REQUEST} per-request limit`
    )
  }
  const url = `https://bigquery.googleapis.com/bigquery/v2/projects/${encodeURIComponent(input.config.projectId)}/datasets/${encodeURIComponent(input.config.datasetId)}/tables/${encodeURIComponent(input.config.tableId)}/insertAll`
  /**
   * `skipInvalidRows: false` and `ignoreUnknownValues: false` surface schema
   * mismatches as `insertErrors` instead of silently dropping data — drains
   * should fail loudly so operators notice the schema drift.
   */
  const payload = {
    skipInvalidRows: false,
    ignoreUnknownValues: false,
    rows: input.rows.map((row, index) => {
      const rowBytes = Buffer.byteLength(JSON.stringify(row), 'utf8')
      if (rowBytes > MAX_ROW_BYTES) {
        throw new Error(
          `BigQuery row at index ${index} is ${rowBytes} bytes, exceeds the ${MAX_ROW_BYTES}-byte per-row limit`
        )
      }
      return {
        insertId: buildInsertId(input.metadata, index),
        json: row,
      }
    }),
  }
  const body = JSON.stringify(payload)
  const byteLength = Buffer.byteLength(body, 'utf8')
  if (byteLength > MAX_REQUEST_BYTES) {
    throw new Error(
      `BigQuery insertAll body is ${byteLength} bytes, exceeds the ${MAX_REQUEST_BYTES}-byte per-request limit`
    )
  }
  let attempt = 0
  let response: Response | undefined
  let refreshedOnce = false
  while (true) {
    attempt++
    try {
      response = await postInsertAll(input, url, body)
      /** A 401 retry doesn't count against the 5xx/429 budget — token refresh is a one-shot recovery. */
      if (response.status === 401 && !refreshedOnce) {
        refreshedOnce = true
        logger.debug('BigQuery returned 401; refreshing access token and retrying once')
        /** Drain the 401 body before discarding so undici can return the socket to the keep-alive pool. */
        await response.text().catch(() => '')
        response = await postInsertAll(input, url, body, true)
      }
      if (!RETRYABLE_STATUSES.has(response.status)) break
      if (attempt >= MAX_RETRY_ATTEMPTS) break
      const retryAfterHeaderMs = parseRetryAfter(response.headers.get('retry-after'))
      const retryAfterMs = backoffWithJitter(attempt, retryAfterHeaderMs, {
        baseMs: BASE_RETRY_DELAY_MS,
      })
      logger.warn('BigQuery insertAll transient error; retrying', {
        status: response.status,