refactor: stream batched items to db during backfill (#229)

Streams and upserts in chunks as items arrive, instead of collecting all first.

- Chunks of 250 items
- Upsert runs while fetch continues
- Logging: removes mid-sync count log (total unknown until done)

Author: cescox

packages/sync-engine/src/stripeSync.ts

@@ -1009,25 +1009,27 @@ export class StripeSync {
     fetch: () => Stripe.ApiListPromise<T>,
     upsert: (items: T[]) => Promise<T[]>
   ): Promise<Sync> {
-    const items: T[] = []
+    const chunkSize = 250
+    let chunk: T[] = []
+    let synced = 0
 
     this.config.logger?.info('Fetching items to sync from Stripe')
     for await (const item of fetch()) {
-      items.push(item)
-    }
+      chunk.push(item)
+      synced++
+      if (synced % 1000 === 0) this.config.logger?.info(`Synced ${synced} items`)
 
-    if (!items.length) return { synced: 0 }
+      if (chunk.length >= chunkSize) {
+        await upsert(chunk)
+        chunk = []
+      }
+    }
 
-    this.config.logger?.info(`Upserting ${items.length} items`)
-    const chunkSize = 250
-    for (let i = 0; i < items.length; i += chunkSize) {
-      const chunk = items.slice(i, i + chunkSize)
+    if (chunk.length > 0) await upsert(chunk)
 
-      await upsert(chunk)
-    }
-    this.config.logger?.info('Upserted items')
+    this.config.logger?.info(`Upserted ${synced} items`)
 
-    return { synced: items.length }
+    return { synced }
   }
 
   private async upsertCharges(