--- title: "Write records to destination" method: POST path: "/pipeline_write" tags: ["Stateless Sync API"] --- # Write records to destination `POST /pipeline_write` Writes messages to the destination. Pass an array of messages in the request body. ## Request body - object - `pipeline` PipelineConfig, required - `source` union, required - object - `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) - `rate_limit` integer — Override max requests per second (default: auto-derived from API key mode — 20 live, 10 test). - object - `type` 'metronome', required - `metronome` SourceMetronomeConfig, required - `api_key` string, required — Metronome API bearer token - `base_url` string, uri — Override the Metronome API base URL (default: https://api.metronome.com) - `rate_limit` integer — Max requests per second (default: no limit) - `backfill_limit` integer — Max records to fetch per stream (useful for testing) - `webhook_secret` string — Webhook signing secret for HMAC-SHA256 signature verification - `webhook_port` integer — Port for built-in webhook HTTP listener (e.g. 4243) - `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, nullable - `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 - object - `type` 'redis', required - `redis` DestinationRedisConfig, required - `url` string — Redis connection URL (redis://host:port) - `host` string — Redis host (default: localhost) - `port` number — Redis port (default: 6379) - `password` string — Redis password - `db` number — Redis database number (default: 0) - `tls` boolean — Enable TLS - `key_prefix` string — Prefix for all Redis keys (default: empty) - `batch_size` number — Records to buffer before flushing via pipeline - `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. - `stdin` Message[], required — Array of messages to write to the destination. - union - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'record', required - `record` object, required — One record for one stream. - `stream` string, required — Stream (table) name this record belongs to. - `data` object, required — The record payload as a key-value map. - `emitted_at` string, date-time, required — ISO 8601 timestamp when the record was emitted by the source. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'source_state', required - `source_state` union, required - object — Per-stream checkpoint for resumable syncs. - `state_type` 'stream' - `stream` string, required — Stream being checkpointed. - `data` unknown, required - object — Sync-wide checkpoint shared across all streams. - `state_type` 'global', required - `data` unknown, required - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'catalog', required - `catalog` object, required — Catalog of available streams. - `streams` object[], required — All streams available from this source. - `name` string, required — Collection name (e.g. "customers", "invoices", "pg_public.users"). - `primary_key` array[], required — Paths to fields that uniquely identify a record within this stream. Supports composite keys and nested paths. e.g. [["id"]] or [["account_id"], ["created"]] - … - `json_schema` object — JSON Schema describing the record shape. Discovered at runtime or provided by config. - `metadata` object — Source-specific metadata that applies to every record in this stream. The destination can use these for schema naming, partitioning, etc. Examples: Stripe: { api_version, account_id, live_mode }. - `newer_than_field` string, required — Field whose value increases monotonically. Destination uses it to skip stale writes (e.g. "updated"). - `soft_delete_field` string — Field in record data that signals a soft delete (e.g. "deleted"). Destination uses this to classify upserts as deletes when the field is truthy. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'log', required - `log` object, required — Structured log output from a connector. - `level` 'debug' | 'info' | 'warn' | 'error', required — Log severity level. - `message` string, required — Human-readable log message. - `data` object — Structured log fields emitted alongside the message. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'spec', required - `spec` object, required — JSON Schema describing the configuration a connector requires. - `config` object, required — JSON Schema for the connector's configuration object. - `source_state_stream` object — JSON Schema for per-stream state (cursor/checkpoint shape). See also SourceState.global for sync-wide cursors. - `source_input` object — JSON Schema for the read() input parameter (e.g. a webhook event). - `soft_limit_fraction` number — Fraction of `time_limit` to use as default `soft_time_limit` (e.g. 0.5). - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'connection_status', required - `connection_status` object, required — Result of a connection check. - `status` 'succeeded' | 'failed', required — Whether the connection check passed. - `message` string — Human-readable explanation of the check result. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'stream_status', required - `stream_status` union, required — Stream lifecycle event. Sources emit these; the engine tracks stream progress from them. - object - `stream` string, required — Stream being reported on. - `status` 'start', required - `time_range` object — Full backfill time span for this stream. - … - object - `stream` string, required — Stream being reported on. - `status` 'range_complete', required - `range_complete` object, required — The sub-range that finished. - … - object - `stream` string, required — Stream being reported on. - `status` 'complete', required - object - `stream` string, required — Stream being reported on. - `status` 'error', required - `error` string, required — Human-readable error description. - object - `stream` string, required — Stream being reported on. - `status` 'skip', required - `reason` string, required — Why the stream was skipped. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'control', required - `control` union, required — Control signal from a connector to the orchestrator. - object - `control_type` 'source_config', required - `source_config` union, required - … - object - `control_type` 'destination_config', required - `destination_config` union, required - … - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'progress', required - `progress` ProgressPayload, required — Periodic sync progress emitted by the engine as a top-level message. Each emission is a full replacement. - `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. - `total_record_count` integer, required — Total records across all streams. - `total_state_count` integer, required — Total source_state messages across all streams. - `streams` object, required — Per-stream progress, keyed by stream name. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'eof', required - `eof` EofPayload, required — Terminal message signaling end of this request. - `status` 'started' | 'succeeded' | 'failed', required — succeeded = all streams completed/skipped; failed = connection_status failed OR any stream errored. - `has_more` boolean, required — Whether the client should continue with another request. true when cut off by limits; false when the source iterator exhausted naturally. - `ending_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. - … - `destination` object, required — Destination connector state. - `sync_run` object, required — Engine-managed run state — run_id, time_ceiling, accumulated progress. - … - `run_progress` ProgressPayload, required — Periodic sync progress emitted by the engine as a top-level message. Each emission is a full replacement. - `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. - … - `derived` object, required — Computed aggregates. - … - `streams` object, required — Per-stream progress, keyed by stream name. - `request_progress` ProgressPayload, required — Periodic sync progress emitted by the engine as a top-level message. Each emission is a full replacement. - `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. - … - `derived` object, required — Computed aggregates. - … - `streams` object, required — Per-stream progress, keyed by stream name. - object - `_emitted_by` string — Who emitted this message: "source/{type}", "destination/{type}", or "engine". Set by the engine. - `_ts` string, date-time — ISO 8601 timestamp when the engine observed this message. - `type` 'source_input', required - `source_input` unknown, required ## Response `200` NDJSON stream of write result messages ## Other responses - `400` — Invalid params ## Changes - **2026-04-30** `4b2cc3fc8044` — 4 info - added the new optional request property `pipeline/source/oneOf[subschema #2]/metronome/webhook_port` - added the new optional request property `pipeline/source/oneOf[subschema #2]/metronome/webhook_secret` - added the new optional request property `stdin/items/oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #1]/source_config/oneOf[#/components/schemas/SourceMetronomeConfig]/webhook_port` - added the new optional request property `stdin/items/oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #1]/source_config/oneOf[#/components/schemas/SourceMetronomeConfig]/webhook_secret` - **2026-04-30** `3b4bee18fd4d` — 1 breaking, 3 info - removed `subschema #1, subschema #2` from the `stdin/items/oneOf[#/components/schemas/ControlMessage]/control` request property `oneOf` list - added `subschema #1, subschema #2` to the `stdin/items/oneOf[#/components/schemas/ControlMessage]/control` request property `oneOf` list - added `subschema #2` to the `pipeline/source` request property `oneOf` list - added `subschema #3` to the `pipeline/destination` request property `oneOf` list - …earlier changes not shown [Full history](https://skmtc.dev/stripe/apis/stripe-sync-engine/changes/pipeline_write/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/4b2cc3fc8044/schema)