--- title: "Run sync pipeline (read → write)" method: POST path: "/pipeline_sync" tags: ["Stateless Sync API"] --- # Run sync pipeline (read → write) `POST /pipeline_sync` Without a request body, reads from the source connector and writes to the destination (backfill mode). With an NDJSON request body, uses the provided messages as input instead of reading from the source (push mode — e.g. piped webhook events). Alternatively, send Content-Type: application/json with {pipeline, state?, body?} to pass config in the body. ## Query parameters - `state_limit` integer — Stop streaming after N state messages. - `time_limit` number — Stop streaming after N seconds. ## Request body - object - `pipeline` PipelineConfig, required - `source` SourceConfig, required - `type` 'stripe', required - `stripe` SourceStripeConfig, required - `api_key` string, required — Stripe API key (sk_test_... or sk_live_...) - `account_id` string — Stripe account ID (resolved from API if omitted) - `account_created` integer — Stripe account creation timestamp in unix seconds (resolved from API if omitted) - `livemode` boolean — Whether this is a live mode sync - `api_version` '2026-03-25.dahlia' | '2026-02-25.clover' | '2026-01-28.clover' | '2025-12-15.clover' | '2025-11-17.clover' | '2025-10-29.clover' | '2025-09-30.clover' | '2025-08-27.basil' | '2025-07-30.basil' | '2025-06-30.basil' | '2025-05-28.basil' | '2025-04-30.basil' | '2025-03-31.basil' | '2025-02-24.acacia' | '2025-01-27.acacia' | '2024-12-18.acacia' | '2024-11-20.acacia' | '2024-10-28.acacia' | '2024-09-30.acacia' | '2024-06-20' | '2024-04-10' | '2024-04-03' | '2023-10-16' | '2023-08-16' | '2022-11-15' | '2022-08-01' | '2020-08-27' | '2020-03-02' | '2019-12-03' | '2019-11-05' | '2019-10-17' | '2019-10-08' | '2019-09-09' | '2019-08-14' | '2019-05-16' | '2019-03-14' | '2019-02-19' | '2019-02-11' | '2018-11-08' | '2018-10-31' | '2018-09-24' | '2018-09-06' | '2018-08-23' | '2018-07-27' | '2018-05-21' | '2018-02-28' | '2018-02-06' | '2018-02-05' | '2018-01-23' | '2017-12-14' | '2017-08-15' - `base_url` string, uri — Override the Stripe API base URL (e.g. http://localhost:12111 for stripe-mock) - `webhook_url` string, uri — URL for managed webhook endpoint registration - `webhook_secret` string — Webhook signing secret (whsec_...) for signature verification - `websocket` boolean — Enable WebSocket streaming for live events - `poll_events` boolean — Enable events API polling for incremental sync after backfill - `webhook_port` integer — Port for built-in webhook HTTP listener (e.g. 4242) - `revalidate_objects` string[] — Object types to re-fetch from Stripe API on webhook (e.g. ["subscription"]) - `backfill_limit` integer — Max objects to backfill per stream (useful for testing) - `max_concurrent_streams` integer — Max streams paginating in parallel (default: 5, capped at catalog size). - `rate_limit` integer — Override max requests per second (default: auto-derived from API key mode — 20 live, 10 test). - `destination` union, required - object - `type` 'postgres', required - `postgres` DestinationPostgresConfig, required - `url` string — Postgres connection string - `connection_string` string — Deprecated alias for url; prefer url - `schema` string — Target schema name (e.g. "stripe") - `batch_size` number — Records to buffer before flushing - `aws` object — AWS RDS IAM authentication config - `host` string, required — Postgres host for RDS IAM auth - `port` number — Postgres port for RDS IAM auth - `database` string, required — Database name for RDS IAM auth - `user` string, required — Database user for RDS IAM auth - `region` string, required — AWS region for RDS instance - `role_arn` string — IAM role ARN to assume (cross-account) - `external_id` string — External ID for STS AssumeRole - `ssl_ca_pem` string — PEM-encoded CA certificate for SSL verification (required for verify-ca / verify-full with a private CA) - object - `type` 'google_sheets', required - `google_sheets` DestinationGoogleSheetsConfig, required - `client_id` string — Google OAuth2 client ID (env: GOOGLE_CLIENT_ID) - `client_secret` string — Google OAuth2 client secret (env: GOOGLE_CLIENT_SECRET) - `access_token` string, required — OAuth2 access token - `refresh_token` string, required — OAuth2 refresh token - `spreadsheet_id` string — Target spreadsheet ID (created if omitted) - `spreadsheet_title` string — Title when creating a new spreadsheet - `batch_size` number — Rows per Sheets API append call - `streams` object[] - `name` string, required — Stream (table) name to sync. - `sync_mode` 'incremental' | 'full_refresh' — How the source reads this stream. Defaults to full_refresh. - `fields` string[] — If set, only these fields are synced. - `backfill_limit` integer — Cap backfill to this many records, then mark the stream complete. - `state` SyncState — Full sync checkpoint with separate sections for source, destination, and sync run. Connectors only see their own section; the engine manages routing. - `source` SourceState, required — Source connector state — cursors, backfill progress, events cursors. - `streams` object, required — Per-stream checkpoint data, keyed by stream name. - `global` object, required — Source-wide state shared across all streams. - `destination` object, required — Destination connector state. - `sync_run` object, required — Engine-managed run state — sync_run_id, time_ceiling, accumulated progress. - `sync_run_id` string — Identifies a finite backfill run. Omit for continuous sync. - `time_ceiling` string — Frozen upper bound (ISO 8601). Set on first invocation when sync_run_id is present; reused on continuation. - `progress` object, required — Accumulated progress from prior requests in this run. - `started_at` string, required — When this sync started (ISO 8601); generally equals time_ceiling. - `elapsed_ms` integer, required — Wall-clock milliseconds since the sync run started. - `global_state_count` integer, required — Total source_state messages observed so far. - `connection_status` object — Set when source or destination emits connection_status: failed. - `status` 'succeeded' | 'failed', required — Whether the connection check passed. - `message` string — Human-readable explanation of the check result. - `derived` object, required — Computed aggregates. - `status` 'started' | 'succeeded' | 'failed', required — succeeded = all streams completed/skipped; failed = connection_status failed OR any stream errored. - `records_per_second` number, required — Overall throughput for the entire run. - `states_per_second` number, required — State checkpoints per second. - `streams` object, required — Per-stream progress, keyed by stream name. - `body` unknown[] - unknown ## Response `200` NDJSON stream of sync messages ## Other responses - `400` — Invalid params ## Changes - **2026-04-20** `cef36747fb38` — 4 breaking, 8 warning, 8 info - added the new required request property `pipeline/destination/oneOf[subschema #1]/postgres/aws/database` - added the new required request property `pipeline/destination/oneOf[subschema #1]/postgres/aws/host` - added the new required request property `pipeline/destination/oneOf[subschema #1]/postgres/aws/user` - the response property `oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #2]/destination_config/oneOf[#/components/schemas/DestinationPostgresConfig]/schema` became optional for the status `200` - …16 more - **2026-04-19** `846342af247c` — 2 info - added the new optional request property `pipeline/source/oneOf[subschema #1]/stripe/account_created` - added the optional property `oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #1]/source_config/account_created` to the response with the `200` status - …earlier changes not shown [Full history](https://skmtc.dev/stripe/apis/stripe-sync-engine/changes/pipeline_sync/post.md) --- [API](https://skmtc.dev/stripe/apis/stripe-sync-engine.md) · [All operations](https://skmtc.dev/stripe/apis/stripe-sync-engine/llms.txt) · [OpenAPI document](https://skmtc-service-production.skmtc.workers.dev/v1/apis/stripe/stripe-sync-engine/revisions/cef36747fb38/schema)