feat(service): Google Sheets workflow — 3 parallel loops, generic read activity

apps/dashboard/e2e/global-setup.ts

@@ -11,7 +11,7 @@ export default async function globalSetup(_config: FullConfig) {
 
   const resolver = createConnectorResolver({
     sources: { stripe: sourceStripe },
-    destinations: { postgres: destinationPostgres, 'google-sheets': destinationGoogleSheets },
+    destinations: { postgres: destinationPostgres, 'google_sheets': destinationGoogleSheets },
   })
 
   const app = createApp(resolver)

apps/dashboard/e2e/pipeline-create.test.ts

@@ -46,7 +46,7 @@ test('full pipeline creation flow: source → discover → streams → destinati
   // Click next to destination
   await page.getByRole('button', { name: /next.*configure destination/i }).click()
 
-  // Step 3: Destination — should show postgres and google-sheets
+  // Step 3: Destination — should show postgres and google_sheets
   await expect(page.locator('select').first()).toBeVisible()
   await page.locator('select').first().selectOption('postgres')
 

apps/engine/src/api/index.ts

@@ -13,7 +13,7 @@ const port = Number(process.env.PORT || 3001)
 async function main() {
   const resolver = await createConnectorResolver({
     sources: { stripe: source },
-    destinations: { postgres: pgDestination, 'google-sheets': sheetsDestination },
+    destinations: { postgres: pgDestination, 'google_sheets': sheetsDestination },
   })
   const app = await createApp(resolver)
 

apps/engine/src/lib/default-connectors.ts

@@ -8,6 +8,6 @@ export const defaultConnectors: RegisteredConnectors = {
   sources: { stripe: sourceStripe },
   destinations: {
     postgres: destinationPostgres,
-    'google-sheets': destinationGoogleSheets,
+    'google_sheets': destinationGoogleSheets,
   },
 }

apps/service/src/__tests__/workflow.test.ts

@@ -22,12 +22,12 @@ function stubActivities(overrides: Partial<SyncActivities> = {}): SyncActivities
     discoverCatalog: async () => ({ streams: [] }),
     pipelineSetup: async () => ({}),
     pipelineSync: async () => noErrors,
-    readGoogleSheetsIntoQueue: async () => ({ count: 0, state: emptyState }),
+    readIntoQueue: async () => ({ count: 0, state: emptyState }),
     writeGoogleSheetsFromQueue: async () => ({
       errors: [],
       state: emptyState,
       written: 0,
-      rowAssignments: {},
+      rowIndexMap: {},
     }),
     pipelineTeardown: async () => {},
     updatePipelineStatus: async () => {},
@@ -310,9 +310,7 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
           if (opts?.input) return noErrors
 
           reconcileCalls++
-          return reconcileCalls === 1
-            ? { ...noErrors, eof: { reason: 'complete' } }
-            : noErrors
+          return reconcileCalls === 1 ? { ...noErrors, eof: { reason: 'complete' } } : noErrors
         },
       }),
     })
@@ -330,7 +328,9 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
       await handle.signal('desired_status', 'deleted')
       await handle.result()
 
-      expect(statusWrites).toEqual(expect.arrayContaining(['backfill', 'ready', 'paused', 'teardown']))
+      expect(statusWrites).toEqual(
+        expect.arrayContaining(['backfill', 'ready', 'paused', 'teardown'])
+      )
     })
   })
 
@@ -507,7 +507,7 @@ describe('googleSheetPipelineWorkflow (unit — stubbed activities)', () => {
           discoverCalls++
           return { streams: [] }
         },
-        readGoogleSheetsIntoQueue: async () => {
+        readIntoQueue: async () => {
           readCalls++
           return { count: 0, state: emptyState }
         },
@@ -563,7 +563,7 @@ describe('googleSheetPipelineWorkflow (unit — stubbed activities)', () => {
       workflowsPath,
       activities: stubActivities({
         discoverCatalog: async () => discoveredCatalog,
-        readGoogleSheetsIntoQueue: async () => {
+        readIntoQueue: async () => {
           readCalls++
           return readCalls === 1
             ? { count: 1, state: { streams: { customers: { cursor: 'cus_1' } }, global: {} } }
@@ -575,7 +575,7 @@ describe('googleSheetPipelineWorkflow (unit — stubbed activities)', () => {
             errors: [],
             state: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
             written: 0,
-            rowAssignments: {},
+            rowIndexMap: {},
           }
         },
       }),

apps/service/src/api/app.test.ts

@@ -20,7 +20,7 @@ let resolver: ConnectorResolver
 beforeAll(async () => {
   resolver = await createConnectorResolver({
     sources: { test: sourceTest },
-    destinations: { test: destinationTest, 'google-sheets': destinationGoogleSheets },
+    destinations: { test: destinationTest, 'google_sheets': destinationGoogleSheets },
   })
 })
 
@@ -80,7 +80,7 @@ describe('GET /health', () => {
 })
 
 describe('POST /pipelines workflow dispatch', () => {
-  it('starts google-sheets pipelines on the dedicated workflow', async () => {
+  it('starts google_sheets pipelines on the dedicated workflow', async () => {
     const start = vi.fn(async () => ({}))
     const res = await createApp({
       temporal: { client: { start } as unknown as WorkflowClient, taskQueue: 'unused' },
@@ -92,8 +92,8 @@ describe('POST /pipelines workflow dispatch', () => {
       body: JSON.stringify({
         source: { type: 'test', test: {} },
         destination: {
-          type: 'google-sheets',
-          'google-sheets': {
+          type: 'google_sheets',
+          'google_sheets': {
             spreadsheet_id: 'sheet_123',
             spreadsheet_title: 'Test Sheet',
             client_id: 'client',
@@ -133,12 +133,12 @@ function stubActivities(): SyncActivities {
     discoverCatalog: async () => ({ streams: [] }),
     pipelineSetup: async () => ({}),
     pipelineSync: async () => noErrors,
-    readGoogleSheetsIntoQueue: async () => ({ count: 0, state: emptyState }),
+    readIntoQueue: async () => ({ count: 0, state: emptyState }),
     writeGoogleSheetsFromQueue: async () => ({
       errors: [],
       state: emptyState,
       written: 0,
-      rowAssignments: {},
+      rowIndexMap: {},
     }),
     pipelineTeardown: async () => {},
     updatePipelineStatus: async () => {},
@@ -240,7 +240,7 @@ describe('pipeline CRUD', () => {
     await a.request(`/pipelines/${created.id}`, { method: 'DELETE' })
   })
 
-  it('rejects changing the target spreadsheet for a google-sheets pipeline', async () => {
+  it('rejects changing the target spreadsheet for a google_sheets pipeline', async () => {
     const a = liveApp()
 
     const createRes = await a.request('/pipelines', {
@@ -249,8 +249,8 @@ describe('pipeline CRUD', () => {
       body: JSON.stringify({
         source: { type: 'test', test: {} },
         destination: {
-          type: 'google-sheets',
-          'google-sheets': {
+          type: 'google_sheets',
+          'google_sheets': {
             spreadsheet_id: 'sheet_123',
             spreadsheet_title: 'Original Sheet',
             client_id: 'client',
@@ -269,8 +269,8 @@ describe('pipeline CRUD', () => {
       headers: { 'content-type': 'application/json' },
       body: JSON.stringify({
         destination: {
-          type: 'google-sheets',
-          'google-sheets': {
+          type: 'google_sheets',
+          'google_sheets': {
             spreadsheet_id: 'sheet_456',
             spreadsheet_title: 'Replacement Sheet',
             client_id: 'client',
@@ -285,7 +285,7 @@ describe('pipeline CRUD', () => {
     expect(updateRes.status).toBe(400)
     expect(await updateRes.json()).toEqual({
       error:
-        'Changing the target spreadsheet for a google-sheets pipeline requires recreating the pipeline',
+        'Changing the target spreadsheet for a google_sheets pipeline requires recreating the pipeline',
     })
 
     await a.request(`/pipelines/${created.id}`, { method: 'DELETE' })
@@ -300,8 +300,8 @@ describe('pipeline CRUD', () => {
       body: JSON.stringify({
         source: { type: 'test', test: {} },
         destination: {
-          type: 'google-sheets',
-          'google-sheets': {
+          type: 'google_sheets',
+          'google_sheets': {
             spreadsheet_id: 'sheet_123',
             spreadsheet_title: 'Original Sheet',
             client_id: 'client',
@@ -320,8 +320,8 @@ describe('pipeline CRUD', () => {
       headers: { 'content-type': 'application/json' },
       body: JSON.stringify({
         destination: {
-          type: 'google-sheets',
-          'google-sheets': {
+          type: 'google_sheets',
+          'google_sheets': {
             spreadsheet_id: 'sheet_123',
             spreadsheet_title: 'Renamed Sheet',
             client_id: 'client',
@@ -335,8 +335,8 @@ describe('pipeline CRUD', () => {
 
     expect(updateRes.status).toBe(200)
     const updated = await updateRes.json()
-    expect(updated.destination['google-sheets'].spreadsheet_id).toBe('sheet_123')
-    expect(updated.destination['google-sheets'].spreadsheet_title).toBe('Renamed Sheet')
+    expect(updated.destination['google_sheets'].spreadsheet_id).toBe('sheet_123')
+    expect(updated.destination['google_sheets'].spreadsheet_title).toBe('Renamed Sheet')
 
     await a.request(`/pipelines/${created.id}`, { method: 'DELETE' })
   })

apps/service/src/api/app.ts

@@ -13,7 +13,7 @@ const DEFAULT_PIPELINE_WORKFLOW = 'pipelineWorkflow'
 const GOOGLE_SHEETS_PIPELINE_WORKFLOW = 'googleSheetPipelineWorkflow'
 
 function workflowTypeForPipeline(pipeline: Pipeline): string {
-  return pipeline.destination.type === 'google-sheets'
+  return pipeline.destination.type === 'google_sheets'
     ? GOOGLE_SHEETS_PIPELINE_WORKFLOW
     : DEFAULT_PIPELINE_WORKFLOW
 }
@@ -232,7 +232,7 @@ export function createApp(options: AppOptions) {
         }
       }
 
-      // Validate google-sheets constraints
+      // Validate google_sheets constraints
       const next = {
         ...current,
         ...(patch.source ? { source: patch.source } : {}),
@@ -243,22 +243,22 @@ export function createApp(options: AppOptions) {
         return c.json(
           {
             error:
-              'Changing destination.type between google-sheets and non-google-sheets requires recreating the pipeline',
+              'Changing destination.type between google_sheets and non-google_sheets requires recreating the pipeline',
           },
           400
         )
       }
       if (
-        current.destination.type === 'google-sheets' &&
+        current.destination.type === 'google_sheets' &&
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        (current.destination as any)['google-sheets']?.spreadsheet_id !==
+        (current.destination as any)['google_sheets']?.spreadsheet_id !==
           // eslint-disable-next-line @typescript-eslint/no-explicit-any
-          (next.destination as any)['google-sheets']?.spreadsheet_id
+          (next.destination as any)['google_sheets']?.spreadsheet_id
       ) {
         return c.json(
           {
             error:
-              'Changing the target spreadsheet for a google-sheets pipeline requires recreating the pipeline',
+              'Changing the target spreadsheet for a google_sheets pipeline requires recreating the pipeline',
           },
           400
         )

apps/service/src/cli.ts

@@ -9,11 +9,14 @@ import destinationGoogleSheets from '@stripe/sync-destination-google-sheets'
 import { createApp } from './api/app.js'
 import { filePipelineStore } from './lib/stores-fs.js'
 import type { WorkflowClient } from '@temporalio/client'
+import { homedir } from 'node:os'
 import { logger } from './logger.js'
 
+const defaultDataDir = process.env.DATA_DIR ?? `${homedir()}/.stripe-sync`
+
 const resolverPromise = createConnectorResolver({
   sources: { stripe: sourceStripe },
-  destinations: { postgres: destinationPostgres, 'google-sheets': destinationGoogleSheets },
+  destinations: { postgres: destinationPostgres, 'google_sheets': destinationGoogleSheets },
 })
 
 async function createTemporalClient(
@@ -47,8 +50,8 @@ const serveCmd = defineCommand({
     },
     'data-dir': {
       type: 'string',
-      required: true,
-      description: 'Directory to persist pipeline configs as JSON files.',
+      default: defaultDataDir,
+      description: `Directory to persist pipeline configs as JSON files (default: ${defaultDataDir}).`,
     },
   },
   async run({ args }) {
@@ -108,8 +111,8 @@ const workerCmd = defineCommand({
     },
     'data-dir': {
       type: 'string',
-      required: true,
-      description: 'Directory to persist pipeline configs as JSON files.',
+      default: defaultDataDir,
+      description: `Directory to persist pipeline configs as JSON files (default: ${defaultDataDir}).`,
     },
   },
   async run({ args }) {
@@ -170,8 +173,8 @@ const webhookCmd = defineCommand({
     },
     'data-dir': {
       type: 'string',
-      required: true,
-      description: 'Directory to persist pipeline configs as JSON files.',
+      default: defaultDataDir,
+      description: `Directory to persist pipeline configs as JSON files (default: ${defaultDataDir}).`,
     },
   },
   async run({ args }) {