sheets

Committed-By-Agent: claude

Author: sgresh-stripe

apps/engine/src/api/app.ts

@@ -20,8 +20,8 @@ import {
 import { takeStateCheckpoints } from '../lib/pipeline.js'
 import { ndjsonResponse } from '@stripe/sync-ts-cli/ndjson'
 import { logger } from '../logger.js'
-import { createStripeSource, DEFAULT_MAX_RPS } from '@stripe/sync-source-stripe'
-import type { RateLimiter } from '@stripe/sync-source-stripe'
+import { createStripeSource, DEFAULT_MAX_RPS, fetchById } from '@stripe/sync-source-stripe'
+import type { RateLimiter, Config as StripeConfig } from '@stripe/sync-source-stripe'
 import {
   acquire,
   createRateLimiterTable,
@@ -208,8 +208,12 @@ export function createApp(resolver: ConnectorResolver) {
     logger.info(context, 'Engine API /setup started')
     const engine = await createEngineFromParams(params.pipeline, resolver, noopStateStore())
     try {
-      await engine.setup()
+      const result = await engine.setup()
       logger.info({ ...context, durationMs: Date.now() - startedAt }, 'Engine API /setup completed')
+      logger.info(result, 'Engine API /setup result')
+      if (result && Object.keys(result).length > 0) {
+        return c.json(result, 200)
+      }
       return c.body(null, 204)
     } catch (error) {
       logger.error(
@@ -276,6 +280,74 @@ export function createApp(resolver: ConnectorResolver) {
     )
   })
 
+  app.post('/write-events', async (c) => {
+    const params = parseSyncParams(c)
+    const context = { path: '/write-events', ...syncRequestContext(params) }
+    const startedAt = Date.now()
+
+    // Body: [{stream, id, row_number?}]
+    const ThinEventSchema = z.array(
+      z.object({
+        stream: z.string(),
+        id: z.string(),
+        row_number: z.number().int().positive().optional(),
+      })
+    )
+    let events: z.infer<typeof ThinEventSchema>
+    try {
+      events = ThinEventSchema.parse(await c.req.json())
+    } catch {
+      return c.json({ error: 'Body must be an array of {stream, id, row_number?}' }, 400)
+    }
+    if (events.length === 0) {
+      return c.json({ error: 'Events array must not be empty' }, 400)
+    }
+
+    logger.info({ ...context, eventCount: events.length }, 'Engine API /write-events started')
+
+    const sourceConfig = params.pipeline.source as unknown as StripeConfig
+
+    // Fetch full objects from Stripe concurrently then build record messages
+    const now = Date.now()
+    const fetched = await Promise.all(
+      events.map(async (event) => {
+        const data = await fetchById(sourceConfig, event.stream, event.id)
+        return { event, data }
+      })
+    )
+
+    // Log any IDs that couldn't be resolved
+    for (const { event, data } of fetched) {
+      if (!data) {
+        logger.warn(
+          { stream: event.stream, id: event.id },
+          'Engine API /write-events: fetchById returned null — skipping'
+        )
+      }
+    }
+
+    const records = fetched.filter(({ data }) => data != null)
+
+    async function* recordMessages(): AsyncIterable<
+      import('@stripe/sync-protocol').DestinationInput
+    > {
+      for (const { event, data } of records) {
+        yield {
+          type: 'record' as const,
+          stream: event.stream,
+          data: data!,
+          emitted_at: now,
+          row_number: event.row_number,
+        }
+      }
+    }
+
+    const engine = await createEngineFromParams(params.pipeline, resolver, noopStateStore())
+    return ndjsonResponse(
+      logApiStream('Engine API /write-events', engine.write(recordMessages()), context, startedAt)
+    )
+  })
+
   app.post('/sync', async (c) => {
     const params = parseSyncParams(c)
     const stateStore = await selectStateStore(params.pipeline)

apps/engine/src/lib/engine.ts

@@ -18,7 +18,7 @@ import { logger } from '../logger.js'
 // MARK: - Engine interface
 
 export interface Engine {
-  setup(): Promise<void>
+  setup(): Promise<Record<string, unknown> | undefined>
   teardown(): Promise<void>
   check(): Promise<{ source: CheckResult; destination: CheckResult }>
   read(input?: AsyncIterable<unknown>): AsyncIterable<Message>
@@ -168,22 +168,34 @@ export function createEngine(
     return _catalog
   }
 
+  async function setupSource(catalog: ConfiguredCatalog): Promise<void> {
+    if (connectors.source.setup) {
+      await withLoggedStep('Engine source setup', baseContext, () =>
+        connectors.source.setup!({ config: sourceConfig, catalog })
+      )
+    }
+  }
+
+  async function setupDestination(
+    catalog: ConfiguredCatalog
+  ): Promise<Record<string, unknown> | undefined> {
+    if (connectors.destination.setup) {
+      const result = await withLoggedStep('Engine destination setup', baseContext, () =>
+        connectors.destination.setup!({ config: destConfig, catalog })
+      )
+      return result ?? undefined
+    }
+  }
+
   return {
     async setup() {
       const catalog = await getCatalog()
       const filteredCatalog = applySelection(catalog)
-      await Promise.all([
-        connectors.source.setup
-          ? withLoggedStep('Engine source setup', baseContext, () =>
-              connectors.source.setup!({ config: sourceConfig, catalog })
-            )
-          : Promise.resolve(),
-        connectors.destination.setup
-          ? withLoggedStep('Engine destination setup', baseContext, () =>
-              connectors.destination.setup!({ config: destConfig, catalog: filteredCatalog })
-            )
-          : Promise.resolve(),
+      const [, destResult] = await Promise.all([
+        setupSource(catalog),
+        setupDestination(filteredCatalog),
       ])
+      return destResult ?? undefined
     },
 
     async teardown() {
@@ -244,7 +256,14 @@ export function createEngine(
     },
 
     async *sync(input?: AsyncIterable<unknown>) {
-      await this.setup()
+      if (connectors.destination.skipAutoSetup) {
+        // Destination manages its own setup externally (e.g. via /setup endpoint).
+        // Only set up the source here.
+        const catalog = await getCatalog()
+        await setupSource(catalog)
+      } else {
+        await this.setup()
+      }
       yield* pipe(this.read(input), this.write, persistState(stateStore))
     },
   }

docs/openapi/engine.json

@@ -526,6 +526,11 @@
           },
           "emitted_at": {
             "type": "number"
+          },
+          "row_number": {
+            "type": "integer",
+            "exclusiveMinimum": 0,
+            "maximum": 9007199254740991
           }
         },
         "required": ["type", "stream", "data", "emitted_at"]

packages/dashboard/.gitignore

@@ -0,0 +1 @@
+.next/

packages/destination-google-sheets/.gitignore

@@ -0,0 +1 @@
+scripts/.state.json

packages/destination-google-sheets/__tests__/memory-sheets.ts

@@ -76,17 +76,19 @@ export function createMemorySheets() {
         const ss = getSpreadsheet(params.spreadsheetId)
         const requests = (params.requestBody?.requests ?? []) as Record<string, unknown>[]
 
+        const replies: unknown[] = []
+
         for (const req of requests) {
           if (req.addSheet) {
             const props = (req.addSheet as { properties?: { title?: string } }).properties
             const name = props?.title ?? `Sheet${ss.sheets.size + 1}`
             if (ss.sheets.has(name)) {
               throw Object.assign(new Error(`Sheet already exists: ${name}`), { code: 400 })
             }
-            ss.sheets.set(name, { sheetId: nextSheetId++, values: [] })
-          }
-
-          if (req.updateSheetProperties) {
+            const sheetId = nextSheetId++
+            ss.sheets.set(name, { sheetId, values: [] })
+            replies.push({ addSheet: { properties: { sheetId, title: name } } })
+          } else if (req.updateSheetProperties) {
             const update = req.updateSheetProperties as {
               properties: { sheetId: number; title: string }
               fields: string
@@ -99,10 +101,13 @@ export function createMemorySheets() {
                 break
               }
             }
+            replies.push({})
+          } else {
+            replies.push({})
           }
         }
 
-        return { data: {} }
+        return { data: { replies } }
       },
 
       values: {

packages/destination-google-sheets/scripts/_state.ts

@@ -0,0 +1,95 @@
+// Shared helpers for the destination-google-sheets scripts.
+// Loads .env and manages a local .state.json that acts as a fake DB for the sheet ID.
+
+import { readFileSync, writeFileSync, unlinkSync } from 'node:fs'
+import { resolve, dirname } from 'node:path'
+import { fileURLToPath } from 'node:url'
+
+const __dirname = dirname(fileURLToPath(import.meta.url))
+const STATE_FILE = resolve(__dirname, '.state.json')
+
+// ── Env loading ──────────────────────────────────────────────────────────────
+
+export function loadEnv(): void {
+  const envPath = resolve(__dirname, '../.env')
+  try {
+    const content = readFileSync(envPath, 'utf-8')
+    for (const line of content.split('\n')) {
+      const trimmed = line.trim()
+      if (!trimmed || trimmed.startsWith('#')) continue
+      const eqIdx = trimmed.indexOf('=')
+      if (eqIdx === -1) continue
+      const key = trimmed.slice(0, eqIdx).trim()
+      const value = trimmed.slice(eqIdx + 1).trim()
+      if (!(key in process.env)) process.env[key] = value
+    }
+  } catch {
+    // .env is optional
+  }
+}
+
+// ── Sheet state ───────────────────────────────────────────────────────────────
+
+export interface SheetState {
+  spreadsheet_id: string
+  /** Per-stream cursor state, persisted across sync calls for resumable pagination. */
+  sync_state?: Record<string, unknown>
+}
+
+export function loadState(): SheetState | null {
+  try {
+    return JSON.parse(readFileSync(STATE_FILE, 'utf-8')) as SheetState
+  } catch {
+    return null
+  }
+}
+
+export function saveState(state: SheetState): void {
+  writeFileSync(STATE_FILE, JSON.stringify(state, null, 2) + '\n')
+  console.error(`Saved state → ${STATE_FILE}`)
+}
+
+export function clearState(): void {
+  try {
+    unlinkSync(STATE_FILE)
+    console.error(`Cleared state (${STATE_FILE})`)
+  } catch {
+    // already gone
+  }
+}
+
+// ── Pipeline builder ──────────────────────────────────────────────────────────
+
+export function buildDestinationConfig(spreadsheetId?: string): Record<string, unknown> {
+  return {
+    name: 'google-sheets',
+    client_id: process.env['GOOGLE_CLIENT_ID'],
+    client_secret: process.env['GOOGLE_CLIENT_SECRET'],
+    access_token: 'unused',
+    refresh_token: process.env['GOOGLE_REFRESH_TOKEN'],
+    ...(spreadsheetId ? { spreadsheet_id: spreadsheetId } : {}),
+  }
+}
+
+export const STREAMS = ['products', 'customers', 'prices', 'subscriptions'] as const
+
+export function buildPipeline(spreadsheetId?: string): Record<string, unknown> {
+  return {
+    source: { name: 'stripe', api_key: process.env['STRIPE_API_KEY'], backfill_limit: 10 },
+    destination: buildDestinationConfig(spreadsheetId),
+    streams: STREAMS.map((name) => ({ name })),
+  }
+}
+
+export function requireEnv(...keys: string[]): void {
+  const missing = keys.filter((k) => !process.env[k])
+  if (missing.length > 0) {
+    console.error(`Error: missing required env vars: ${missing.join(', ')}`)
+    process.exit(1)
+  }
+}
+
+export function getPort(): string {
+  const idx = process.argv.indexOf('--port')
+  return idx !== -1 ? process.argv[idx + 1] : '3000'
+}

packages/destination-google-sheets/scripts/check-via-server.ts

@@ -0,0 +1,29 @@
+#!/usr/bin/env node
+// GET /check — validates credentials and sheet accessibility
+// Usage: npx tsx scripts/check-via-server.ts [--port 3000]
+
+import { loadEnv, buildPipeline, requireEnv, loadState, getPort } from './_state.js'
+
+loadEnv()
+requireEnv('STRIPE_API_KEY', 'GOOGLE_CLIENT_ID', 'GOOGLE_CLIENT_SECRET', 'GOOGLE_REFRESH_TOKEN')
+
+const state = loadState()
+if (!state) {
+  console.error('No sheet state found — run setup-via-server.ts first')
+  process.exit(1)
+}
+
+const serverUrl = `http://localhost:${getPort()}`
+const pipeline = buildPipeline(state.spreadsheet_id)
+
+console.error(`Hitting ${serverUrl}/check ...`)
+console.error(`Sheet: https://docs.google.com/spreadsheets/d/${state.spreadsheet_id}`)
+
+const res = await fetch(`${serverUrl}/check`, {
+  headers: { 'X-Pipeline': JSON.stringify(pipeline) },
+})
+
+const result = await res.json()
+console.log(JSON.stringify(result, null, 2))
+
+if (res.status !== 200) process.exit(1)

packages/destination-google-sheets/scripts/setup-via-server.ts

@@ -0,0 +1,31 @@
+#!/usr/bin/env node
+// POST /setup — creates a new Google Sheet, saves its ID to .state.json
+// Usage: npx tsx scripts/setup-via-server.ts [--port 3000]
+
+import { loadEnv, buildPipeline, requireEnv, saveState, getPort } from './_state.js'
+
+loadEnv()
+requireEnv('STRIPE_API_KEY', 'GOOGLE_CLIENT_ID', 'GOOGLE_CLIENT_SECRET', 'GOOGLE_REFRESH_TOKEN')
+
+const serverUrl = `http://localhost:${getPort()}`
+
+// No spreadsheet_id — setup always creates a new sheet
+const pipeline = buildPipeline()
+
+console.error(`Hitting ${serverUrl}/setup ...`)
+
+const res = await fetch(`${serverUrl}/setup`, {
+  method: 'POST',
+  headers: { 'X-Pipeline': JSON.stringify(pipeline) },
+})
+
+if (res.status === 200) {
+  const result = (await res.json()) as { spreadsheet_id: string }
+  saveState({ spreadsheet_id: result.spreadsheet_id })
+  console.log(JSON.stringify(result, null, 2))
+} else {
+  const body = await res.text()
+  console.error(`Error: ${res.status} ${res.statusText}`)
+  if (body) console.error(body)
+  process.exit(1)
+}