### NPM Installation and Start Command

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-06-self-contained-local-bundle.md

Instructions for installing the sync engine package globally via npm and starting the services.

```sh
npm install -g @stripe/sync-engine
stripe-sync start
```

--------------------------------

### Example Output: Pipeline Setup Log and Control Messages

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Example JSON output for pipeline setup, showing log and control messages.

```json
{"type":"log","log":{"level":"info","message":"Starting pipeline setup","data":{"source_type":"stripe","destination_type":"postgres","run_source":true,"run_destination":true}}}
{"type":"control","control":{"control_type":"source_config","source_config":{"api_key":"sk_test_...","api_version":"2025-04-30.basil","account_id":"acct_test_123","account_created":1700000000}}}
```

--------------------------------

### Local Development Setup

Source: https://github.com/stripe/sync-engine/blob/dev/docs/service/temporal.md

Commands to start all necessary services for local development: Temporal dev server, sync engine, sync service, webhook server, and worker.

```sh
# Terminal 1: Temporal dev server
temporal server start-dev

# Terminal 2: Sync engine (stateless execution)
sync-engine serve --port 4010

# Terminal 3: Sync service (config CRUD + config resolution)
sync-service serve --port 4020 --temporal-address localhost:7233

# Terminal 4: Webhook server (public ingress)
sync-service webhook --port 4030 --temporal-address localhost:7233

# Terminal 5: Worker
sync-service worker --temporal-address localhost:7233
```

--------------------------------

### Optional chaining for setup() call

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-03-18-plan-004-source-setup-teardown.md

Demonstrates how to safely call the optional `setup()` method on a source object using optional chaining (`?.`). This ensures that the code does not throw an error if the `setup()` method is not implemented on the source.

```typescript
await source.setup?.({ config, catalog })
```

--------------------------------

### Start Message Example

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-start-end-message.md

An example of the 'start' message as the first line of the NDJSON request body. It includes pipeline configuration, initial state, and time limits.

```json
{
  "type": "start",
  "start": {
    "pipeline": {
      "source": { "type": "stripe", "api_key": "sk_test_...", "api_version": "2024-04-10" },
      "destination": { "type": "postgres", "connection_string": "..." },
      "streams": [{ "name": "customer", "sync_mode": "incremental" }]
    },
    "state": null,
    "time_limit": 30
  }
}
```

--------------------------------

### Clone and Install Dependencies

Source: https://github.com/stripe/sync-engine/blob/dev/demo/README.md

Clone the sync-engine repository and install Node.js and pnpm. Ensure you have nvm installed for Node version management.

```sh
git clone git@github.com:stripe/sync-engine.git
cd sync-engine

# nvm (Node version manager) — skip if you already have it
# see https://github.com/nvm-sh/nvm?tab=readme-ov-file#installing-and-updating
curl -o- https://raw.githubusercontent.com/nvm/nvm/v0.40.3/install.sh | bash

# Node 24+
nvm install 24
nvm use 24

# pnpm (auto-provided via corepack)
corepack enable
pnpm install
```

--------------------------------

### CLI Usage Examples

Source: https://github.com/stripe/sync-engine/blob/dev/packages/protocol/CONNECTORS.md

Provides examples of how to use the sync-engine connectors via the command-line interface (CLI).

```APIDOC
## CLI Usage

The `ts-cli` bridge maps `--key value` flags to the named parameters object for connector operations.

**Examples:**

- **`spec` (no parameters):**
  ```sh
  source-stripe spec
  ```

- **`check` / `discover` (requires `--config`):**
  ```sh
  source-stripe check   --config '{"api_key":"sk_test..."}'
  source-stripe discover --config '{"api_key":"sk_test..."}'
  ```

- **`read` (requires `--config` and `--catalog`, optional `--state`):**
  ```sh
  source-stripe read \
    --config '{"api_key":"sk_test...","base_url":"http://localhost:12111"}' \
    --catalog '{"streams":[{"stream":{"name":"customer","primary_key":[["id"]]},"sync_mode":"full_refresh","destination_sync_mode":"append"}]}'
  ```

- **`write` (requires `--config` and `--catalog`, messages piped from stdin):**
  ```sh
  source-stripe read --config '...' --catalog '...' \
    | dest-postgres write --config '{"url":"postgres://..."}' --catalog '...' 
  ```
```

--------------------------------

### Install, Build, Format, Lint, and Test Packages

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/active/2026-03-20-plan-008-scope-rename-stripe-sync.md

Commands to install dependencies, build the project, format code, run linters, and execute tests across all packages.

```bash
pnpm install
pnpm build
pnpm format
pnpm lint
pnpm -r test
```

--------------------------------

### Activity Setup Function Signature

Source: https://github.com/stripe/sync-engine/blob/dev/docs/service/temporal.md

Defines the signature for the `setup` activity, which initiates the sync process on the engine.

```typescript
setup(syncId)
```

--------------------------------

### Example Output: Source Discover Catalog

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Example JSON output when a source's catalog is discovered.

```json
{"type":"catalog","catalog":{"streams":[{"name":"customer","primary_key":[["id"]],"newer_than_field":"_updated_at"}]}}
```

--------------------------------

### Install and Build Sync Engine

Source: https://github.com/stripe/sync-engine/blob/dev/CLAUDE.md

Commands to install dependencies and build the project. Building is required before running the CLI or end-to-end tests.

```sh
pnpm install
pnpm build          # required before running CLI or e2e tests
```

--------------------------------

### HTTP API POST /pipeline_sync Example

Source: https://context7.com/stripe/sync-engine/llms.txt

This example demonstrates how to initiate a full read-write pipeline sync using the HTTP API. It shows the `curl` command to send a pipeline configuration and stream NDJSON output, filtering for non-record messages using `jq`.

```sh
# Start the engine server
npx @stripe/sync-engine serve --port 3000

# Run a sync — streams NDJSON to stdout
curl -s -X POST http://localhost:3000/pipeline_sync \
  -H 'Content-Type: application/json' \
  -d '{
    "pipeline": {
      "source": { "type": "stripe", "stripe": { "api_key": "sk_test_...", "api_version": "2024-06-20" } },
      "destination": { "type": "postgres", "postgres": { "url": "postgresql://localhost/mydb", "schema": "stripe" } },
      "streams": [
        { "name": "customers", "sync_mode": "incremental" }
      ]
    },
    "time_limit": 300,
    "run_id": "run-001"
  }' | while IFS= read -r line; do echo "$line" | jq -r 'select(.type != "record") | [.type, ._emitted_by] | @tsv'; done

# Expected output lines (non-record messages):
# stream_status  source/stripe
# progress       engine
# source_state   destination/postgres
# eof            engine
```

--------------------------------

### Add setup() and teardown() to Source interface

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-03-18-plan-004-source-setup-teardown.md

Adds optional `setup()` and `teardown()` methods to the `Source` interface. `setup()` provisions resources before `read()` and is idempotent. `teardown()` releases resources after `read()` completes or on sync deletion. Connectors that do not require these operations can omit them.

```typescript
export interface Source<TConfig extends Record<string, unknown> = Record<string, unknown>> {
  spec(): ConnectorSpecification
  check(params: { config: TConfig }): Promise<CheckResult>
  discover(params: { config: TConfig }): Promise<CatalogMessage>
  read(params: {
    config: TConfig
    catalog: ConfiguredCatalog
    state?: StateMessage[]
  }): AsyncIterableIterator<Message>

  /**
   * Provision external resources needed before read() can produce data.
   * Called once when a sync starts or resumes. Idempotent — safe to call
   * multiple times (e.g. after a crash recovery).
   *
   * Examples: register a webhook endpoint, create a replication slot,
   * subscribe to a topic.
   *
   * Sources that don't need setup can omit this method.
   */
  setup?(params: { config: TConfig; catalog: ConfiguredCatalog }): Promise<void>

  /**
   * Release external resources provisioned by setup().
   * Called when a sync is deleted or permanently stopped.
   *
   * Examples: delete a webhook endpoint, drop a replication slot,
   * remove a consumer group.
   *
   * Sources that don't need teardown can omit this method.
   */
  teardown?(params: { config: TConfig }): Promise<void>
}
```

--------------------------------

### Install ts-cli

Source: https://github.com/stripe/sync-engine/blob/dev/packages/ts-cli/README.md

Install the ts-cli package using pnpm.

```sh
pnpm add @stripe/sync-ts-cli
```

--------------------------------

### Install and Build Sync Engine

Source: https://github.com/stripe/sync-engine/blob/dev/AGENTS.md

Commands to install dependencies, build the project, and run tests. Ensure the project is built before running CLI commands or e2e tests.

```sh
pnpm install
pnpm build          # required before running CLI or e2e tests
pnpm test           # unit tests (no deps needed)
pnpm test:integration  # needs local Postgres
pnpm test:e2e       # needs Docker + Stripe API keys in .env
```

--------------------------------

### Build and Run Stripe Schema Visualizer

Source: https://github.com/stripe/sync-engine/blob/dev/apps/visualizer/README.md

Common commands to build the explorer data and start the visualizer application. Use `visualizer:with-data` to rebuild explorer data before starting.

```bash
pnpm explorer:build
pnpm visualizer
pnpm visualizer:with-data
```

--------------------------------

### Add setup() and teardown() to Destination interface

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-03-18-plan-004-source-setup-teardown.md

Adds optional `setup()` and `teardown()` methods to the `Destination` interface. `setup()` provisions resources before `write()` and is idempotent. `teardown()` releases resources after `write()` completes or on sync deletion. Connectors that do not require these operations can omit them.

```typescript
export interface Destination<TConfig extends Record<string, unknown> = Record<string, unknown>> {
  spec(): ConnectorSpecification
  check(params: { config: TConfig }): Promise<CheckResult>
  write(params: {
    config: TConfig
    catalog: ConfiguredCatalog
    messages: AsyncIterableIterator<DestinationInput>
  }): AsyncIterableIterator<DestinationOutput>

  /**
   * Provision external resources needed before write() can consume data.
   * Called once when a sync starts. Idempotent.
   *
   * Examples: create destination schema/tables, ensure bucket exists,
   * create spreadsheet tabs.
   *
   * Destinations that don't need setup can omit this method.
   */
  setup?(params: { config: TConfig; catalog: ConfiguredCatalog }): Promise<void>

  /**
   * Release external resources provisioned by setup().
   * Called when a sync is deleted or permanently stopped.
   *
   * Examples: drop schema (if configured), remove temp resources.
   *
   * Destinations that don't need teardown can omit this method.
   */
  teardown?(params: { config: TConfig }): Promise<void>
}
```

--------------------------------

### Source and Destination Setup Calls

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine/catalog-workflow.md

Calls to provision resources on both the source and destination systems based on the provided configuration and catalog. This prepares for the initial sync.

```javascript
source.setup({ config, catalog })        → creates webhook endpoints, replication slots, etc.
destination.setup({ config, catalog })    → creates schemas, tables, indexes
```

--------------------------------

### Run pnpm install

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-03-29-openapi-inject-fetch.md

Execute `pnpm install` to update the lockfile after removing the `undici` dependency.

```bash
pnpm install
```

--------------------------------

### Opt-in to Start npm Registry with Compose

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/local-registries.md

This command specifically starts the npm registry service within your Docker Compose setup. It's used to keep the main `docker compose up` command fast for everyday development.

```sh
docker compose --profile npm-registry up -d npm-registry
```

--------------------------------

### Remote Engine Test Setup

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-04-01-remote-engine.md

Sets up a local Hono server for testing the remote engine. It starts the server, captures the assigned port, and makes the engine URL available for tests. Ensures proper cleanup by closing the server after all tests are completed.

```typescript
let server: ReturnType<typeof serve>
let engineUrl: string

beforeAll(async () => {
  const app = createApp(resolver)
  await new Promise<void>((resolve) => {
    server = serve({ fetch: app.fetch, port: 0 }, (info) => {
      engineUrl = `http://localhost:${info.port}`
      resolve()
    })
  })
})

afterAll(
  () =>
    new Promise<void>((resolve, reject) => server.close((err) => (err ? reject(err) : resolve())))
)
```

--------------------------------

### Configured Catalog Example

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine/catalog-workflow.md

An array of configured streams that specifies which streams to sync and their associated settings. Streams not included are not synced.

```json
{
  "streams": [
    { "stream": { "name": "customer", ... }, "sync_mode": "incremental", "destination_sync_mode": "append_dedup" },
    { "stream": { "name": "product", ... }, "sync_mode": "full_refresh", "destination_sync_mode": "overwrite" }
  ]
}
```

--------------------------------

### Start HTTP API Server

Source: https://context7.com/stripe/sync-engine/llms.txt

Starts the HTTP API server for the sync engine. Connectors are typically discovered from the system's PATH.

```sh
npx @stripe/sync-engine serve
```

--------------------------------

### Serve Bundled-Only Server

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/cli-spec.md

Starts the bundled-only server binary for the sync engine, specifying the port via an environment variable.

```sh
PORT=3000 sync-engine-serve
```

--------------------------------

### Verification Commands

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-03-18-plan-003-producer-consumer-queue.md

Shell commands to set up the PostgreSQL database, install dependencies, compile the queue package, and run its tests.

```bash
docker compose up -d postgres
pnpm install
npx tsc --noEmit -p packages/queue/tsconfig.json
cd packages/queue && pnpm test
```

--------------------------------

### Start the HTTP API server

Source: https://github.com/stripe/sync-engine/blob/dev/docs/service/cli.md

Invoke the `serve` command to start the HTTP API server. This is the default action when the CLI is invoked without a specific command. You can specify the port using the `--port` flag.

```sh
sync-engine serve --port 3000
# or
sync-engine --port 3000
```

--------------------------------

### Activity Creation Function

Source: https://github.com/stripe/sync-engine/blob/dev/docs/service/temporal.md

Shows how to create the three core activities (setup, run, teardown) by providing the service and engine URLs.

```typescript
createActivities({ serviceUrl, engineUrl })
```

--------------------------------

### Start Sync Service with Engine MITM

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Starts the sync service with the engine MITM flag, enabling it to point its engine traffic to the reverse proxy at http://127.0.0.1:9090 for inspection.

```bash
sync-service --engine-mitm
```

--------------------------------

### Serve HTTP API

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/cli-spec.md

Starts the HTTP API server for the sync engine on a specified port.

```sh
sync-engine serve --port 3000
```

--------------------------------

### Initialization State Example

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-source-stripe.md

Illustrates the initial state of the Stripe source when no existing state is present and a time range is assigned by the engine.

```json
state: {
  remaining: [
    { gte: "2018-01-01", lt: "2024-04-17", cursor: null }
  ]
}
```

--------------------------------

### CLI Usage Examples

Source: https://github.com/stripe/sync-engine/blob/dev/packages/protocol/CONNECTORS.md

Demonstrates how to use the sync-protocol CLI for interacting with source and destination connectors, including commands for `spec`, `check`, `discover`, and `read`/`write` operations with configuration and catalog parameters.

```bash
# spec (no params)
source-stripe spec

# check / discover (just --config)
source-stripe check   --config '{"api_key":"sk_test_..."}'
source-stripe discover --config '{"api_key":"sk_test_..."}'

# read (--config + --catalog, optional --state)
source-stripe read \
  --config '{"api_key":"sk_test_...","base_url":"http://localhost:12111"}' \
  --catalog '{"streams":[{"stream":{"name":"customer","primary_key":[["id"]]},"sync_mode":"full_refresh","destination_sync_mode":"append"}]}'

# write (--config + --catalog, messages from stdin pipe)
source-stripe read --config '...' --catalog '...' \
  | dest-postgres write --config '{"url":"postgres://..."}' --catalog '...'
```

--------------------------------

### Build and Test Sync-Engine CLI

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-18-engine-binary-split.md

Build the project and then test the sync-engine CLI by running its help command. This verifies that the build process is successful and the CLI binary is correctly generated and accessible.

```bash
pnpm build
node dist/bin/sync-engine.js --help
```

--------------------------------

### Build and Run Sync Test Server

Source: https://github.com/stripe/sync-engine/blob/dev/packages/test-utils/README.md

Build the package and then execute the sync test server. This command starts a Hono HTTP server that serves Stripe-compatible responses backed by Postgres.

```sh
pnpm --filter @stripe/sync-test-utils build
pnpm --filter @stripe/sync-test-utils exec sync-test-utils-server
```

--------------------------------

### POST /pipeline_setup

Source: https://context7.com/stripe/sync-engine/llms.txt

Creates the necessary tables in the destination database based on the schema defined in the source connector's OpenAPI catalog. This prepares the destination for data synchronization.

```APIDOC
## POST /pipeline_setup

### Description
Creates destination tables from the Stripe OpenAPI catalog.

### Method
POST

### Endpoint
/pipeline_setup

### Parameters
#### Request Body
- **pipeline** (object) - Required - Defines the source and destination for the setup.
  - **source** (object) - Required - Configuration for the data source.
    - **type** (string) - Required - The type of the source connector (e.g., "stripe").
    - **stripe** (object) - Required - Stripe-specific configuration.
      - **api_key** (string) - Required - Stripe API key.
      - **api_version** (string) - Required - Stripe API version.
  - **destination** (object) - Required - Configuration for the data destination.
    - **type** (string) - Required - The type of the destination connector (e.g., "postgres").
    - **postgres** (object) - Required - PostgreSQL-specific configuration.
      - **url** (string) - Required - PostgreSQL connection URL.
      - **schema** (string) - Required - The schema to use in the PostgreSQL database.
- **only** (string) - Optional - Specifies which part of the pipeline to set up ('destination').

### Request Example
```json
{
  "pipeline": {
    "source": {
      "type": "stripe",
      "stripe": { "api_key": "sk_test_...", "api_version": "2024-06-20" }
    },
    "destination": {
      "type": "postgres",
      "postgres": { "url": "postgresql://localhost/mydb", "schema": "stripe_data" }
    }
  },
  "only": "destination"
}
```

### Response
- Streams NDJSON control/log messages; no output on success means clean setup.
```

--------------------------------

### Command String Examples for Connector Strategies

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine/connector-loading.md

Illustrates the command strings produced by different connector loading strategies: commandMap, path, and npm. These strings are used by the engine to spawn connector processes.

```shell
Strategy     | Example command                              
```

```shell
commandMap | "npx @acme/source-salesforce"              
```

```shell
path       | "/path/to/node_modules/.bin/source-stripe" 
```

```shell
npm        | "npx @stripe/sync-source-stripe"           
```

--------------------------------

### Run Stripe Sync Engine Quick Start

Source: https://github.com/stripe/sync-engine/blob/dev/docs/index.md

Use this command to initiate a sync of Stripe data to PostgreSQL. Ensure you replace placeholder values with your actual Stripe API key and database URL.

```sh
npx @stripe/sync-engine sync \
  --stripe-api-key sk_live_... \
  --database-url postgresql://localhost:5432/mydb \
  --schema stripe
```

--------------------------------

### Example Output: Pipeline Teardown Log Message

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Example JSON output for pipeline teardown, showing a log message.

```json
{"type":"log","log":{"level":"info","message":"Tearing down destination resources","data":{"destination_type":"postgres"}}}
```

--------------------------------

### Full Example: Request 1 - Initial backfill

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-source-stripe.md

Illustrates the initial state and messages emitted during the first request of a backfill, where the full range is requested but not completed.

```text
Source initializes: remaining: [{ gte: "2018", lt: "2024", cursor: null }]

← stream_status: { stream: "customers", status: "start" } }
← record  { stream: "customers", data: { id: "cus_001", ... } }
  ... 100 records (page 1) ...
← state   { stream: "customers", data: { remaining: [{ gte: "2018", lt: "2024", cursor: "cus_100" }] } }
← record  { stream: "customers", data: { ... } }
  ... 100 records (page 2) ...
← state   { stream: "customers", data: { remaining: [{ gte: "2018", lt: "2024", cursor: "cus_200" }] } }
  ... pages 3-50 (5000 records total) ...
← state   { stream: "customers", data: { remaining: [{ gte: "2018", lt: "2024", cursor: "cus_5000" }] } }
  ... source cut off (time limit / state limit) ...

← end     { has_more: true }
```

--------------------------------

### Start Server on Custom Port

Source: https://context7.com/stripe/sync-engine/llms.txt

Starts the HTTP API server on a custom port (e.g., 8080). Connectors are resolved from the PATH.

```sh
PORT=8080 npx @stripe/sync-engine serve
```

--------------------------------

### Implement get() for Reconstructing SyncState

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-04-better-state.md

Modify the get() method in createPgStateStore to query for both stream and global state rows and reconstruct the SyncState object.

```typescript
async get(syncId: string): Promise<SyncState | undefined> {
  const { rows } = await pool.query<{ stream: string; state: unknown }>(
    sql`SELECT stream, state FROM "${schema}"."_sync_state" WHERE sync_id = $1`,
    [syncId]
  )
  if (rows.length === 0) return undefined
  const streams: Record<string, unknown> = {}
  let global: Record<string, unknown> = {}
  for (const row of rows) {
    if (row.stream === '_global') {
      global = row.state as Record<string, unknown>
    } else {
      streams[row.stream] = row.state
    }
  }
  return { streams, global }
}
```

--------------------------------

### Example Output: Pipeline Sync Batch Result

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Example JSON output for a successful batch sync operation, including status and ending state.

```json
{
  "status": "succeeded",
  "has_more": true,
  "ending_state": {
    "source": { "streams": { "customer": { "cursor": "2" } }, "global": {} },
    "destination": {},
    "sync_run": { "run_id": "run_batch" }
  },
  "run_progress": { "derived": { "status": "started" }, "streams": {} },
  "request_progress": { "derived": { "status": "started" }, "streams": {} }
}
```

--------------------------------

### Stream Metadata Examples

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine/protocol.md

Examples of stream metadata for Stripe and Postgres sources. This metadata provides source-specific context like API versions or schema names.

```typescript
// Stripe source
{ api_version: "2025-04-30.basil", account_id: "acct_123", live_mode: true }
```

```typescript
// Postgres source
{ schema: "public", database: "mydb" }
```

--------------------------------

### Start mitmweb Reverse Proxy

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Starts a reverse proxy on port 9090 and its UI on 9091, forwarding traffic to a local engine listening on port 3000.

```bash
scripts/mitmweb-reverse-proxy.sh http://localhost:3000
```

--------------------------------

### Invariant Violation After Streaming Started

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-20-stream-message-state-machine.md

Demonstrates an invariant violation that occurs after streaming has started, resulting in an EOF message with an 'error' reason and specific violation details.

```json
{"type":"log","log":{"level":"info","message":"starting sync"}}
{"type":"progress","progress":{"derived":{"status":"started"}}}
{"type":"eof","eof":{"reason":"error","has_more":false,"error":{"code":"invariant_violation","message":"progress emitted before stream start"},"ending_state":{},"run_progress":{},"request_progress":{}}}
```

--------------------------------

### Successful Bounded Sync Example

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-20-stream-message-state-machine.md

Illustrates a successful bounded sync process, showing the sequence of log, progress, source state, and EOF messages with a time limit reason.

```json
{"type":"log","log":{"level":"info","message":"starting sync"}}
{"type":"progress","progress":{"derived":{"status":"started"}}}
{"type":"source_state","source_state":{"state_type":"stream","stream":"customer","data":{"cursor":"cus_123"}}}
{"type":"eof","eof":{"reason":"time_limit","has_more":true,"ending_state":{},"run_progress":{},"request_progress":{}}}
```

--------------------------------

### Kafka Partitioning Example

Source: https://github.com/stripe/sync-engine/blob/dev/docs/architecture/cloud.md

Visualizes how sync IDs are mapped to Kafka partitions and subsequently consumed by destination workers, ensuring ordering and single-consumer processing per sync.

```plaintext
sync_abc --+
sync_def --+--> partition 0 --> consumer 0 (destination worker)
           |
sync_ghi --+--> partition 1 --> consumer 1 (destination worker)
           |
sync_jkl --+--> partition 2 --> consumer 2 (destination worker)
```

--------------------------------

### GET /meta/sources and GET /meta/destinations

Source: https://context7.com/stripe/sync-engine/llms.txt

Retrieves metadata about available connector types, including their configuration schema. This helps in understanding what parameters are required for configuring sources and destinations.

```APIDOC
## GET /meta/sources and GET /meta/destinations

### Description
Lists available connector types with their JSON Schema configuration specs.

### Method
GET

### Endpoints
- /meta/sources
- /meta/destinations

### Usage Examples

#### List available sources
```sh
curl -s http://localhost:3000/meta/sources | jq '.items[].type'
# "stripe"
```

#### Get the Stripe source config schema
```sh
curl -s http://localhost:3000/meta/sources/stripe | jq '.config_schema.properties | keys'
# ["api_key", "api_version", "account_id", "base_url", "rate_limit", "webhook_url", ...]
```

#### List available destinations
```sh
curl -s http://localhost:3000/meta/destinations | jq '.items[].type'
# "postgres"
```

#### Get the Postgres destination config schema
```sh
curl -s http://localhost:3000/meta/destinations/postgres | jq '.config_schema.properties | keys'
# ["url", "schema", "batch_size", "ssl_ca_pem", "aws"]
```
```

--------------------------------

### Create Dockerfile for Sync Service

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-04-02-service-docker-e2e-plan.md

Builds a Docker image for the sync-service, including installing dependencies and deploying the production bundle. Expects pre-built dist/ directories.

```dockerfile
# Install deps and create standalone deployment
# Expects pre-built dist/ directories in the build context (from `pnpm build`)
FROM node:24-alpine AS build

ENV PNPM_HOME="/pnpm"
ENV PATH="$PNPM_HOME:$PATH"

RUN corepack enable

WORKDIR /app
COPY . ./ 

RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile
RUN pnpm --filter @stripe/sync-service deploy --prod /deploy

# Final image — just the bundle + external node_modules
FROM node:24-alpine
WORKDIR /app

COPY --from=build /deploy/package.json ./
COPY --from=build /deploy/dist ./dist
COPY --from=build /deploy/node_modules ./node_modules

ENV NODE_ENV=production
ENTRYPOINT ["node", "dist/bin/sync-service.js"]
CMD ["serve", "--temporal-address", "temporal:7233", "--temporal-task-queue", "sync-engine"]

```

--------------------------------

### Start Payload Type Definition

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-start-end-message.md

Defines the structure of the 'start' message payload, which will be the first line of the NDJSON request body. It carries pipeline configuration and optional sync state.

```typescript
type StartPayload = {
  pipeline: PipelineConfig // was: JSON body `pipeline` or X-Pipeline header
  state?: SyncState // was: JSON body `state` or X-State header
  state_limit?: number // was: query param ?state_limit
  time_limit?: number // was: query param ?time_limit
}
```

--------------------------------

### Define Service Docker E2E Test Suite

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-04-02-service-docker-e2e-plan.md

Sets up the main test suite using `describeWithEnv` to manage environment variables. It defines `beforeAll` and `afterAll` hooks for setting up and tearing down the Dockerized service environment, including building, starting, and stopping containers, and initializing a PostgreSQL connection.

```typescript
// ---------------------------------------------------------------------------
// Suite
// ---------------------------------------------------------------------------

describeWithEnv(
  'service docker e2e: stripe → postgres',
  ['STRIPE_API_KEY'],
  ({ STRIPE_API_KEY }) => {
    let pool: pg.Pool
    let schema: string

    beforeAll(async () => {
      schema = `docker_e2e_${Date.now()}`

      // 1. Build TypeScript so Dockerfiles have fresh dist/
      console.log('\n  Building packages...')
      execSync('pnpm build', { cwd: REPO_ROOT, stdio: 'pipe' })

      // 2. Start engine + service + worker containers (infra already running)
      console.log('  Starting containers...')
      execSync(`${COMPOSE_CMD} up --build -d engine service worker`, {
        cwd: REPO_ROOT,
        stdio: 'pipe',
      })

      // 3. Wait for service HTTP API to be ready
      console.log('  Waiting for service health...')
      await pollUntil(async () => {
        try {
          const r = await fetch(`${SERVICE_URL}/health`)
          return r.ok
        } catch {
          return false
        }
      })

      // 4. Open Postgres pool on host-mapped port for verification
      pool = new pg.Pool({ connectionString: POSTGRES_HOST_URL })
      await pool.query('SELECT 1')

      console.log(`  Service:  ${SERVICE_URL}`)
      console.log(`  Schema:   ${schema}`)
      console.log(`  Postgres: ${POSTGRES_HOST_URL}`)
      console.log(`  Cleanup:  ${SKIP_CLEANUP ? 'no (SKIP_CLEANUP=1)' : 'yes'}`)
    }, 5 * 60_000) // 5 min — includes docker build

    afterAll(async () => {
      if (!SKIP_CLEANUP) {
        await pool?.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`).catch(() => {})
      }
      await pool?.end().catch(() => {})

      // Stop only app containers — leave infra (postgres, temporal, stripe-mock) running
      execSync(`${COMPOSE_CMD} stop engine service worker`, { cwd: REPO_ROOT, stdio: 'pipe' })
      execSync(`${COMPOSE_CMD} rm -f engine service worker`, { cwd: REPO_ROOT, stdio: 'pipe' })
    })

    it('create pipeline → data lands in Postgres → delete', async () => {
      const c = api()

      // --- Create ---
      const { data: created, error: createErr } = await c.POST('/pipelines', {
        body: {
          source: { type: 'stripe', api_key: STRIPE_API_KEY },
          destination: {
            type: 'postgres',
            connection_string: POSTGRES_CONTAINER_URL,
            schema,
          },
          streams: [{ name: 'product', backfill_limit: 500 }],
        },
      })
      expect(createErr).toBeUndefined()
      expect(created!.id).toMatch(/^pipe_/) 
      const id = created!.id
      console.log(`\n  Pipeline: ${id}`)

      // --- Wait for data ---
      await pollUntil(async () => {
        try {
          const r = await pool.query(`SELECT count(*)::int AS n FROM "${schema}"."product"`)
          return r.rows[0].n > 0
        } catch {
          return false
        }
      })

      const { rows } = await pool.query(`SELECT count(*)::int AS n FROM "${schema}"."product"`)
      console.log(`  Synced:   ${rows[0].n} products`)
      expect(rows[0].n).toBeGreaterThan(0)

      // Verify shape
    })
  }
)
```

--------------------------------

### Response Stream Example with End Message

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-start-end-message.md

An example of a response NDJSON stream, including record messages and the final 'end' message. The 'end' message signals completion and provides progress information.

```json
{"type":"record","record":{"stream":"customer","data":{"id":"cus_123"}}}
{"type":"source_state","source_state":{"stream":"customer","data":{"starting_after":"cus_123"}}}
{"type":"end","end":{"reason":"complete","has_more":false,"ending_state":{"source":{"streams":{"customer":{"starting_after":null}},"global":{}},"engine":{}},"request_progress":{"elapsed_ms":3200,"run_record_count":5000,"rows_per_second":1562,"state_checkpoint_count":2},"stream_progress":{"customer":{"status":"complete","run_record_count":5000,"records_per_second":1562}}}}
```

--------------------------------

### Start mitmweb Forward Proxy

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Starts a forward proxy on port 9080 and its UI on 9081. Exports HTTP/HTTPS_PROXY environment variables and sets NODE_OPTIONS to ensure Node fetch uses the proxy.

```bash
source scripts/mitmweb-forward-proxy.sh
```

--------------------------------

### Verify engine build

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-04-better-state.md

Run the build command for the sync-engine and check the last 10 lines of the output for any errors.

```bash
pnpm --filter @stripe/sync-engine build 2>&1 | tail -10
```

--------------------------------

### Start, Publish, and Stop Local npm Registry

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/local-registries.md

Use these commands to manage a local Verdaccio npm registry for testing package publishing. This includes starting the registry, publishing all packages, and then stopping the registry.

```sh
# Start
docker compose --profile npm-registry up -d npm-registry

# Publish all packages and smoke test
bash tests/e2e-publish.sh

# Stop
docker compose --profile npm-registry down
```

--------------------------------

### End-to-End Test Script for Service Docker

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/completed/2026-04-02-service-docker-e2e.md

The end-to-end test script orchestrates the Docker Compose setup, runs the test for the Stripe to Postgres pipeline, and cleans up resources. It includes setup, test execution, and teardown phases.

```typescript
// beforeAll (timeout 5 min):
// 1. pnpm build (ensures dist/ is fresh for Docker build context)
// 2. docker compose -f compose.yml -f compose.service.yml up --build -d
// 3. pollUntil http://localhost:4020/health returns 200 (timeout 2 min)
// 4. Open Postgres pool on localhost:55432

// Test: stripe → postgres via docker containers:
// - POST /pipelines to localhost:4020 with:
//   - source: { type: 'stripe', api_key: STRIPE_API_KEY } — real Stripe
//   - destination: { type: 'postgres', connection_string: 'postgresql://postgres:postgres@postgres:5432/postgres', schema: SCHEMA }
//   - streams: [{ name: 'product', backfill_limit: 500 }]
// - pollUntil rows appear in "SCHEMA"."product" on localhost:55432
// - Assert count > 0 and id matches /^prod_/
// - DELETE /pipelines/{id} and assert { deleted: true }
// - Assert GET /pipelines/{id} returns 404
// - Assert GET /pipelines no longer lists the pipeline

// afterAll:
// - Drop test schema (unless SKIP_CLEANUP=1)
// - docker compose -f compose.yml -f compose.service.yml stop engine service worker
// - docker compose -f compose.yml -f compose.service.yml rm -f engine service worker
// - Close Postgres pool
```

--------------------------------

### Run Full Sync (JSON Config File)

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/cli-spec.md

Executes the sync pipeline using parameters defined in a JSON configuration file.

```sh
sync-engine sync --config sync.json
```

--------------------------------

### Client Sync Loop with Start and End Messages

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine-refactor/sync-lifecycle-start-end-message.md

This TypeScript code demonstrates the client's loop for data synchronization. It sends a 'start' message to the engine and then iterates through NDJSON responses, handling individual messages and updating the state based on 'end' messages until the sync is complete.

```typescript
let state: SyncState | undefined
do {
  const response = await fetch('/pipeline_sync', {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-ndjson' },
    body:
      JSON.stringify({
        type: 'start',
        start: { pipeline, state, time_limit: 30 },
      }) + '\n',
  })

  let end: EndPayload
  for await (const msg of parseNdjson(response.body)) {
    if (msg.type === 'end') end = msg.end
    else handleMessage(msg)
  }

  state = end.ending_state
} while (end.has_more)
```

--------------------------------

### Config Resolution - `--config` Flag Usage

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/cli-spec.md

Demonstrates how to use the `--config` flag to provide configuration either as a file path or as inline JSON.

```sh
--config
```

--------------------------------

### Stripe to External Postgres Sync Setup

Source: https://github.com/stripe/sync-engine/blob/dev/docs/service/scenarios.md

Set up a standard sync from Stripe to a user's Postgres database. This includes creating credentials for both sources and destinations, and configuring the sync with specific streams.

```bash
POST /credentials  { type: "stripe", api_key: "sk_test..." }
POST /credentials  { type: "postgres", host: "user-db.example.com", ... }
POST /syncs        { source: { type: "stripe", credential_id: "cred_stripe_..." },
                     destination: { type: "postgres", schema_name: "stripe", credential_id: "cred_pg_..." } }
```

--------------------------------

### NDJSON Example Payload

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/demo.md

Illustrates the newline-delimited JSON format for messages, including record, state, and log types.

```bash
{"type":"record","stream":"product","data":{"id":"prod_1","name":"Widget"},"emitted_at":"2024-01-01T00:00:00.000Z"}
{"type":"state","stream":"product","data":{"cursor":"evt_123"}}
{"type":"log","level":"info","message":"Fetched page 1"}
```

--------------------------------

### Run Unit Tests for Protocol Helpers

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/active/2026-04-03-structured-stream-state.md

Execute unit tests for the protocol helper functions. Navigate to the protocol package directory before running the test command.

```sh
cd packages/protocol && pnpm test
```

--------------------------------

### Destination Interface

Source: https://github.com/stripe/sync-engine/blob/dev/docs/slides/knowledge-transfer.md

Defines the methods available for interacting with a data destination, including spec, check, write, setup, and teardown.

```APIDOC
## Destination Interface

### Methods

- **spec()**: AsyncIterable<SpecMessage | LogMessage>
  - Description: Retrieves the specification of the destination.
- **check(params: { config: TConfig })**: AsyncIterable<ConnectionStatusMessage | LogMessage>
  - Description: Checks the connection status of the destination with the given configuration.
- **write(params: { config: TConfig, catalog: ConfiguredCatalog }, $stdin: AsyncIterable<DestinationInput>)**: AsyncIterable<DestinationOutput>
  - Description: Writes data to the destination using the provided configuration and input stream.
- **setup?(params: { config: TConfig, catalog: ConfiguredCatalog })**: AsyncIterable<ControlMessage | LogMessage>
  - Description: Sets up the destination with the given configuration and catalog. (Optional)
- **teardown?(params: { config: TConfig })**: AsyncIterable<LogMessage>
  - Description: Tears down the destination with the given configuration. (Optional)
```

--------------------------------

### macOS Launchd Plist for Temporal Service

Source: https://github.com/stripe/sync-engine/blob/dev/docs/plans/2026-04-06-self-contained-local-bundle.md

Example launchd plist configuration for the Temporal service on macOS. This is generated and not hand-written.

```xml
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" ...>
<plist version="1.0">
<dict>
  <key>Label</key><string>com.stripe.sync.temporal</string>
  <key>ProgramArguments</key>
  <array>
    <string>/path/to/node</string>
    <string>/path/to/stripe-sync-temporal.js</string>
    <string>--data-dir</string>
    <string>~/.stripe-sync</string>
  </array>
  <key>KeepAlive</key><true/>
  <key>ThrottleInterval</key><integer>10</integer>
  <key>RunAtLoad</key><true/>
  <key>StandardOutPath</key><string>~/.stripe-sync/logs/temporal.log</string>
  <key>StandardErrorPath</key><string>~/.stripe-sync/logs/temporal.err</string>
</dict>
</plist>
```

--------------------------------

### Run Full Sync (Minimal)

Source: https://github.com/stripe/sync-engine/blob/dev/docs/guides/cli-spec.md

Executes the complete sync pipeline with minimal configuration, connecting Stripe to Postgres.

```sh
sync-engine sync \
  --stripe-api-key sk_test_... \
  --postgres-url postgres://localhost/mydb
```

--------------------------------

### Backfill to Live Transition in Source Read

Source: https://github.com/stripe/sync-engine/blob/dev/docs/engine/ARCHITECTURE.md

Illustrates the sequence of messages emitted by `source.read()` during a backfill to live transition. The source notes the backfill start time and ensures the live phase picks up with overlap, with duplicates deduped by the destination.

```text
source.read(params)
  → RecordMessage  (backfill)
  → RecordMessage  (backfill)
  → ...
  → StateMessage   {stream: "customer", data: {phase: "live", cursor: "2026-03-17T00:00:00Z"}}
  → RecordMessage  (live)
  → RecordMessage  (live)
  → ...            (infinite)
```