stripe
diff --git a/‎.github/workflows/ci.yml‎
Lines changed: 36 additions & 0 deletions b/‎.github/workflows/ci.yml‎
Lines changed: 36 additions & 0 deletions
diff --git a/‎Dockerfile‎
Lines changed: 2 additions & 2 deletions b/‎Dockerfile‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎apps/engine/src/cli/backfill.ts‎
Lines changed: 114 additions & 0 deletions b/‎apps/engine/src/cli/backfill.ts‎
Lines changed: 114 additions & 0 deletions
diff --git a/‎apps/engine/src/cli/command.ts‎
Lines changed: 2 additions & 0 deletions b/‎apps/engine/src/cli/command.ts‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎apps/engine/src/cli/index.ts‎
Lines changed: 3 additions & 0 deletions b/‎apps/engine/src/cli/index.ts‎
Lines changed: 3 additions & 0 deletions
diff --git a/‎apps/engine/src/lib/backfill.test.ts‎
Lines changed: 94 additions & 0 deletions b/‎apps/engine/src/lib/backfill.test.ts‎
Lines changed: 94 additions & 0 deletions
diff --git a/‎apps/engine/src/lib/backfill.ts‎
Lines changed: 72 additions & 0 deletions b/‎apps/engine/src/lib/backfill.ts‎
Lines changed: 72 additions & 0 deletions
diff --git a/‎apps/service/src/bin/sync-service.ts‎
Lines changed: 3 additions & 0 deletions b/‎apps/service/src/bin/sync-service.ts‎
Lines changed: 3 additions & 0 deletions
@@ -158,6 +158,42 @@ jobs:
         if: always()
         run: '[ -f /tmp/vitest-skip-warnings.txt ] && cat /tmp/vitest-skip-warnings.txt || true'
 
+  # ---------------------------------------------------------------------------
+  # mitmweb — proxy intercept smoke test (curl, Node, Bun → httpbin.org)
+  # ---------------------------------------------------------------------------
+  mitmweb:
+    name: mitmweb proxy intercept test
+    runs-on: ubuntu-24.04-arm
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Install pnpm
+        uses: pnpm/action-setup@v5
+
+      - name: Set up Node
+        uses: actions/setup-node@v6
+        with:
+          node-version-file: ./.nvmrc
+
+      - name: Install Bun
+        uses: oven-sh/setup-bun@v2
+        with:
+          bun-version: latest
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Install mitmproxy
+        run: pip3 install --quiet mitmproxy
+
+      - name: Run mitmweb intercept test
+        run: bash scripts/mitmweb-env.test.sh
+
+      - name: Stop mitmweb
+        if: always()
+        run: pkill -f mitmweb || true
+
   # ---------------------------------------------------------------------------
   # Publish npm — publish @stripe/* packages to GitHub Packages
   # ---------------------------------------------------------------------------
 
@@ -49,7 +49,7 @@ ENV NODE_ENV=production
 ENV GIT_COMMIT=$GIT_COMMIT
 ENV BUILD_DATE=$BUILD_DATE
 ENV COMMIT_URL=$COMMIT_URL
-ENTRYPOINT ["node", "dist/cli/index.js"]
+ENTRYPOINT ["node", "--use-env-proxy", "dist/cli/index.js"]
 CMD ["serve"]
 
 # ===========================================================================
@@ -88,5 +88,5 @@ ENV NODE_ENV=production
 ENV GIT_COMMIT=$GIT_COMMIT
 ENV BUILD_DATE=$BUILD_DATE
 ENV COMMIT_URL=$COMMIT_URL
-ENTRYPOINT ["node", "dist/bin/sync-service.js"]
+ENTRYPOINT ["node", "--use-env-proxy", "dist/bin/sync-service.js"]
 CMD ["serve", "--temporal-address", "temporal:7233", "--temporal-task-queue", "sync-engine"]
@@ -0,0 +1,114 @@
+import { access, readFile, writeFile } from 'node:fs/promises'
+import { defineCommand } from 'citty'
+import { parseJsonOrFile } from '@stripe/sync-ts-cli'
+import {
+  type PipelineConfig,
+  PipelineConfig as PipelineConfigSchema,
+  coerceSyncState,
+  createRemoteEngine,
+} from '../index.js'
+import { pipelineSyncUntilComplete } from '../lib/backfill.js'
+
+async function readState(path?: string) {
+  if (!path) return undefined
+  try {
+    await access(path)
+  } catch {
+    return undefined
+  }
+  return coerceSyncState(JSON.parse(await readFile(path, 'utf8')))
+}
+
+async function writeState(path: string | undefined, state: unknown) {
+  if (!path) return
+  await writeFile(path, `${JSON.stringify(state, null, 2)}\n`, 'utf8')
+}
+
+export const backfillCmd = defineCommand({
+  meta: {
+    name: 'backfill',
+    description: 'Call a remote sync engine until pipeline_sync reaches eof:complete',
+  },
+  args: {
+    syncEngineUrl: {
+      type: 'string',
+      description: 'Remote sync engine base URL (or SYNC_ENGINE_URL env)',
+    },
+    pipeline: {
+      type: 'string',
+      description: 'Pipeline config as inline JSON or a JSON file path',
+    },
+    statePath: {
+      type: 'string',
+      description: 'Optional JSON file to load/save SyncState between attempts',
+    },
+    stateLimit: {
+      type: 'string',
+      default: '100',
+      description: 'Per-call state_limit passed to pipeline_sync (default: 100)',
+    },
+    timeLimit: {
+      type: 'string',
+      default: '10',
+      description: 'Per-call time_limit in seconds passed to pipeline_sync (default: 10)',
+    },
+  },
+  async run({ args }) {
+    const syncEngineUrl = args.syncEngineUrl || process.env.SYNC_ENGINE_URL
+    if (!syncEngineUrl) throw new Error('Missing --sync-engine-url or SYNC_ENGINE_URL env')
+    if (!args.pipeline) throw new Error('Missing --pipeline')
+
+    const pipeline = PipelineConfigSchema.parse(parseJsonOrFile(args.pipeline)) as PipelineConfig
+    const state = await readState(args.statePath)
+    const stateLimit = parseInt(args.stateLimit, 10)
+    const timeLimit = parseInt(args.timeLimit, 10)
+
+    if (!Number.isInteger(stateLimit) || stateLimit <= 0) {
+      throw new Error('--state-limit must be a positive integer')
+    }
+    if (!Number.isInteger(timeLimit) || timeLimit <= 0) {
+      throw new Error('--time-limit must be a positive integer')
+    }
+
+    const engine = createRemoteEngine(syncEngineUrl)
+    const result = await pipelineSyncUntilComplete(engine, pipeline, {
+      state,
+      state_limit: stateLimit,
+      time_limit: timeLimit,
+      onAttempt: (attempt, currentState) => {
+        console.error(
+          JSON.stringify({
+            event: 'pipeline_sync_attempt_started',
+            attempt,
+            state_provided: currentState != null,
+          })
+        )
+      },
+      onMessage: (message, attempt) => {
+        process.stdout.write(`${JSON.stringify(message)}\n`)
+        if (message.type === 'eof') {
+          console.error(
+            JSON.stringify({
+              event: 'pipeline_sync_attempt_finished',
+              attempt,
+              eof_reason: message.eof.reason,
+            })
+          )
+        }
+      },
+      onState: async (nextState) => {
+        await writeState(args.statePath, nextState)
+      },
+    })
+
+    await writeState(args.statePath, result.state)
+    console.error(
+      JSON.stringify({
+        event: 'pipeline_sync_complete',
+        attempts: result.attempts,
+        eof_reason: result.eof.reason,
+        state_path: args.statePath ?? null,
+      })
+    )
+  },
+})
@@ -6,6 +6,7 @@ import { parseJsonOrFile } from '@stripe/sync-ts-cli'
 import { createConnectorResolver, createEngine } from '../lib/index.js'
 import { createApp } from '../api/app.js'
 import { serveAction } from '../serve-command.js'
+import { backfillCmd } from './backfill.js'
 import { supabaseCmd } from './supabase.js'
 import { createSyncCmd } from './sync.js'
 import { defaultConnectors } from '../lib/default-connectors.js'
@@ -100,6 +101,7 @@ export async function createProgram() {
   return defineCommand({
     ...specCli,
     subCommands: {
+      backfill: backfillCmd,
       serve: serveCmd,
       supabase: supabaseCmd,
       sync: createSyncCmd(engine, resolver),
 
@@ -1,6 +1,9 @@
 #!/usr/bin/env node
+import { assertUseEnvProxy } from '@stripe/sync-ts-cli/env-proxy'
 import { runMain } from 'citty'
 import { createProgram } from './command.js'
 
+assertUseEnvProxy()
+
 const program = await createProgram()
 runMain(program)
@@ -0,0 +1,94 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { Engine } from './engine.js'
+import type { PipelineConfig, SyncOutput } from '@stripe/sync-protocol'
+import { pipelineSyncUntilComplete } from './backfill.js'
+
+const pipeline: PipelineConfig = {
+  source: { type: 'test', test: {} },
+  destination: { type: 'test', test: {} },
+}
+
+async function* toAsync<T>(items: T[]): AsyncIterable<T> {
+  for (const item of items) yield item
+}
+
+describe('pipelineSyncUntilComplete', () => {
+  it('retries until eof complete and carries forward state', async () => {
+    const calls: Array<unknown> = []
+    const onState = vi.fn()
+    const engine: Engine = {
+      meta_sources_list: vi.fn(),
+      meta_sources_get: vi.fn(),
+      meta_destinations_list: vi.fn(),
+      meta_destinations_get: vi.fn(),
+      source_discover: vi.fn(),
+      pipeline_check: vi.fn(),
+      pipeline_setup: vi.fn(),
+      pipeline_teardown: vi.fn(),
+      pipeline_read: vi.fn(),
+      pipeline_write: vi.fn(),
+      pipeline_sync: vi.fn((_pipeline, opts) => {
+        calls.push(opts?.state)
+        const outputs: SyncOutput[] =
+          calls.length === 1
+            ? [
+                {
+                  type: 'source_state',
+                  source_state: { stream: 'customers', data: { cursor: 'cus_1' } },
+                },
+                {
+                  type: 'eof',
+                  eof: {
+                    reason: 'state_limit',
+                    state: {
+                      source: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
+                      destination: { streams: {}, global: {} },
+                      engine: { streams: {}, global: {} },
+                    },
+                  },
+                },
+              ]
+            : [{ type: 'eof', eof: { reason: 'complete' } }]
+        return toAsync(outputs)
+      }),
+    } as unknown as Engine
+
+    const result = await pipelineSyncUntilComplete(engine, pipeline, { state_limit: 1, onState })
+
+    expect(calls).toEqual([
+      undefined,
+      {
+        source: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
+        destination: { streams: {}, global: {} },
+        engine: { streams: {}, global: {} },
+      },
+    ])
+    expect(result.attempts).toBe(2)
+    expect(result.eof.reason).toBe('complete')
+    expect(onState).toHaveBeenLastCalledWith({
+      source: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
+      destination: { streams: {}, global: {} },
+      engine: { streams: {}, global: {} },
+    }, 1)
+  })
+
+  it('throws when pipeline_sync ends with an unexpected eof reason', async () => {
+    const engine: Engine = {
+      meta_sources_list: vi.fn(),
+      meta_sources_get: vi.fn(),
+      meta_destinations_list: vi.fn(),
+      meta_destinations_get: vi.fn(),
+      source_discover: vi.fn(),
+      pipeline_check: vi.fn(),
+      pipeline_setup: vi.fn(),
+      pipeline_teardown: vi.fn(),
+      pipeline_read: vi.fn(),
+      pipeline_write: vi.fn(),
+      pipeline_sync: vi.fn(() => toAsync<SyncOutput>([{ type: 'eof', eof: { reason: 'aborted' } }])),
+    } as unknown as Engine
+
+    await expect(pipelineSyncUntilComplete(engine, pipeline)).rejects.toThrow(
+      /unexpected eof reason: aborted/
+    )
+  })
+})
@@ -0,0 +1,72 @@
+import type { EofPayload, PipelineConfig, SourceStateMessage } from '@stripe/sync-protocol'
+import { emptySyncState, type SyncOutput, type SyncState } from '@stripe/sync-protocol'
+import type { Engine, SourceReadOptions } from './engine.js'
+
+export interface PipelineSyncUntilCompleteOptions
+  extends Omit<SourceReadOptions, 'state'> {
+  state?: SyncState
+  onAttempt?: (attempt: number, state: SyncState | undefined) => void | Promise<void>
+  onMessage?: (message: SyncOutput, attempt: number) => void | Promise<void>
+  onState?: (state: SyncState, attempt: number) => void | Promise<void>
+}
+
+export interface PipelineSyncUntilCompleteResult {
+  attempts: number
+  state: SyncState
+  eof: EofPayload
+}
+
+function mergeStateMessage(state: SyncState | undefined, msg: SourceStateMessage): SyncState {
+  const next = structuredClone(state ?? emptySyncState())
+  if (msg.source_state.state_type === 'global') {
+    next.source.global = msg.source_state.data as Record<string, unknown>
+    return next
+  }
+  next.source.streams[msg.source_state.stream] = msg.source_state.data
+  return next
+}
+
+export async function pipelineSyncUntilComplete(
+  engine: Engine,
+  pipeline: PipelineConfig,
+  opts: PipelineSyncUntilCompleteOptions = {}
+): Promise<PipelineSyncUntilCompleteResult> {
+  const { state: initialState, onAttempt, onMessage, onState, ...readOpts } = opts
+  let state = initialState
+  let attempts = 0
+
+  while (true) {
+    attempts += 1
+    await onAttempt?.(attempts, state)
+
+    let eof: EofPayload | undefined
+    for await (const message of engine.pipeline_sync(pipeline, { ...readOpts, state })) {
+      await onMessage?.(message, attempts)
+
+      if (message.type === 'source_state') {
+        state = mergeStateMessage(state, message)
+        await onState?.(state, attempts)
+      }
+
+      if (message.type === 'eof') {
+        eof = message.eof
+        if (message.eof.state) {
+          state = message.eof.state
+          await onState?.(state, attempts)
+        }
+      }
+    }
+
+    if (!eof) {
+      throw new Error(`pipeline_sync attempt ${attempts} ended without eof`)
+    }
+
+    if (eof.reason === 'complete') {
+      return { attempts, state: state ?? emptySyncState(), eof }
+    }
+
+    if (eof.reason !== 'state_limit' && eof.reason !== 'time_limit') {
+      throw new Error(`pipeline_sync attempt ${attempts} ended with unexpected eof reason: ${eof.reason}`)
+    }
+  }
+}
@@ -1,7 +1,10 @@
 #!/usr/bin/env node
 import 'dotenv/config'
+import { assertUseEnvProxy } from '@stripe/sync-ts-cli/env-proxy'
 import { runMain } from 'citty'
 import { createProgram } from '../cli.js'
 
+assertUseEnvProxy()
+
 const program = await createProgram()
 runMain(program)