fix asyncBatching, add debugging

plugins/autocast/src/autocast.plugin.ts

@@ -10,6 +10,7 @@ export function autocast(
   options?: {
     chunkSize?: number
     parallel?: number
+    debug?: boolean
   }
 ) {
   return async (listener: FlatfileListener) => {

plugins/delimiter-extractor/src/index.ts

@@ -22,6 +22,7 @@ interface DelimiterOptions {
   transform?: (value: any) => Flatfile.CellValueUnion
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }
 
 export const DelimiterExtractor = (

plugins/json-extractor/src/index.ts

@@ -4,6 +4,7 @@ import { parseBuffer } from './parser'
 export const JSONExtractor = (options?: {
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }) => {
   return Extractor('.json', 'json', parseBuffer, options)
 }

plugins/psv-extractor/src/index.ts

@@ -7,6 +7,7 @@ export const PSVExtractor = (options?: {
   transform?: (value: any) => Flatfile.CellValueUnion
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }) => {
   return DelimiterExtractor('.psv', { delimiter: '|', ...options })
 }

plugins/record-hook/src/RecordHook.ts

@@ -1,16 +1,24 @@
-import { FlatfileEvent } from '@flatfile/listener'
-import { FlatfileRecord, FlatfileRecords } from '@flatfile/hooks'
 import { Record_, Records } from '@flatfile/api/api'
-import { RecordTranslater } from './record.translater'
+import { FlatfileRecord, FlatfileRecords } from '@flatfile/hooks'
+import { FlatfileEvent } from '@flatfile/listener'
 import { asyncBatch } from '@flatfile/util-common'
+import { RecordTranslater } from './record.translater'
 
 export const RecordHook = async (
   event: FlatfileEvent,
-  handler: (record: FlatfileRecord, event?: FlatfileEvent) => any | Promise<any>
+  handler: (
+    record: FlatfileRecord,
+    event?: FlatfileEvent
+  ) => any | Promise<any>,
+  options: { debug?: boolean } = {}
 ) => {
-  return BulkRecordHook(event, async (records, event) => {
-    return records.map((record) => handler(record, event))
-  })
+  return BulkRecordHook(
+    event,
+    async (records, event) => {
+      return records.map((record) => handler(record, event))
+    },
+    options
+  )
 }
 
 export const BulkRecordHook = async (
@@ -19,7 +27,7 @@ export const BulkRecordHook = async (
     records: FlatfileRecord[],
     event?: FlatfileEvent
   ) => any | Promise<any>,
-  options: { chunkSize?: number; parallel?: number } = {}
+  options: { chunkSize?: number; parallel?: number; debug?: boolean } = {}
 ) => {
   try {
     const records = await event.cache.init<Records>(

plugins/record-hook/src/record.hook.plugin.ts

@@ -1,17 +1,18 @@
-import { FlatfileListener, FlatfileEvent } from '@flatfile/listener'
-import { BulkRecordHook, RecordHook } from './RecordHook'
 import type { FlatfileRecord } from '@flatfile/hooks'
+import { FlatfileEvent, FlatfileListener } from '@flatfile/listener'
+import { BulkRecordHook, RecordHook } from './RecordHook'
 
 export const recordHookPlugin = (
   sheetSlug: string,
   callback: (
     record: FlatfileRecord,
     event?: FlatfileEvent
-  ) => any | Promise<any>
+  ) => any | Promise<any>,
+  options: { debug?: boolean } = {}
 ) => {
   return (client: FlatfileListener) => {
     client.on('commit:created', { sheetSlug }, (event: FlatfileEvent) => {
-      return RecordHook(event, callback)
+      return RecordHook(event, callback, options)
     })
   }
 }
@@ -22,7 +23,7 @@ export const bulkRecordHookPlugin = (
     records: FlatfileRecord[],
     event?: FlatfileEvent
   ) => any | Promise<any>,
-  options: { chunkSize?: number; parallel?: number } = {}
+  options: { chunkSize?: number; parallel?: number; debug?: boolean } = {}
 ) => {
   return (client: FlatfileListener) => {
     client.on('commit:created', { sheetSlug }, (event: FlatfileEvent) => {
@@ -32,6 +33,6 @@ export const bulkRecordHookPlugin = (
 }
 
 export {
-  recordHookPlugin as recordHook,
   bulkRecordHookPlugin as bulkRecordHook,
+  recordHookPlugin as recordHook,
 }

plugins/tsv-extractor/src/index.ts

@@ -7,6 +7,7 @@ export const TSVExtractor = (options?: {
   transform?: (value: any) => Flatfile.CellValueUnion
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }) => {
   return DelimiterExtractor('.tsv', { delimiter: '\t', ...options })
 }

plugins/xlsx-extractor/src/index.ts

@@ -5,6 +5,7 @@ export const ExcelExtractor = (options?: {
   rawNumbers?: boolean
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }) => {
   return Extractor(
     /\.(xlsx?|xlsm|xlsb|xltx?|xltm)$/i,

plugins/xml-extractor/src/index.ts

@@ -7,6 +7,7 @@ export const XMLExtractor = (options?: {
   transform?: (row: Record<string, any>) => Record<string, any>
   chunkSize?: number
   parallel?: number
+  debug?: boolean
 }) => {
   return Extractor('.xml', 'xml', parseBuffer, options)
 }

utils/common/src/async.batch.ts

@@ -3,57 +3,50 @@ import { FlatfileEvent } from '@flatfile/listener'
 export async function asyncBatch<T, R>(
   arr: T[],
   callback: (chunk: T[], event?: FlatfileEvent) => Promise<R>,
-  options: { chunkSize?: number; parallel?: number } = {},
+  options: { chunkSize?: number; parallel?: number; debug?: boolean } = {},
   event?: FlatfileEvent
 ): Promise<R[]> {
-  const { chunkSize, parallel } = { chunkSize: 3000, parallel: 1, ...options }
-  const results: R[] = []
-
-  // Split the array into chunks
-  const chunks: T[][] = []
-  for (let i = 0; i < arr.length; i += chunkSize) {
-    chunks.push(arr.slice(i, i + chunkSize))
+  const { chunkSize = 3000, parallel = 1, debug = false } = options
+  const chunks = Array.from(
+    { length: Math.ceil(arr.length / chunkSize) },
+    (_, i) => arr.slice(i * chunkSize, i * chunkSize + chunkSize)
+  )
+
+  if (debug) {
+    console.log(`${chunks.length} chunks to be processed`)
   }
 
-  // Create a helper function to process a chunk
-  async function processChunk(chunk: T[]): Promise<void> {
-    const result = await callback(chunk, event)
-    results.push(result)
-  }
+  const results: Map<number, R> = new Map()
 
-  // Execute the chunks in parallel
-  const promises: Promise<void>[] = []
-  let running = 0
-  let currentIndex = 0
-
-  function processNext(): void {
-    if (currentIndex >= chunks.length) {
-      // All chunks have been processed
-      return
+  async function processChunk(
+    chunkIndex: number,
+    threadId: number
+  ): Promise<void> {
+    if (debug) {
+      console.log(`Thread ${threadId} processing chunk ${chunkIndex}`)
     }
 
-    const currentChunk = chunks[currentIndex]
-    const promise = processChunk(currentChunk).finally(() => {
-      running--
-      processNext() // Process next chunk
-    })
-
-    promises.push(promise)
-    currentIndex++
-    running++
+    const result = await callback(chunks[chunkIndex], event)
+    results.set(chunkIndex, result)
+  }
 
-    if (running < parallel) {
-      processNext() // Process another chunk if available
+  let currentIndex = 0
+  async function processChunks(threadId: number): Promise<void> {
+    while (currentIndex < chunks.length) {
+      const chunkIndex = currentIndex++
+      await processChunk(chunkIndex, threadId)
     }
   }
 
-  // Start processing the chunks
-  for (let i = 0; i < parallel && i < chunks.length; i++) {
-    processNext()
-  }
+  const promises: Promise<void>[] = Array.from({ length: parallel }, (_, i) =>
+    processChunks(i)
+  )
 
-  // Wait for all promises to resolve
   await Promise.all(promises)
 
-  return results
+  if (debug) {
+    console.log('All chunks processed')
+  }
+
+  return Array.from(results.values())
 }

utils/extractor/src/index.ts

@@ -59,9 +59,10 @@ export const Extractor = (
             progress: 50,
             info: 'Adding records to Sheets',
           })
-          const { chunkSize, parallel } = {
+          const { chunkSize, parallel, debug } = {
             chunkSize: 3000,
             parallel: 1,
+            debug: false,
             ...options,
           }
           for (const sheet of workbook.sheets) {
@@ -73,7 +74,7 @@ export const Extractor = (
               async (chunk) => {
                 await api.records.insert(sheet.id, chunk)
               },
-              { chunkSize, parallel }
+              { chunkSize, parallel, debug }
             )
           }
           await api.files.update(file.id, {