feat(destination): Google Sheets destination connector + row-index workflow

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/package.json

@@ -27,6 +27,7 @@
     "zod": "^4.3.6"
   },
   "devDependencies": {
+    "@types/node": "^25.5.0",
     "vitest": "^3.2.4"
   }
 }

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)
+}

packages/destination-google-sheets/scripts/sheet-size.ts

@@ -0,0 +1,67 @@
+#!/usr/bin/env node
+// Calculates total cell count across all sheets in the saved spreadsheet.
+//
+// Usage: npx tsx scripts/sheet-size.ts
+
+import { readFileSync } from 'node:fs'
+import { resolve, dirname } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { google } from 'googleapis'
+
+const __dirname = dirname(fileURLToPath(import.meta.url))
+
+// Load .env
+const envPath = resolve(__dirname, '../.env')
+try {
+  for (const line of readFileSync(envPath, 'utf-8').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 */
+}
+
+// Load spreadsheet ID from .state.json
+const stateFile = resolve(__dirname, '.state.json')
+let spreadsheetId: string
+try {
+  const state = JSON.parse(readFileSync(stateFile, 'utf-8')) as { spreadsheet_id: string }
+  spreadsheetId = state.spreadsheet_id
+} catch {
+  console.error('No .state.json found — run setup-via-server.ts first')
+  process.exit(1)
+}
+
+const auth = new google.auth.OAuth2(
+  process.env['GOOGLE_CLIENT_ID'],
+  process.env['GOOGLE_CLIENT_SECRET']
+)
+auth.setCredentials({ refresh_token: process.env['GOOGLE_REFRESH_TOKEN'] })
+const sheets = google.sheets({ version: 'v4', auth })
+
+// Fetch spreadsheet metadata (includes all sheet grid properties)
+const res = await sheets.spreadsheets.get({
+  spreadsheetId,
+  fields: 'sheets(properties(title,gridProperties))',
+})
+
+console.error(`Sheet: https://docs.google.com/spreadsheets/d/${spreadsheetId}\n`)
+
+let grandTotal = 0
+for (const sheet of res.data.sheets ?? []) {
+  const title = sheet.properties?.title ?? '(untitled)'
+  const rowCount = sheet.properties?.gridProperties?.rowCount ?? 0
+  const columnCount = sheet.properties?.gridProperties?.columnCount ?? 0
+  const cells = rowCount * columnCount
+  grandTotal += cells
+  console.error(
+    `  ${title}: ${rowCount} rows × ${columnCount} cols = ${cells.toLocaleString()} cells`
+  )
+}
+
+console.error(`\n  Total: ${grandTotal.toLocaleString()} cells`)

packages/destination-google-sheets/scripts/stripe-to-google-sheets.ts

@@ -0,0 +1,89 @@
+#!/usr/bin/env node
+// Sync Stripe → Google Sheets via the sync-engine CLI.
+// Reads credentials from packages/destination-google-sheets/.env
+//
+// Usage: npx tsx scripts/stripe-to-google-sheets.ts
+//   or:  node --import tsx scripts/stripe-to-google-sheets.ts
+
+import { readFileSync } from 'node:fs'
+import { resolve, dirname } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { execFileSync, spawnSync } from 'node:child_process'
+
+const __dirname = dirname(fileURLToPath(import.meta.url))
+
+// Load .env from the package root
+const envPath = resolve(__dirname, '../.env')
+try {
+  const envContent = readFileSync(envPath, 'utf-8')
+  for (const line of envContent.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; env vars may already be set
+}
+
+const {
+  STRIPE_API_KEY,
+  GOOGLE_CLIENT_ID,
+  GOOGLE_CLIENT_SECRET,
+  GOOGLE_REFRESH_TOKEN,
+  GOOGLE_SPREADSHEET_ID,
+} = process.env
+
+if (!STRIPE_API_KEY) {
+  console.error('Error: STRIPE_API_KEY is required (set it in .env or the environment)')
+  process.exit(1)
+}
+
+// Fetch Stripe account ID
+const accountRes = await fetch('https://api.stripe.com/v1/account', {
+  headers: {
+    Authorization: `Basic ${Buffer.from(`${STRIPE_API_KEY}:`).toString('base64')}`,
+  },
+})
+const account = (await accountRes.json()) as { id: string }
+console.error(`Stripe: ${account.id}`)
+console.error(`Sheet: https://docs.google.com/spreadsheets/d/${GOOGLE_SPREADSHEET_ID}`)
+
+const pipeline = JSON.stringify({
+  source: { name: 'stripe', api_key: STRIPE_API_KEY, backfill_limit: 10 },
+  destination: {
+    name: 'google-sheets',
+    client_id: GOOGLE_CLIENT_ID,
+    client_secret: GOOGLE_CLIENT_SECRET,
+    access_token: 'unused',
+    refresh_token: GOOGLE_REFRESH_TOKEN,
+    spreadsheet_id: GOOGLE_SPREADSHEET_ID,
+  },
+  streams: [{ name: 'products' }, { name: 'customers' }],
+})
+
+const repoRoot = resolve(__dirname, '../../..')
+const cliPath = resolve(repoRoot, 'apps/engine/src/cli/index.ts')
+
+// Use bun if available, else tsx
+function hasBun(): boolean {
+  try {
+    execFileSync('bun', ['--version'], { stdio: 'ignore' })
+    return true
+  } catch {
+    return false
+  }
+}
+
+const tsxBin = resolve(repoRoot, 'node_modules/.bin/tsx')
+const [cmd, ...cmdArgs] = hasBun() ? ['bun', cliPath] : [tsxBin, cliPath]
+
+const result = spawnSync(cmd, [...cmdArgs, 'sync', '--xPipeline', pipeline], {
+  stdio: 'inherit',
+  cwd: repoRoot,
+})
+
+process.exit(result.status ?? 1)