removedInput schema / $defs
Removed value: -{
- "AggregateConsistency": {
- "description": "How an `aggregate` middleware keeps its states consistent.",
- "oneOf": [
- {
- "const": "shared",
- "description": "Loads and writes the states of every batch; correct with several instances. Needs a `store`.",
- "type": "string"
- },
- {
- "const": "single_writer",
- "description": "Keeps the states in memory and writes them behind; correct with one instance only. Without a `store` they are lost on restart.",
- "type": "string"
- }
- ]
- },
- "AggregateEmit": {
- "description": "Which state an `aggregate` entry writes into the message.",
- "oneOf": [
- {
- "const": "updated",
- "description": "The state including this message.",
- "type": "string"
- },
- {
- "const": "previous",
- "description": "The state before this message; `null` for a key seen for the first time.",
- "type": "string"
- }
- ]
- },
- "AggregateEntry": {
- "additionalProperties": false,
- "description": "One aggregate of an `aggregate` middleware.",
- "properties": {
- "emit": {
- "$ref": "#/$defs/AggregateEmit",
- "default": "updated",
- "description": "Which state the message carries: after (`updated`) or before (`previous`) its own update."
- },
- "expression": {
- "description": "Expression returning the new state; reads payload fields, `meta.<key>` and `state`.",
- "type": [
- "string",
- "null"
- ]
- },
- "fields": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Built-in aggregates instead of `expression`, e.g. `n: count`, `avg: ema(reading, 0.1)`.",
- "type": [
- "object",
- "null"
- ]
- },
- "into": {
- "description": "Dotted payload path the result is written to, e.g. `features.sensor`.",
- "type": "string"
- },
- "key": {
- "description": "Key template selecting the state, e.g. `${payload:sensor_id}`.",
- "type": "string"
- },
- "output": {
- "description": "Expression over `state` shaping what is written to the message. Defaults to the state.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "key",
- "into"
- ],
- "type": "object"
- },
- "AggregateMiddleware": {
- "additionalProperties": false,
- "description": "Aggregate middleware configuration.\n\nKeeps one state document per key in this process and updates it with every message:\nrunning sums, counters, moving averages. The message is not consumed; it leaves with\nthe state written into its payload. Requires the `aggregate` feature.",
- "properties": {
- "consistency": {
- "$ref": "#/$defs/AggregateConsistency",
- "default": "shared",
- "description": "`shared` needs a `store`; `single_writer` is for one instance and works without one, in memory only."
- },
- "emit": {
- "$ref": "#/$defs/AggregateEmit",
- "default": "updated",
- "description": "Which state the message carries: after (`updated`) or before (`previous`) its own update."
- },
- "entries": {
- "description": "Further aggregates over other keys, updated by the same message.",
- "items": {
- "$ref": "#/$defs/AggregateEntry"
- },
- "type": "array"
- },
- "expression": {
- "description": "Expression returning the new state; reads payload fields, `meta.<key>` and `state`.",
- "type": [
- "string",
- "null"
- ]
- },
- "fields": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Built-in aggregates instead of `expression`, e.g. `n: count`, `avg: ema(reading, 0.1)`.",
- "type": [
- "object",
- "null"
- ]
- },
- "into": {
- "description": "Dotted payload path the result is written to, e.g. `features.sensor`.",
- "type": [
- "string",
- "null"
- ]
- },
- "key": {
- "description": "Key template selecting the state, e.g. `${payload:sensor_id}`. Use `entries` for several.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_keys": {
- "description": "States kept in memory per entry; beyond it the least recently used are dropped. Defaults to 1000000, 0 is unlimited.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "on_error": {
- "$ref": "#/$defs/AggregateOnError",
- "default": "drop",
- "description": "A message that cannot be folded: `drop` it, `fail` it, or `skip` the affected entries and pass it on."
- },
- "output": {
- "description": "Expression over `state` shaping what is written to the message. Defaults to the state.",
- "type": [
- "string",
- "null"
- ]
- },
- "read_only": {
- "description": "Reads the states of the `store` and never changes them.",
- "type": "boolean"
- },
- "store": {
- "description": "Store URL keeping the states: `postgres|sqlite://…[/table]` or `mongodb://host/db[/collection]`. Without it `consistency: single_writer` is required.",
- "type": [
- "string",
- "null"
- ]
- },
- "time": {
- "description": "Payload path of the event time (epoch seconds or milliseconds, or RFC 3339) for `fields`. Defaults to the clock.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "AggregateOnError": {
- "description": "What an `aggregate` middleware does with a message it cannot fold.",
- "oneOf": [
- {
- "const": "drop",
- "description": "On an input the message is logged, acked and dropped; on an output it fails.",
- "type": "string"
- },
- {
- "const": "fail",
- "description": "On an input the message is nacked, so its source sees the failure; on an output it fails.",
- "type": "string"
- },
- {
- "const": "skip",
- "description": "Only the entries that cannot be computed are left out; the message goes on.",
- "type": "string"
- }
- ]
- },
- "AmqpConfig": {
- "additionalProperties": false,
- "description": "General AMQP connection configuration.",
- "properties": {
- "delayed_ack": {
- "default": false,
- "description": "(Publisher only) If true, do not wait for an acknowledgement when sending to broker. Defaults to false.",
- "type": "boolean"
- },
- "exchange": {
- "description": "The exchange to publish to or bind the queue to.",
- "type": [
- "string",
- "null"
- ]
- },
- "no_declare_queue": {
- "default": false,
- "description": "(Publisher only) If true, do not attempt to declare the queue. Assumes the queue already exists. Defaults to false.",
- "type": "boolean"
- },
- "no_persistence": {
- "default": false,
- "description": "If true, declare queues as non-durable (transient). Defaults to false. Affects both Consumer (queue durability) and Publisher (message persistence).",
- "type": "boolean"
- },
- "password": {
- "description": "Optional password for authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "prefetch_count": {
- "description": "(Consumer only) Number of messages to prefetch. Defaults to 100.",
- "format": "uint16",
- "maximum": 65535,
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "queue": {
- "description": "The AMQP queue name.",
- "type": [
- "string",
- "null"
- ]
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only) Include authoritative `mqb.src.amqp_*` source positions. Defaults to false.",
- "type": "boolean"
- },
- "subscribe_mode": {
- "default": false,
- "description": "(Consumer only) If true, act as a **Subscriber** (fan-out). Defaults to false.",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "url": {
- "description": "AMQP connection URI. The `lapin` client connects to a single host specified in the URI. If it contains userinfo, it will be treated as a secret.\nFor high availability, provide the address of a load balancer or use DNS resolution\nthat points to multiple brokers. Example: \"amqp://localhost:5672/vhost\".",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username for authentication.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "AwsConfig": {
- "additionalProperties": false,
- "properties": {
- "access_key": {
- "description": "AWS Access Key ID.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "binary_payload_mode": {
- "default": false,
- "description": "Use binary payloads in SQS/SNS messages.",
- "type": "boolean"
- },
- "endpoint_url": {
- "description": "Custom endpoint URL (e.g., for LocalStack).",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "max_messages": {
- "description": "(Consumer only) Maximum number of messages to receive in a batch (1-10).",
- "format": "int32",
- "maximum": 10,
- "minimum": 1,
- "type": [
- "integer",
- "null"
- ]
- },
- "queue_url": {
- "description": "The SQS queue URL. Required for Consumer. Optional for Publisher if `topic_arn` is set. If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "region": {
- "description": "AWS Region (e.g., \"us-east-1\").",
- "type": [
- "string",
- "null"
- ]
- },
- "secret_key": {
- "description": "AWS Secret Access Key.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "session_token": {
- "description": "AWS Session Token.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "topic_arn": {
- "description": "(Publisher only) The SNS topic ARN.",
- "type": [
- "string",
- "null"
- ]
- },
- "wait_time_seconds": {
- "description": "(Consumer only) Wait time for long polling in seconds (0-20).",
- "format": "int32",
- "maximum": 20,
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "BufferMiddleware": {
- "additionalProperties": false,
- "description": "Publisher-side buffer middleware configuration.\n\nBuffers outbound messages briefly so multiple single-message sends can be\nforwarded as one `send_batch` call to the wrapped publisher.",
- "properties": {
- "max_delay_ms": {
- "description": "Maximum time to wait before flushing a non-full buffer.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "max_messages": {
- "description": "Maximum number of messages to accumulate before flushing immediately.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "max_messages",
- "max_delay_ms"
- ],
- "type": "object"
- },
- "CipherKind": {
- "description": "AEAD cipher selection for [`EncryptionConfig`].",
- "oneOf": [
- {
- "const": "xchacha20poly1305",
- "description": "XChaCha20-Poly1305 (default): 192-bit random nonce, safe at high message rates.",
- "type": "string"
- },
- {
- "const": "aes256gcm",
- "description": "AES-256-GCM: 96-bit counter nonce, for interoperability with other systems.",
- "type": "string"
- }
- ]
- },
- "ClickHouseConfig": {
- "additionalProperties": false,
- "description": "ClickHouse endpoint configuration (talks the ClickHouse HTTP interface).\n\nAs a **publisher** it batch-inserts messages using `FORMAT JSONEachRow` — by default the whole\nmessage payload (which must be a JSON object) becomes one row; set `columns` to build each row\nfrom explicit `${payload:<field>}` / `${metadata:<key>}` tokens instead. As a **consumer** it\nreads an existing table **non-destructively** by paging over a monotonic `cursor_column`\n(ClickHouse has no native queue/pub-sub), serializing each row to a JSON payload.",
- "properties": {
- "async_insert": {
- "default": false,
- "description": "(Publisher only) If true, set the ClickHouse `async_insert=1` server setting so inserts are\nbuffered server-side. Defaults to false.",
- "type": "boolean"
- },
- "checkpoint_store": {
- "description": "(Consumer only) Where to persist the resume cursor. Because ClickHouse is unsuited to per-row\ncursor upserts, a durable checkpoint requires an **external** store URL:\n- `file:///var/lib/mqb/cursors.json` → local JSON file\n- `postgres://user@host/db/table` / `mysql://host/db/table` → external SQL table (table optional)\n- `mongodb://host/db/collection` → external MongoDB collection (collection optional)\n- `s3://bucket/prefix` (also `gs://`, `az://`, `abfs://`) → cloud object store; creds via env\n\nMay embed connection credentials, so it is treated as a secret.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "columns": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "(Publisher only) Optional per-column mapping. Each entry maps a target column name to a value\ntoken: `${payload:<field>}` takes the top-level JSON field `<field>` of the payload (JSON type\npreserved), `${metadata:<key>}` takes `message.metadata[\"<key>\"]` (as a string), and any other\nvalue is inserted literally. When omitted, the whole payload JSON object is inserted as one row.",
- "type": [
- "object",
- "null"
- ]
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "gzip",
- "description": "HTTP body compression for inserts and cursor reads (`none`, `gzip`, `lz4`, `zstd`). Applied\nas `Content-Encoding` on the request body and negotiated on the response via `Accept-Encoding`.\n`lz4`/`zstd` are faster than `gzip`; all are understood natively by ClickHouse. Defaults to `gzip`."
- },
- "connect_timeout_ms": {
- "description": "Connection (TCP + TLS handshake) timeout in milliseconds. Defaults to 10000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "cursor_column": {
- "description": "(Consumer only) Read an existing table **non-destructively** and resumably, paging by this\nmonotonic column (`SELECT … WHERE {cursor_column} > {last} ORDER BY {cursor_column} ASC LIMIT n`)\nand persisting the last read value under `cursor_id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "cursor_id": {
- "description": "(Consumer only) Cursor id used to key the persisted resume position. Without it, progress is not\npersisted and every restart re-copies from the beginning.",
- "type": [
- "string",
- "null"
- ]
- },
- "database": {
- "description": "Database name. Defaults to `default`.",
- "type": [
- "string",
- "null"
- ]
- },
- "lookup_query": {
- "description": "(Publisher only) SELECT for `lookup`, answering with the first row. Write `WHERE id IN\n(${payload:id})` to answer a whole batch in one query, matching rows by `id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_polling_interval_ms": {
- "description": "(Consumer only) If set, the poll interval backs off exponentially from `polling_interval_ms`\nup to this value while drained, resetting on new rows. Unset = constant interval.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "password": {
- "default": null,
- "description": "Optional password. Takes precedence over any credentials embedded in the `url`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "polling_interval_ms": {
- "description": "(Consumer only) Polling interval in milliseconds when the table is drained. Defaults to 100ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds for ClickHouse HTTP calls (inserts, cursor reads, status).\nUnset = no timeout (wait indefinitely), which suits very large batch inserts.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "select_columns": {
- "description": "(Consumer only) Columns to select in `cursor_column` mode. Defaults to `*`.",
- "type": [
- "string",
- "null"
- ]
- },
- "table": {
- "description": "The table to read from / write to. May be schema-qualified (`db.table`).",
- "type": "string"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration for `https://` connections."
- },
- "url": {
- "description": "ClickHouse HTTP endpoint URL, e.g. `http://localhost:8123` (or `https://…`). If it contains\nuserinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "default": null,
- "description": "Optional username. Takes precedence over any credentials embedded in the `url`. Defaults to `default`.",
- "type": [
- "string",
- "null"
- ]
- },
- "wait_for_async_insert": {
- "default": null,
- "description": "(Publisher only) With `async_insert`, wait for the server to flush before acking. Defaults to\ntrue (durable). False = fire-and-forget: faster, but a crash before flush can drop the batch.",
- "type": [
- "boolean",
- "null"
- ]
- }
- },
- "required": [
- "url",
- "table"
- ],
- "type": "object"
- },
- "Compression": {
- "description": "Compression algorithm. Used for at-rest batches (file, object_store) and for HTTP\nbody compression (http, clickhouse). Orthogonal to `format`.",
- "oneOf": [
- {
- "const": "none",
- "description": "No compression (default).",
- "type": "string"
- },
- {
- "const": "gzip",
- "description": "gzip; each batch is a self-contained member, so files stay readable with `zcat`.",
- "type": "string"
- },
- {
- "const": "lz4",
- "description": "lz4 frame format; each batch is a self-contained frame (`lz4 -d` compatible).",
- "type": "string"
- },
- {
- "const": "zstd",
- "description": "zstd; each batch is a self-contained frame, concatenated frames decode as one\nstream (`zstd -d` compatible). Better ratio than lz4, still fast.",
- "type": "string"
- }
- ]
- },
- "CompressionMiddleware": {
- "additionalProperties": false,
- "description": "Payload-compression middleware configuration.\n\nCompresses each message payload on the output side and decompresses it on the input\nside; metadata and routing keys are left untouched. Requires the `compression` feature.\nDistinct from the `file`/`object_store` batch `compression` field, which stays\nCLI-decodable at rest — this operates per message on any transport.",
- "properties": {
- "algorithm": {
- "$ref": "#/$defs/Compression",
- "default": "zstd",
- "description": "Algorithm: `none`, `gzip`, `lz4`, or `zstd`. Defaults to `zstd`."
- },
- "max_decompressed_bytes": {
- "default": null,
- "description": "Reject a decompressed payload larger than this many bytes (decompression-bomb guard).\nConsumer side only; unset means no limit.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "on_error": {
- "$ref": "#/$defs/InputErrorPolicy",
- "default": "fail",
- "description": "What an input does with a payload that will not decompress. Defaults to `fail`."
- }
- },
- "type": "object"
- },
- "CookieJarMiddleware": {
- "additionalProperties": false,
- "description": "Cookie/session jar middleware configuration.\n\nOptimized for HTTP by default: it can read `cookie` and `set-cookie` metadata,\npersist session cookies, and inject them into later outgoing requests.\n\nThe middleware can also capture arbitrary metadata values into the same session store\nand optionally expose stored values back into message metadata.",
- "properties": {
- "capture_metadata_keys": {
- "default": [],
- "description": "Additional metadata keys to persist into the session value store.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "cookie_metadata_key": {
- "default": "cookie",
- "description": "Metadata key used to read/write HTTP Cookie headers. Defaults to `cookie`.",
- "type": "string"
- },
- "export_metadata_prefix": {
- "default": null,
- "description": "Optional metadata prefix used to export stored values back onto each message.\n\nExported keys use `PREFIXcookie.<name>` for cookies and `PREFIXvalue.<name>` for\ncaptured generic values.",
- "type": [
- "string",
- "null"
- ]
- },
- "inject_metadata": {
- "additionalProperties": {
- "type": "string"
- },
- "default": {},
- "description": "Optional mapping of outgoing metadata keys to stored session value names.\n\nExample: `{ \"authorization\": \"access_token\" }` copies the stored value\n`access_token` into outgoing metadata key `authorization` when not already present.",
- "type": "object"
- },
- "max_cookies": {
- "default": 256,
- "description": "Maximum cookies kept per session; the least recently set are dropped. Defaults to 256.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "set_cookie_metadata_key": {
- "default": "set-cookie",
- "description": "Metadata key used to read HTTP Set-Cookie responses. Defaults to `set-cookie`.",
- "type": "string"
- },
- "shared_scope": {
- "default": null,
- "description": "Optional shared scope name. When set, middleware instances using the same scope\nshare one session store across endpoints/routes in the process.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "CsvConfig": {
- "additionalProperties": false,
- "description": "CSV dialect for `format: csv`: field separator, quoting, header and nested values.",
- "properties": {
- "columns": {
- "description": "Column names. A source uses them instead of the header's; a sink writes exactly these, in this order.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "header": {
- "description": "Whether the first record names the columns. Defaults to true; a source without one needs `columns`.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "nested": {
- "$ref": "#/$defs/CsvNested",
- "default": "flatten",
- "description": "(Sink only) Nested objects: `flatten` into `parent.child` columns (default) or `json` text in one cell."
- },
- "on_mismatch": {
- "$ref": "#/$defs/CsvMismatch",
- "description": "(Sink only) A record whose keys differ from the columns: `warn` (default) writes it anyway, `fail` rejects it."
- },
- "quote": {
- "description": "Quote character, or `none` for unquoted fields. Defaults to `\"`.",
- "type": [
- "string",
- "null"
- ]
- },
- "separator": {
- "description": "Field separator: one character, `tab`, `space`, hex (`0x1f`) or `auto` (source: guessed from the first record). Defaults to `,`.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "CsvMismatch": {
- "description": "What a CSV sink does with a record whose keys differ from the columns.",
- "oneOf": [
- {
- "const": "warn",
- "description": "Missing keys are written as empty fields and extra keys are dropped, logged once.",
- "type": "string"
- },
- {
- "const": "fail",
- "description": "The record fails as non-retryable and is not written.",
- "type": "string"
- }
- ]
- },
- "CsvNested": {
- "description": "How a CSV sink writes a nested JSON object.",
- "oneOf": [
- {
- "const": "flatten",
- "description": "One column per leaf, named by its dotted path (`stats.avg`).",
- "type": "string"
- },
- {
- "const": "json",
- "description": "The object's JSON text in a single cell.",
- "type": "string"
- }
- ]
- },
- "DatePartitionStyle": {
- "description": "Layout of the object_store sink's date folders.",
- "oneOf": [
- {
- "const": "nested",
- "description": "`YYYY/MM/DD/`.",
- "type": "string"
- },
- {
- "const": "hive",
- "description": "`year=YYYY/month=MM/day=DD/`, read as partition columns by Spark, DuckDB, DataFusion, Athena.",
- "type": "string"
- }
- ]
- },
- "DeadLetterQueueMiddleware": {
- "additionalProperties": false,
- "description": "Dead-Letter Queue (DLQ) middleware configuration.\n\nRoutes failed messages to a designated endpoint for later analysis and recovery.\nIt is recommended to pair this with the Retry middleware to avoid message loss.\n\nFailed messages are sent to the configured endpoint when they are exhausted after retry attempts.",
- "properties": {
- "endpoint": {
- "$ref": "#/$defs/Endpoint",
- "description": "The endpoint to send failed messages to."
- }
- },
- "required": [
- "endpoint"
- ],
- "type": "object"
- },
- "DecodeErrorPolicy": {
- "description": "What an `object_store` source does with an object it cannot decode after five tries.",
- "oneOf": [
- {
- "const": "skip",
- "description": "Move past it with a warning. The first object a source reads stops the route instead:\na failure there means the settings are wrong, not the object.",
- "type": "string"
- },
- {
- "const": "fail",
- "description": "Stop the route.",
- "type": "string"
- }
- ]
- },
- "DeduplicationMiddleware": {
- "additionalProperties": false,
- "description": "Deduplication middleware configuration.\n\nPrevents duplicate messages from being processed using a sled, in-memory, MongoDB, or SQL backend.\nMessages are identified by their deduplication key and removed after the TTL expires.",
- "properties": {
- "key": {
- "default": null,
- "description": "Dedup key template, e.g. `${payload:order_id}`. Defaults to `message_id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "replay_response": {
- "description": "Answer a duplicate request with the reply its first delivery produced. Off by default.",
- "type": "boolean"
- },
- "sled_path": {
- "default": null,
- "description": "Local Sled directory (legacy). Prefer `store`.",
- "type": [
- "string",
- "null"
- ]
- },
- "store": {
- "default": null,
- "description": "Store URL: `sled:///path` or `memory://[name]` (local), `mongodb://host/db[/collection]`, or `postgres|mysql|mariadb|sqlite://…[/table]` (shared).",
- "type": [
- "string",
- "null"
- ]
- },
- "ttl_seconds": {
- "description": "Time-to-live for deduplication entries in seconds.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "ttl_seconds"
- ],
- "type": "object"
- },
- "DelayMiddleware": {
- "additionalProperties": false,
- "description": "Delay middleware configuration.\n\nIntroduces a fixed delay before processing each message.\nUseful for rate limiting, testing, or allowing time for dependent systems to become ready.",
- "properties": {
- "delay_ms": {
- "description": "Delay duration in milliseconds.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "delay_ms"
- ],
- "type": "object"
- },
- "DeliveryGuarantee": {
- "description": "What a route guarantees about its sink effect, as inferred from its configuration.\n\nOrdered weakest to strongest, so a requirement is met by any guarantee `>=` it.",
- "oneOf": [
- {
- "const": "at_most_once",
- "description": "The source forgets a message once it hands it over; a crash can lose it.",
- "type": "string"
- },
- {
- "const": "at_least_once",
- "description": "Acked only after the sink accepted it; a crash or retry can repeat the write.",
- "type": "string"
- },
- {
- "const": "effectively_once",
- "description": "At-least-once delivery absorbed by an idempotent sink write: the effect lands once.",
- "type": "string"
- }
- ]
- },
- "DetailedMappingRule": {
- "additionalProperties": false,
- "description": "Full mapping form with a fallback value and/or a presence requirement.",
- "properties": {
- "default": {
- "description": "Value used when the source path is absent."
- },
- "path": {
- "description": "Source path in the input document (e.g. `$.user.id`, `user.id`, `$.items[0]`).",
- "type": "string"
- },
- "required": {
- "default": false,
- "description": "Reject the message when the source path is absent and no `default` is set.",
- "type": "boolean"
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "DirSpoolConfig": {
- "additionalProperties": false,
- "description": "Configuration for a `dir_spool` endpoint: a crash-safe FIFO queue backed by a directory.\n\nAs a **sink** each message becomes a pair of files under `path` — a payload file holding\nthe raw `CanonicalMessage.payload` bytes, and an optional JSON sidecar holding its\nmetadata. Both are written to a `.tmp` name, fsynced, and renamed into place, so a reader\nnever observes a partial write. As a **source** the directory is listed in lexical order,\neach finalized pair is emitted as one message, and (with `drain_on_read`) the pair is\ndeleted once the message is acknowledged.\n\nThe names are what makes the queue ordered, so `naming_pattern` must keep the sequence\nnumber zero-padded and leading. The publisher resumes the sequence from the highest\nnumber already present in the directory, so a restart appends rather than overwrites.",
- "properties": {
- "atomic": {
- "default": true,
- "description": "(Sink only) Write to a `.tmp` name and rename into place, so a reader never sees a\npartial chunk. Defaults to true; turning it off trades that guarantee for one less\nrename per chunk.",
- "type": "boolean"
- },
- "claim": {
- "$ref": "#/$defs/SpoolClaim",
- "default": "exclusive",
- "description": "How this endpoint claims its side of the spool against a second instance in the\nsame role. Defaults to `exclusive`. See [`SpoolClaim`]."
- },
- "consumer_file": {
- "default": "CONSUMER",
- "description": "Name of the file holding the consumer lock, which keeps a second *draining*\nconsumer out. Defaults to `CONSUMER`. Same rules as `producer_file`.",
- "type": "string"
- },
- "done_file": {
- "default": "DONE",
- "description": "Name of the producer-completion sentinel file. Defaults to `DONE`.\n\nProduction can span several producers — they run one at a time, which\n`producer_file` enforces — so \"a producer closed\" is not \"production finished\".\nThis file is what says the latter, and only the last producer should write it.",
- "type": "string"
- },
- "drain_on_read": {
- "default": true,
- "description": "(Source only) Delete each chunk's files once its message is acknowledged. Defaults to\ntrue — this is what makes the directory a queue rather than a growing archive. With\nit off, chunks are left in place. Acknowledged chunks are emitted at most once per\nconsumer run; nacked chunks are redelivered.",
- "type": "boolean"
- },
- "emit_done": {
- "$ref": "#/$defs/SpoolDone",
- "default": "never",
- "description": "(Sink only) When to create `done_file`, marking production finished for a\n`stop_on_done` consumer. Defaults to `never`. See [`SpoolDone`].\n\nSet it on the *last* producer only. A publisher opening the spool deletes an\nexisting sentinel, since it is producing again."
- },
- "fsync": {
- "$ref": "#/$defs/SpoolFsync",
- "default": "chunk",
- "description": "How hard to work at making a chunk survive a power loss. Defaults to `chunk`, which\nis two fsyncs per message with a sidecar. See [`SpoolFsync`]."
- },
- "metadata_extension": {
- "default": "json",
- "description": "Extension of the JSON metadata sidecar, with or without the leading dot. Defaults to\n`json`. Set to an empty string to write and expect payload files only.",
- "type": "string"
- },
- "naming_pattern": {
- "default": "{seq:09}",
- "description": "(Sink only) Name template for each chunk, without extension. Supports `{seq}`,\n`{seq:06}` / `{seq:06d}` (zero-padded), `{timestamp}` (unix millis) and\n`{message_id}`. Defaults to `{seq:09}`. Lexical order must match sequence order,\nso keep a zero-padded `{seq}` first.",
- "type": "string"
- },
- "path": {
- "description": "Directory holding the spool. Created if missing.",
- "type": "string"
- },
- "payload_extension": {
- "default": "bin",
- "description": "Extension of the payload file, with or without the leading dot. Defaults to `bin`.",
- "type": "string"
- },
- "poll_interval_ms": {
- "default": 100,
- "description": "(Source only) Idle poll interval in milliseconds when the directory holds no new\nchunks. Defaults to 100.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "producer_file": {
- "default": "PRODUCER",
- "description": "Name of the file holding the producer lock, which keeps a second producer out.\nDefaults to `PRODUCER`.\n\nEvery instance sharing the directory must agree on this name — that is what makes\nthem exclude each other — and it must not collide with the other control files or\nwith a chunk name. See [`SpoolClaim`].",
- "type": "string"
- },
- "shard_depth": {
- "default": 0,
- "description": "How many levels of shard subdirectory to spread chunks over. Defaults to 0, which\nwrites every chunk straight into `path`.\n\nA flat spool is one directory per stream, and most filesystems degrade long before\nmq-bridge does: 30fps with sidecars is over 200,000 files an hour. Sharding takes the\nleading digits of the sequence number as directory names, so `{seq:09}` with a depth\nof 2 and a width of 3 writes chunk 1 as `000/000/001.bin` and gives every directory\nat most 1000 entries.\n\nBoth ends must agree: a consumer only descends as far as its own `shard_depth`. It\nwarns when a scan finds subdirectories it is not configured to enter, since the\nalternative is a spool that reads as permanently empty.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "shard_width": {
- "default": 3,
- "description": "How many characters of the sequence number each shard level consumes. Defaults to 3,\ngiving 1000 entries per level. Ignored when `shard_depth` is 0.\n\nWith sharding on, `naming_pattern` must start with a zero-padded `{seq:0N}` wider\nthan `shard_depth * shard_width`, so that every chunk shards identically and one\ncharacter is left for the file itself.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "source_metadata": {
- "default": false,
- "description": "(Source only) Include `mqb.src.spool_*` source positions in each message's metadata.\nDefaults to false.",
- "type": "boolean"
- },
- "stop_on_done": {
- "default": false,
- "description": "(Source only) End the stream once the directory holds no unread chunks *and*\n`done_file` is present. Defaults to false, which tails the directory indefinitely.\n\nBoth halves matter: a producer that finished long ago still has its backlog drained\nfirst, and a spool that is merely empty keeps the stream open, because another\nproducer may still be coming.",
- "type": "boolean"
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "ElasticsearchConfig": {
- "additionalProperties": false,
- "description": "The `elasticsearch` endpoint: documents into one index through `_bulk`.",
- "properties": {
- "api_key": {
- "description": "API key sent with every request.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "auth": {
- "description": "`oauth2` or `aws_sigv4` credentials, as on `http_bulk`; JSON in a URI."
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Compression of request bodies."
- },
- "id_field": {
- "description": "Top-level payload field that becomes the document `_id`. Defaults to `id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "index": {
- "description": "Index the documents are written to.",
- "type": "string",
- "x-mqb-uri": "path"
- },
- "mode": {
- "$ref": "#/$defs/ElasticsearchMode",
- "default": "index",
- "description": "`index` replaces a document; `update` merges the payload into it."
- },
- "operation": {
- "description": "Template for a message's operation, e.g. `${metadata:postgres.operation}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Address of the server, e.g. `elasticsearch://localhost:9200`.",
- "type": "string",
- "x-mqb-uri": "origin"
- }
- },
- "required": [
- "url",
- "index"
- ],
- "title": "Elasticsearch",
- "type": "object"
- },
- "ElasticsearchMode": {
- "description": "How `elasticsearch` writes a document.",
- "oneOf": [
- {
- "const": "index",
- "description": "Replaces the stored document.",
- "type": "string"
- },
- {
- "const": "update",
- "description": "Merges the payload into the stored document, creating it if missing.",
- "type": "string"
- }
- ]
- },
- "EncryptionConfig": {
- "additionalProperties": false,
- "description": "AEAD encryption settings, shared by the `encryption` middleware (per-message\npayload encryption) and the at-rest `encryption` field of the file and\nobject_store endpoints. Requires the `encryption` feature.",
- "properties": {
- "authenticate_metadata": {
- "description": "Metadata keys bound into the AEAD tag; changing one then fails decryption. Middleware only.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "cipher": {
- "$ref": "#/$defs/CipherKind",
- "default": "xchacha20poly1305",
- "description": "AEAD cipher. Defaults to `xchacha20poly1305`."
- },
- "decrypt_keys": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Extra `key_id -> base64 key` entries accepted when decrypting (key rotation).",
- "format": "password",
- "type": "object"
- },
- "key": {
- "description": "Base64-encoded 32-byte key. Supports `${env:VAR}` to read it from the environment.",
- "format": "password",
- "type": "string"
- },
- "key_id": {
- "default": "default",
- "description": "Key identifier written into each envelope; selects the key when decrypting. Defaults to `default`.",
- "type": "string"
- },
- "on_error": {
- "$ref": "#/$defs/InputErrorPolicy",
- "description": "What an input does with a message that will not decrypt. Defaults to `drop`. Middleware only."
- }
- },
- "required": [
- "key"
- ],
- "type": "object"
- },
- "Endpoint": {
- "description": "Represents a connection point for messages, which can be a source (input) or a sink (output).",
- "oneOf": [
- {
- "properties": {
- "aws": {
- "$ref": "#/$defs/AwsConfig"
- }
- },
- "required": [
- "aws"
- ],
- "type": "object"
- },
- {
- "properties": {
- "kafka": {
- "$ref": "#/$defs/KafkaConfig"
- }
- },
- "required": [
- "kafka"
- ],
- "type": "object"
- },
- {
- "properties": {
- "nats": {
- "$ref": "#/$defs/NatsConfig"
- }
- },
- "required": [
- "nats"
- ],
- "type": "object"
- },
- {
- "properties": {
- "file": {
- "$ref": "#/$defs/FileConfig"
- }
- },
- "required": [
- "file"
- ],
- "type": "object"
- },
- {
- "properties": {
- "dir_spool": {
- "$ref": "#/$defs/DirSpoolConfig"
- }
- },
- "required": [
- "dir_spool"
- ],
- "type": "object"
- },
- {
- "properties": {
- "object_store": {
- "$ref": "#/$defs/ObjectStoreConfig"
- }
- },
- "required": [
- "object_store"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "static": {
- "$ref": "#/$defs/StaticConfig"
- }
- },
- "required": [
- "static"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "ref": {
- "type": "string"
- }
- },
- "required": [
- "ref"
- ],
- "type": "object"
- },
- {
- "properties": {
- "memory": {
- "$ref": "#/$defs/MemoryConfig"
- }
- },
- "required": [
- "memory"
- ],
- "type": "object"
- },
- {
- "properties": {
- "sled": {
- "$ref": "#/$defs/SledConfig"
- }
- },
- "required": [
- "sled"
- ],
- "type": "object"
- },
- {
- "properties": {
- "amqp": {
- "$ref": "#/$defs/AmqpConfig"
- }
- },
- "required": [
- "amqp"
- ],
- "type": "object"
- },
- {
- "properties": {
- "mongodb": {
- "$ref": "#/$defs/MongoDbConfig"
- }
- },
- "required": [
- "mongodb"
- ],
- "type": "object"
- },
- {
- "properties": {
- "mqtt": {
- "$ref": "#/$defs/MqttConfig"
- }
- },
- "required": [
- "mqtt"
- ],
- "type": "object"
- },
- {
- "properties": {
- "http": {
- "$ref": "#/$defs/HttpConfig"
- }
- },
- "required": [
- "http"
- ],
- "type": "object"
- },
- {
- "properties": {
- "websocket": {
- "$ref": "#/$defs/WebSocketConfig"
- }
- },
- "required": [
- "websocket"
- ],
- "type": "object"
- },
- {
- "properties": {
- "ibmmq": {
- "$ref": "#/$defs/IbmMqConfig"
- }
- },
- "required": [
- "ibmmq"
- ],
- "type": "object"
- },
- {
- "properties": {
- "zeromq": {
- "$ref": "#/$defs/ZeroMqConfig"
- }
- },
- "required": [
- "zeromq"
- ],
- "type": "object"
- },
- {
- "properties": {
- "redis_streams": {
- "$ref": "#/$defs/RedisStreamsConfig"
- }
- },
- "required": [
- "redis_streams"
- ],
- "type": "object"
- },
- {
- "properties": {
- "grpc": {
- "$ref": "#/$defs/GrpcConfig"
- }
- },
- "required": [
- "grpc"
- ],
- "type": "object"
- },
- {
- "properties": {
- "sqlx": {
- "$ref": "#/$defs/SqlxConfig"
- }
- },
- "required": [
- "sqlx"
- ],
- "type": "object"
- },
- {
- "properties": {
- "clickhouse": {
- "$ref": "#/$defs/ClickHouseConfig"
- }
- },
- "required": [
- "clickhouse"
- ],
- "type": "object"
- },
- {
- "properties": {
- "http_bulk": {
- "$ref": "#/$defs/HttpBulkConfig"
- }
- },
- "required": [
- "http_bulk"
- ],
- "type": "object"
- },
- {
- "properties": {
- "postgres_cdc": {
- "$ref": "#/$defs/PostgresCdcConfig"
- }
- },
- "required": [
- "postgres_cdc"
- ],
- "type": "object"
- },
- {
- "properties": {
- "elasticsearch": {
- "$ref": "#/$defs/ElasticsearchConfig"
- }
- },
- "required": [
- "elasticsearch"
- ],
- "type": "object"
- },
- {
- "properties": {
- "meilisearch": {
- "$ref": "#/$defs/MeilisearchConfig"
- }
- },
- "required": [
- "meilisearch"
- ],
- "type": "object"
- },
- {
- "properties": {
- "openai_chat": {
- "$ref": "#/$defs/OpenAiChatConfig"
- }
- },
- "required": [
- "openai_chat"
- ],
- "type": "object"
- },
- {
- "properties": {
- "openai_embeddings": {
- "$ref": "#/$defs/OpenAiEmbeddingsConfig"
- }
- },
- "required": [
- "openai_embeddings"
- ],
- "type": "object"
- },
- {
- "properties": {
- "qdrant": {
- "$ref": "#/$defs/QdrantConfig"
- }
- },
- "required": [
- "qdrant"
- ],
- "type": "object"
- },
- {
- "properties": {
- "typesense": {
- "$ref": "#/$defs/TypesenseConfig"
- }
- },
- "required": [
- "typesense"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "fanout": {
- "items": {
- "$ref": "#/$defs/Endpoint"
- },
- "type": "array"
- }
- },
- "required": [
- "fanout"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "sequence": {
- "$ref": "#/$defs/SequenceConfig"
- }
- },
- "required": [
- "sequence"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "stream_buffer": {
- "$ref": "#/$defs/StreamBufferConfig"
- }
- },
- "required": [
- "stream_buffer"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "switch": {
- "$ref": "#/$defs/SwitchConfig"
- }
- },
- "required": [
- "switch"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "response": {
- "$ref": "#/$defs/ResponseConfig"
- }
- },
- "required": [
- "response"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "reader": {
- "$ref": "#/$defs/Endpoint"
- }
- },
- "required": [
- "reader"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "request": {
- "$ref": "#/$defs/RequestForwardConfig"
- }
- },
- "required": [
- "request"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "custom": {
- "properties": {
- "config": true,
- "name": {
- "type": "string"
- }
- },
- "required": [
- "name",
- "config"
- ],
- "type": "object"
- }
- },
- "required": [
- "custom"
- ],
- "type": "object"
- },
- {
- "format": "structural_endpoint",
- "properties": {
- "null": {
- "type": "null"
- }
- },
- "required": [
- "null"
- ],
- "type": "object"
- }
- ],
- "properties": {
- "middlewares": {
- "default": [],
- "description": "(Optional) A list of middlewares to apply to the endpoint.",
- "items": {
- "$ref": "#/$defs/Middleware"
- },
- "type": "array"
- }
- },
- "type": "object",
- "unevaluatedProperties": false
- },
- "FaultMode": {
- "description": "Fault injection modes for testing error handling and recovery mechanisms.",
- "oneOf": [
- {
- "const": "panic",
- "description": "Trigger a thread panic.",
- "type": "string"
- },
- {
- "const": "disconnect",
- "description": "Simulate a connection/network error (retryable).",
- "type": "string"
- },
- {
- "const": "timeout",
- "description": "Simulate a timeout error (retryable).",
- "type": "string"
- },
- {
- "const": "json_format_error",
- "description": "Simulate a JSON format error (non-retryable).",
- "type": "string"
- },
- {
- "const": "nack",
- "description": "Return a negative acknowledgement (for handlers).",
- "type": "string"
- }
- ]
- },
- "FileConfig": {
- "anyOf": [
- {
- "oneOf": [
- {
- "description": "**Queue Mode**: Standard point-to-point consumption. Reads from the start\nof the file. If `delete` is true, processed lines are physically removed\nfrom the file once they are successfully acknowledged. Removing rewrites the\nfile, so `delete` is only safe when nothing outside this process appends to it.",
- "properties": {
- "delete": {
- "default": false,
- "description": "Remove acknowledged lines from the file. Only safe if no other process writes to it.",
- "type": "boolean"
- },
- "mode": {
- "const": "consume",
- "type": "string"
- }
- },
- "required": [
- "mode"
- ],
- "type": "object"
- },
- {
- "description": "**Broadcast Mode**: Pub-sub style consumption. Tails the file by starting\nat the current end. If `delete` is true, lines are removed only after\nall local application subscribers for this specific file have acknowledged them.",
- "properties": {
- "delete": {
- "default": false,
- "description": "Remove lines all local subscribers acknowledged. Only safe if no other process writes to it.",
- "type": "boolean"
- },
- "mode": {
- "const": "subscribe",
- "type": "string"
- }
- },
- "required": [
- "mode"
- ],
- "type": "object"
- },
- {
- "description": "**Persistent Mode**: Consumption with external offset tracking.\nSaves the last read byte position to a `.offset` file identified by the `group_id`.\nThis allows the consumer to resume exactly where it left off after a restart\nwithout deleting data or requiring the bridge to stay running.",
- "properties": {
- "group_id": {
- "description": "The consumer group ID that is used for offset tracking. Should be unique.",
- "type": "string"
- },
- "mode": {
- "const": "group_subscribe",
- "type": "string"
- },
- "read_from_tail": {
- "default": false,
- "description": "If true, starts reading from the end of the file if no offset is stored.\nIf false, starts reading from the beginning.",
- "type": "boolean"
- }
- },
- "required": [
- "mode",
- "group_id"
- ],
- "type": "object"
- }
- ]
- },
- {}
- ],
- "properties": {
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Per-batch compression (`none`, `gzip`, `lz4`, `zstd`). Requires the `compression` feature. Publishers: always. Consumers: must match, and only the default `consume` mode reads it."
- },
- "csv": {
- "$ref": "#/$defs/CsvConfig",
- "description": "CSV dialect (separator, quote, header, columns); only read with `format: csv`."
- },
- "delimiter": {
- "description": "Optional delimiter for messages. Defaults to newline (\"\\n\").\nCan be a string or a hex sequence (e.g. \"0x00\").\nCurrently only single-byte delimiters are supported.",
- "type": [
- "string",
- "null"
- ]
- },
- "encryption": {
- "anyOf": [
- {
- "$ref": "#/$defs/EncryptionConfig"
- },
- {
- "type": "null"
- }
- ],
- "default": null,
- "description": "At-rest AEAD encryption applied after compression. Requires the `encryption` feature. Publishers: always. Consumers: must match, and only the default `consume` mode reads it."
- },
- "format": {
- "$ref": "#/$defs/FileFormat",
- "default": "normal",
- "description": "The format for writing messages to the file (Publisher) or interpreting them (Consumer). Defaults to `normal`."
- },
- "fsync": {
- "$ref": "#/$defs/FileFsync",
- "description": "(Sink only) When appended data is fsynced: `off` (default), `batch` (before each ack) or `periodic`."
- },
- "fsync_interval_ms": {
- "description": "(Sink only) Interval for `fsync: periodic` in milliseconds. Defaults to 1000.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "idempotency": {
- "description": "Deprecated: use `name_by`. true = `source_position`, false = `write_time`; ignored when `name_by` is set.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "name_by": {
- "$ref": "#/$defs/NameBy",
- "default": "auto",
- "description": "(Sink only) `write_time` (default here, appends to `path`) or `source_position` (replay-safe part files, `path` is their directory)."
- },
- "path": {
- "description": "Path to the file, or to the directory holding the part files under `source_position` naming.",
- "type": "string"
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only) Include authoritative `mqb.src.file_*` source positions; only `consume` mode reproduces them across restarts. Defaults to false.",
- "type": "boolean"
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "FileFormat": {
- "description": "Format for messages written to or read from a file.",
- "oneOf": [
- {
- "const": "normal",
- "description": "The full `CanonicalMessage` is serialized to JSON. Payload is either base64 or utf8 text. Also `envelope`.",
- "type": "string"
- },
- {
- "const": "json",
- "description": "The full `CanonicalMessage` is serialized to JSON. Payload is rendered as a JSON value if possible. Also `envelope_json`; for plain JSON lines use `raw`.",
- "type": "string"
- },
- {
- "const": "text",
- "description": "The full `CanonicalMessage` is serialized to JSON. Payload is rendered as a string if possible. Also `envelope_text`.",
- "type": "string"
- },
- {
- "const": "raw",
- "description": "Only the payload is written, one per line: plain JSON lines for JSON payloads. For consumers, the line is read as raw bytes. Also `payload`.",
- "type": "string"
- },
- {
- "const": "csv",
- "description": "CSV rows mapped to/from JSON objects (string values only). The first row is the header/schema.",
- "type": "string"
- },
- {
- "const": "parquet",
- "description": "Parquet, one object per batch, JSON-object rows (object_store only; `parquet` feature).",
- "type": "string"
- },
- {
- "const": "envelope",
- "description": "Same as `normal`.",
- "type": "string",
- "x-alias-of": "normal"
- },
- {
- "const": "envelope_json",
- "description": "Same as `json`.",
- "type": "string",
- "x-alias-of": "json"
- },
- {
- "const": "envelope_text",
- "description": "Same as `text`.",
- "type": "string",
- "x-alias-of": "text"
- },
- {
- "const": "payload",
- "description": "Same as `raw`.",
- "type": "string",
- "x-alias-of": "raw"
- }
- ]
- },
- "FileFsync": {
- "description": "When the appending `file` sink forces written data to disk.",
- "oneOf": [
- {
- "const": "off",
- "description": "Never; the OS flushes. An acknowledged batch can be lost on power loss.",
- "type": "string"
- },
- {
- "const": "batch",
- "description": "Before a batch is acknowledged.",
- "type": "string"
- },
- {
- "const": "periodic",
- "description": "In the background every `fsync_interval_ms`; at most that window is at risk.",
- "type": "string"
- }
- ]
- },
- "FilterMiddleware": {
- "description": "The `filter` middleware. Accepts either a bare expression string or a map with `expression` and an optional `on_error`.",
- "oneOf": [
- {
- "description": "Expression over payload fields and `meta.<key>`, e.g. `amount > 100`.",
- "type": "string"
- },
- {
- "additionalProperties": false,
- "properties": {
- "expression": {
- "description": "Expression over payload fields and `meta.<key>`, e.g. `amount > 100`.",
- "type": "string"
- },
- "on_error": {
- "default": "fail",
- "description": "What to do with a message the expression cannot evaluate.",
- "enum": [
- "fail",
- "drop"
- ],
- "type": "string"
- }
- },
- "required": [
- "expression"
- ],
- "type": "object"
- }
- ]
- },
- "GrpcConfig": {
- "additionalProperties": false,
- "properties": {
- "api_key": {
- "description": "API key sent as `api_key_name` (default `x-api-key`) on dynamic calls and the\nreflection RPC.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "api_key_name": {
- "description": "Metadata key used for `api_key`. Defaults to `x-api-key`.",
- "type": [
- "string",
- "null"
- ]
- },
- "bearer_token": {
- "description": "Bearer token sent as `authorization` on dynamic calls and the reflection RPC;\nrejected in Bridge/server mode.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "binary_metadata": {
- "additionalProperties": {
- "items": {
- "format": "uint8",
- "maximum": 255,
- "minimum": 0,
- "type": "integer"
- },
- "type": "array"
- },
- "description": "Static binary metadata for dynamic calls and the reflection RPC. Keys must end in\n`-bin`; each value is raw bytes, written as a JSON array of byte values 0-255 (the\nonly accepted form — a base64 or text string is rejected). As a URL parameter:\n`?binary_metadata=%7B%22x-trace-bin%22%3A%5B1%2C2%2C3%5D%7D`, which is\n`{\"x-trace-bin\": [1, 2, 3]}` percent-encoded.",
- "type": "object"
- },
- "concurrency_limit_per_connection": {
- "default": null,
- "description": "Maximum number of concurrent requests handled per connection. **Server-mode only.**",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "connect_timeout_ms": {
- "default": null,
- "description": "Maximum time to establish a client connection.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "consumer_id": {
- "default": null,
- "description": "Stable subscription identity used for ACK tracking and redelivery. Defaults to a\nfresh id per consumer; set it to be redelivered unacknowledged messages on reconnect.",
- "type": [
- "string",
- "null"
- ]
- },
- "descriptor_set_bytes": {
- "description": "Compiled protobuf FileDescriptorSet bytes for embedded callers. Takes precedence\nover `descriptor_set_path` and avoids writing a temporary descriptor file.",
- "items": {
- "format": "uint8",
- "maximum": 255,
- "minimum": 0,
- "type": "integer"
- },
- "type": [
- "array",
- "null"
- ]
- },
- "descriptor_set_path": {
- "default": null,
- "description": "Compiled protobuf FileDescriptorSet for dynamic calls or a served method.",
- "type": [
- "string",
- "null"
- ]
- },
- "http2_keepalive_interval_ms": {
- "default": null,
- "description": "HTTP/2 keepalive ping interval in milliseconds. Applies in both modes. Default disabled",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "http2_keepalive_timeout_ms": {
- "default": null,
- "description": "Timeout for a keepalive ping acknowledgement in milliseconds. Applies in both modes.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "idle_stream_timeout_ms": {
- "default": null,
- "description": "Maximum time a dynamic response stream may remain idle between messages.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "initial_connection_window_size": {
- "default": null,
- "description": "HTTP/2 connection-level initial window size in bytes. Applies in both modes.",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "initial_stream_window_size": {
- "default": null,
- "description": "HTTP/2 stream-level initial window size in bytes. Applies in both modes.",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_decoding_message_size": {
- "default": null,
- "description": "Maximum size of a decoded incoming message in bytes. Applies in both modes. Default 4 MiB.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_encoding_message_size": {
- "default": null,
- "description": "Maximum size of an encoded outgoing message in bytes. Default unlimited.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "metadata": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Static ASCII metadata attached to dynamic RPCs and to the reflection RPC that\nfetches their descriptors. Values for keys that look sensitive are extracted by\nmq-bridge's normal secret handling.",
- "type": "object"
- },
- "method_name": {
- "default": null,
- "description": "RPC method name for dynamic calls or a served unary method.",
- "type": [
- "string",
- "null"
- ]
- },
- "overall_timeout_ms": {
- "default": null,
- "description": "Maximum lifetime of a dynamic RPC; exceeding it stops the route instead of reconnecting.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "reflection": {
- "default": false,
- "description": "Discover descriptors from the remote gRPC server reflection v1 service.",
- "type": "boolean"
- },
- "request": {
- "default": null,
- "description": "JSON request mapped to the dynamic protobuf input message."
- },
- "request_timeout_ms": {
- "default": null,
- "description": "Maximum time to establish an RPC and receive its initial response.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "server_mode": {
- "default": false,
- "description": "If `true`, start an embedded tonic gRPC server that accepts `Publish` / `PublishBatch`\nRPCs, plus one descriptor-defined unary method when `method_name` is set. If `false`\n(the default), connect to a remote server as a client.",
- "type": "boolean"
- },
- "server_streaming": {
- "default": false,
- "description": "Deprecated compatibility hint. Dynamic RPC shape is always derived from the descriptor.",
- "type": "boolean"
- },
- "service_name": {
- "default": null,
- "description": "Fully-qualified protobuf service name for dynamic calls or a served method.",
- "type": [
- "string",
- "null"
- ]
- },
- "shared": {
- "default": true,
- "description": "(Publisher only) Share one gRPC channel per connection (default: true); false forces a dedicated channel.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "timeout_ms": {
- "description": "Deprecated compatibility timeout in milliseconds. Used as the fallback for\nconnection and initial-request deadlines. Prefer the dedicated settings.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "topic": {
- "description": "Topic / subject used for both subscribe and publish paths.",
- "type": [
- "string",
- "null"
- ]
- },
- "url": {
- "description": "The gRPC server URL (e.g., \"http://localhost:50051\" for client or \"0.0.0.0:50051\" for server mode).",
- "format": "password",
- "type": "string"
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "HttpBulkAuth": {
- "additionalProperties": false,
- "description": "How `http_bulk` requests are authenticated beyond `headers`. Set at most one field.",
- "properties": {
- "aws_sigv4": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkAwsSigV4"
- },
- {
- "type": "null"
- }
- ],
- "description": "AWS Signature Version 4 on every request. Needs the `aws` feature."
- },
- "oauth2": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkOAuth2"
- },
- {
- "type": "null"
- }
- ],
- "description": "OAuth2 client credentials: a bearer token is fetched, cached and refreshed."
- }
- },
- "type": "object"
- },
- "HttpBulkAwsSigV4": {
- "additionalProperties": false,
- "description": "AWS Signature Version 4 signing for `http_bulk`.",
- "properties": {
- "access_key": {
- "description": "Access key id. Unset = the default AWS credential chain.",
- "type": [
- "string",
- "null"
- ]
- },
- "region": {
- "description": "AWS region, e.g. `eu-west-1`.",
- "type": "string"
- },
- "secret_key": {
- "description": "Secret access key.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "service": {
- "description": "Service name to sign for, e.g. `es` or `aoss`.",
- "type": "string"
- },
- "session_token": {
- "description": "Session token of temporary credentials.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "region",
- "service"
- ],
- "type": "object"
- },
- "HttpBulkConfig": {
- "additionalProperties": false,
- "description": "HTTP APIs that take or return many JSON documents in one request, such as search engines.\n\nAs an output, a batch of upserts is sent as one NDJSON or JSON-array body and deletes as a\nlist of ids; `result` describes how the target reports success. As an input, `read` pages\nthrough a JSON listing by a cursor. With `query` it answers a batch of lookups in one\nrequest. The endpoint knows no product by name.",
- "properties": {
- "auth": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkAuth"
- },
- {
- "type": "null"
- }
- ],
- "description": "Credentials that are fetched or computed per request: an OAuth2 token or AWS SigV4."
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "(Publisher only) Compression of request bodies (`none`, `gzip`, `zstd`, `lz4`)."
- },
- "connect_timeout_ms": {
- "description": "Connection timeout in milliseconds. Defaults to 10000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "delete": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkDelete"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Publisher only) The request that removes documents. Without it a delete message fails."
- },
- "delete_values": {
- "default": [
- "delete",
- "d"
- ],
- "description": "(Publisher only) Operation values that mean delete. Defaults to `delete` and `d`.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "headers": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Headers sent with every request, e.g. `Authorization`. Treated as secrets.",
- "format": "password",
- "type": "object"
- },
- "max_request_bytes": {
- "default": 10485760,
- "description": "(Publisher only) Largest request body in bytes; a bigger batch is split. Defaults to 10 MiB.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "operation": {
- "description": "(Publisher only) Template for a message's operation, e.g. `${metadata:postgres.operation}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "query": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkQuery"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Publisher only) The request that answers lookups. Excludes `upsert` and `delete`."
- },
- "read": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkRead"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Consumer only) The request that reads a page of documents. Required for an input."
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration for `https://` connections."
- },
- "upsert": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkUpsert"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Publisher only) The request that writes documents. An output needs it or `query`."
- },
- "url": {
- "description": "Base URL of the target, e.g. `http://localhost:7700`.",
- "type": "string"
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "HttpBulkCursor": {
- "additionalProperties": false,
- "description": "Where an `http_bulk` input takes its next read position from. Set at most one of\n`response` and `item`; with neither the position is the number of documents read.",
- "properties": {
- "end_when_missing": {
- "description": "With `response`: a response without the field ends the read, instead of stopping the route.",
- "type": "boolean"
- },
- "item": {
- "description": "JSON pointer to the position in the last document of a page, e.g. `/id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "link": {
- "description": "The next page is the `Link` header's `rel=\"next\"` URL. Without one the read ends.",
- "type": "boolean"
- },
- "next_url": {
- "description": "JSON pointer to the URL of the next page, e.g. `/links/next`. Missing or `null` ends the read.",
- "type": [
- "string",
- "null"
- ]
- },
- "response": {
- "description": "JSON pointer to the next position in the response, e.g. `/last_seq`. `null` ends the read.",
- "type": [
- "string",
- "null"
- ]
- },
- "start": {
- "description": "Position before the first read. Defaults to `0` for a count, otherwise to `null`."
- }
- },
- "type": "object"
- },
- "HttpBulkDelete": {
- "additionalProperties": false,
- "description": "The `http_bulk` request that removes documents by id.",
- "properties": {
- "envelope": {
- "description": "The body around the id array, e.g. `{\"points\": {ids}}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "id_field": {
- "default": "id",
- "description": "Top-level payload field holding the document id. Defaults to `id`.",
- "type": "string"
- },
- "line": {
- "description": "Template per id, e.g. `{\"delete\":{\"_id\":{id}}}`: a body line, an array entry or a `{ids}` entry.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_ids": {
- "default": 1000,
- "description": "Most ids in one request. Defaults to 1000.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "method": {
- "description": "HTTP method. Defaults to `POST`.",
- "type": [
- "string",
- "null"
- ]
- },
- "path": {
- "description": "Path and query. With `{ids}` the ids go there, comma-separated; otherwise as a JSON array body.",
- "type": "string"
- },
- "result": {
- "$ref": "#/$defs/HttpBulkResult",
- "default": {},
- "description": "How the target reports the outcome. Unset = the HTTP status alone decides."
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "HttpBulkFormat": {
- "description": "Body shape of an `http_bulk` upsert request.",
- "oneOf": [
- {
- "const": "ndjson",
- "description": "One JSON document per line.",
- "type": "string"
- },
- {
- "const": "json_array",
- "description": "One JSON array holding every document.",
- "type": "string"
- }
- ]
- },
- "HttpBulkItemSuccess": {
- "additionalProperties": false,
- "description": "The value an `items` entry carries when its document was written, e.g. SES's `/Status: SUCCESS`.",
- "properties": {
- "equals": {
- "description": "The status of a written document, e.g. `SUCCESS` or `true`."
- },
- "path": {
- "description": "JSON pointer, inside an entry, to its status, e.g. `/Status`.",
- "type": "string"
- }
- },
- "required": [
- "path",
- "equals"
- ],
- "type": "object"
- },
- "HttpBulkItems": {
- "additionalProperties": false,
- "description": "A JSON response holding an array with one entry per document, in request order.",
- "properties": {
- "error": {
- "description": "JSON pointer to an entry's error text, e.g. `/index/error/reason`. An entry that has it failed.",
- "type": "string"
- },
- "index": {
- "description": "JSON pointer, inside an entry, to its document's position; the array may then list only failures, or be absent.",
- "type": [
- "string",
- "null"
- ]
- },
- "path": {
- "default": "",
- "description": "JSON pointer to the array, e.g. `/items`. Empty = the response itself is the array.",
- "type": "string"
- },
- "success": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkItemSuccess"
- },
- {
- "type": "null"
- }
- ],
- "description": "An entry whose value at `path` is not `equals` failed, even without an error text."
- }
- },
- "required": [
- "error"
- ],
- "type": "object"
- },
- "HttpBulkJob": {
- "additionalProperties": false,
- "description": "A job the target runs after answering, polled until it ends.",
- "properties": {
- "error": {
- "description": "JSON pointer to the error text of a failed job, e.g. `/error/message`.",
- "type": [
- "string",
- "null"
- ]
- },
- "failed": {
- "description": "States that mean the job failed. Any other state is polled again.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "id": {
- "description": "JSON pointer to the job id in the response, e.g. `/taskUid`.",
- "type": "string"
- },
- "poll": {
- "description": "Path polled with GET; `{id}` is replaced by the job id, e.g. `/tasks/{id}`.",
- "type": "string"
- },
- "status": {
- "description": "JSON pointer to the job state in the poll response, e.g. `/status`.",
- "type": "string"
- },
- "succeeded": {
- "description": "States that mean the job succeeded.",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "timeout_ms": {
- "default": 300000,
- "description": "How long the job may run before its documents fail as retryable.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "id",
- "poll",
- "status",
- "succeeded",
- "failed"
- ],
- "type": "object"
- },
- "HttpBulkLines": {
- "additionalProperties": false,
- "description": "A response with one JSON line per document, in request order.",
- "properties": {
- "error": {
- "description": "JSON pointer to the error text of a rejected document, e.g. `/error`.",
- "type": [
- "string",
- "null"
- ]
- },
- "success": {
- "description": "JSON pointer to the boolean that is true for a written document, e.g. `/success`.",
- "type": "string"
- }
- },
- "required": [
- "success"
- ],
- "type": "object"
- },
- "HttpBulkOAuth2": {
- "additionalProperties": false,
- "description": "OAuth2 client credentials grant for `http_bulk`.",
- "properties": {
- "client_id": {
- "description": "Client id.",
- "type": "string"
- },
- "client_secret": {
- "description": "Client secret.",
- "format": "password",
- "type": "string"
- },
- "scope": {
- "description": "Scopes to ask for, separated by spaces.",
- "type": [
- "string",
- "null"
- ]
- },
- "token_url": {
- "description": "Token endpoint, e.g. `https://login.example.com/oauth2/token`.",
- "type": "string"
- }
- },
- "required": [
- "token_url",
- "client_id",
- "client_secret"
- ],
- "type": "object"
- },
- "HttpBulkQuery": {
- "additionalProperties": false,
- "description": "The `http_bulk` request that answers a batch of lookups, one answer per message in order.",
- "properties": {
- "concurrency": {
- "description": "Requests in flight for one batch. Defaults to 8 with `per_message`, else 1.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "content_type": {
- "description": "`Content-Type` header. Defaults to `application/x-ndjson` or `application/json`.",
- "type": [
- "string",
- "null"
- ]
- },
- "envelope": {
- "description": "`json_array` only: the body around the array, e.g. `{\"docs\": {requests}}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "error": {
- "description": "JSON pointer in an answer to its error text. An answer that has it fails its message.",
- "type": [
- "string",
- "null"
- ]
- },
- "format": {
- "$ref": "#/$defs/HttpBulkFormat",
- "default": "ndjson",
- "description": "Body shape: `ndjson` (default) or `json_array`."
- },
- "found": {
- "description": "JSON pointer in an answer to a boolean; `false` or missing means not found.",
- "type": [
- "string",
- "null"
- ]
- },
- "index": {
- "description": "JSON pointer, inside an entry, to the `{index}` of its message; entries are matched by it, not by position.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_messages": {
- "description": "Most messages in one request; a larger batch is split. Unset = `max_request_bytes` alone.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "method": {
- "description": "HTTP method. Defaults to `POST`.",
- "type": [
- "string",
- "null"
- ]
- },
- "parse_value": {
- "description": "The value is JSON text: parse it. Text that is not JSON fails its message.",
- "type": "boolean"
- },
- "path": {
- "description": "Path and query appended to `url`, e.g. `/books/_mget`.",
- "type": "string"
- },
- "per_message": {
- "description": "One request per message; the response is the answer. Excludes `envelope` and `responses`.",
- "type": "boolean"
- },
- "request": {
- "description": "Template of one message's part of the body, e.g. `{\"_id\": \"${payload:id}\"}`. `{index}` is its number in the request.",
- "type": "string"
- },
- "responses": {
- "default": "",
- "description": "JSON pointer to the array of answers, e.g. `/responses`. Empty = the response is the array.",
- "type": "string"
- },
- "responses_text": {
- "description": "JSON pointer to JSON text that holds the answers; `responses` then points into that text.",
- "type": [
- "string",
- "null"
- ]
- },
- "split_on_reject": {
- "description": "A refused request (400, 409, 413, 422) or unmatched answers: ask again in halves.",
- "type": "boolean"
- },
- "stop_reason": {
- "description": "JSON pointer to why the response ended; named when its answers cannot be matched.",
- "type": [
- "string",
- "null"
- ]
- },
- "value": {
- "description": "JSON pointer in an answer to the value returned, e.g. `/_source`. Unset = the whole answer.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "path",
- "request"
- ],
- "type": "object"
- },
- "HttpBulkRead": {
- "additionalProperties": false,
- "description": "The `http_bulk` request that reads one page of documents.",
- "properties": {
- "body": {
- "description": "JSON request body; `{cursor}` and `{limit}` are replaced by JSON values.",
- "type": [
- "string",
- "null"
- ]
- },
- "checkpoint_store": {
- "description": "Where the read position is saved, e.g. `file:///var/lib/mqb/cursors.json`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "cursor": {
- "$ref": "#/$defs/HttpBulkCursor",
- "default": {},
- "description": "Where the next read position comes from. Unset = the number of documents read so far."
- },
- "cursor_id": {
- "description": "Cursor id that keys the saved read position. Unset: every start reads all.",
- "type": [
- "string",
- "null"
- ]
- },
- "items": {
- "default": "",
- "description": "JSON pointer to the array of documents, e.g. `/results`. Empty = the response is the array.",
- "type": "string"
- },
- "max_polling_interval_ms": {
- "description": "If set, the wait doubles after each empty page up to this value.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "method": {
- "description": "HTTP method. Defaults to `GET`, or `POST` with a `body`.",
- "type": [
- "string",
- "null"
- ]
- },
- "path": {
- "description": "Path and query. `{cursor}` is the read position and `{limit}` the batch size.",
- "type": "string"
- },
- "polling_interval_ms": {
- "description": "Wait in milliseconds after an empty page. Defaults to 1000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "stream": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkStream"
- },
- {
- "type": "null"
- }
- ],
- "description": "Read one open response as `sse` events or `ndjson` lines instead of pages."
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "HttpBulkResult": {
- "additionalProperties": false,
- "description": "How an `http_bulk` target reports what happened to a request. Set at most one field.",
- "properties": {
- "items": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkItems"
- },
- {
- "type": "null"
- }
- ],
- "description": "The response is one JSON document holding an array with an entry per document."
- },
- "job": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkJob"
- },
- {
- "type": "null"
- }
- ],
- "description": "The response names a job that finishes later and is polled."
- },
- "lines": {
- "anyOf": [
- {
- "$ref": "#/$defs/HttpBulkLines"
- },
- {
- "type": "null"
- }
- ],
- "description": "The response has one JSON line per document."
- }
- },
- "type": "object"
- },
- "HttpBulkStream": {
- "description": "How an `http_bulk` input that reads one open response is cut into messages.",
- "oneOf": [
- {
- "const": "sse",
- "description": "Server-sent events; the event id is the read position.",
- "type": "string"
- },
- {
- "const": "ndjson",
- "description": "One JSON document per line.",
- "type": "string"
- }
- ]
- },
- "HttpBulkUpsert": {
- "additionalProperties": false,
- "description": "The `http_bulk` request that writes documents.",
- "properties": {
- "action": {
- "description": "`ndjson` only: a line sent before each document, e.g. `{\"index\":{\"_id\":\"${payload:id}\"}}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "content_type": {
- "description": "`Content-Type` header. Defaults to `application/x-ndjson` or `application/json`.",
- "type": [
- "string",
- "null"
- ]
- },
- "document": {
- "description": "Template around each document, e.g. `{\"doc\": {document}, \"doc_as_upsert\": true}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "envelope": {
- "description": "`json_array` only: the body around the array, e.g. `{\"docs\": {documents}}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "format": {
- "$ref": "#/$defs/HttpBulkFormat",
- "default": "ndjson",
- "description": "Body shape: `ndjson` (default) or `json_array`."
- },
- "max_documents": {
- "description": "Most documents in one request; a larger batch is split. Unset = `max_request_bytes` alone.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "method": {
- "description": "HTTP method. Defaults to `POST`.",
- "type": [
- "string",
- "null"
- ]
- },
- "path": {
- "description": "Path and query appended to `url`, e.g. `/indexes/books/documents`.",
- "type": "string"
- },
- "result": {
- "$ref": "#/$defs/HttpBulkResult",
- "default": {},
- "description": "How the target reports the outcome. Unset = the HTTP status alone decides."
- },
- "split_on_reject": {
- "description": "Without `result`: send a request refused with 400, 409, 413 or 422 again in halves, so only the bad document fails.",
- "type": "boolean"
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "HttpConfig": {
- "additionalProperties": false,
- "description": "General HTTP connection configuration.",
- "properties": {
- "basic_auth": {
- "description": "HTTP Basic Authentication credentials (username, password). For consumers: validates incoming requests. For publishers: adds Authorization header.",
- "format": "password",
- "maxItems": 2,
- "minItems": 2,
- "prefixItems": [
- {
- "type": "string"
- },
- {
- "type": "string"
- }
- ],
- "type": [
- "array",
- "null"
- ]
- },
- "batch_concurrency": {
- "description": "(Publisher only) The number of concurrent HTTP requests to send in a batch. Defaults to 20.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "(Publisher only) Codec for the request body (`none`, `gzip`, `lz4`, `zstd`); overrides\n`compression_enabled`. `lz4` is non-standard (mq-bridge peers only). Ignored on a consumer —\nenable response compression with `compression_enabled`. Defaults to `none`."
- },
- "compression_enabled": {
- "description": "Turns compression on. Publisher: compress the request body with gzip (unless `compression`\nsets another codec). Consumer: compress responses, negotiating the best codec the client's\n`Accept-Encoding` accepts. Defaults to off.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "compression_threshold_bytes": {
- "default": null,
- "description": "Minimum message size in bytes to compress. Messages smaller than this are sent uncompressed. Defaults to 1024 bytes.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "concurrency_limit": {
- "description": "(Consumer only) Maximum number of concurrent requests to handle. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "custom_headers": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Custom headers as key-value pairs (e.g., {\"X-API-Key\": \"token123\"}). Added to outgoing HTTP headers for both consumers and publishers.",
- "format": "password",
- "type": "object"
- },
- "fire_and_forget": {
- "default": false,
- "description": "(Consumer only) If true, respond immediately with 202 Accepted without waiting for downstream processing. Defaults to false.",
- "type": "boolean"
- },
- "header_read_timeout_ms": {
- "description": "(Consumer only) Close an HTTP/1 connection whose request head takes longer than this. Unset or 0 = no limit.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "inline_response_fast_path": {
- "default": true,
- "description": "(Consumer only) If true, compatible `http -> response` routes may bypass the normal route consumer/worker/disposition pipeline\nand reply inline for lower latency. Defaults to true. Set to false to force the normal route path.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "internal_buffer_size": {
- "description": "(Consumer only) Internal buffer size for the channel. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_body_bytes": {
- "description": "(Consumer only) Largest request body in bytes, also after decompression; a larger one gets 413. Defaults to 256 MiB.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "message_id_header": {
- "description": "(Consumer only) Header key to extract the message ID from. Defaults to \"message-id\".",
- "type": [
- "string",
- "null"
- ]
- },
- "method": {
- "description": "(Optional) HTTP method. For publishers: the method to use (defaults to POST). For consumers: restrict to this method (others return 405).",
- "type": [
- "string",
- "null"
- ]
- },
- "pass_through_status": {
- "default": false,
- "description": "(Publisher) Treat every HTTP response status as response data instead of classifying\nnon-2xx statuses as publisher errors. Transport and response-read failures remain errors.\n(Consumer) When every output sink opts in, transient sink failures return 502 without\nstopping a non-streaming request/reply route. Defaults to false.",
- "type": "boolean"
- },
- "path": {
- "description": "(Consumer only) Optional request path filter. If set, only requests whose URI path matches exactly are delivered to this consumer.",
- "type": [
- "string",
- "null"
- ]
- },
- "pool_idle_timeout_ms": {
- "description": "(Publisher only) Timeout for idle connections in the connection pool in milliseconds. Defaults to 90000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "receive_streamable": {
- "default": false,
- "description": "(Consumer only) If true, read request bodies as a stream and emit each received stream item as a separate message.",
- "type": "boolean"
- },
- "request_timeout_ms": {
- "description": "Timeout for HTTP requests in milliseconds. For consumers, it's the request-reply timeout. For publishers, it's the timeout for each individual request. Defaults to 30000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "server_protocol": {
- "$ref": "#/$defs/HttpServerProtocol",
- "default": "auto",
- "description": "(Consumer only) Restrict which HTTP protocol versions a server listener accepts.\nDefaults to `auto` (HTTP/1.1 + HTTP/2). On cleartext listeners, `http2_only`\nmeans HTTP/2 prior-knowledge (h2c) only."
- },
- "shared": {
- "default": true,
- "description": "(Publisher only) Share one HTTP client per connection (default: true); false forces a dedicated client.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "stream_response_to": {
- "anyOf": [
- {
- "$ref": "#/$defs/Endpoint"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Publisher only) Optional endpoint that receives streamed HTTP response items as correlated messages.\n\nUse a `stream_buffer` endpoint here when callers need to read streamed\nresponse items later through a normal mq-bridge consumer. Each streamed\nitem is published with `correlation_id`, `http_stream_id`,\n`http_stream_index`, `http_stream_format`, and `http_stream_end`\nmetadata. A stream that breaks after its first item ends with an item\nmarked `http_stream_error` and is not retried. If the request message\nhas no `correlation_id`, the HTTP publisher uses\n`format!(\"{:032x}\", request.message_id)` so callers can derive the\nconsumer correlation id before calling `send`."
- },
- "tcp_keepalive_ms": {
- "description": "(Publisher only) TCP keepalive timeout for the underlying connection pool in milliseconds. Defaults to 60000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "url": {
- "description": "For consumers, the listen address (e.g., \"0.0.0.0:8080\"). For publishers, the target URL.",
- "type": "string"
- },
- "workers": {
- "description": "(Consumer only) Number of worker threads to use. Defaults to 0 for unlimited.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "HttpServerProtocol": {
- "description": "Supported inbound HTTP protocols for server listeners.",
- "oneOf": [
- {
- "const": "auto",
- "description": "Accept both HTTP/1.1 and HTTP/2, matching the current default behavior.",
- "type": "string"
- },
- {
- "const": "http1_only",
- "description": "Accept only HTTP/1.x connections.",
- "type": "string"
- },
- {
- "const": "http2_only",
- "description": "Accept only HTTP/2 connections.",
- "type": "string"
- }
- ]
- },
- "IbmMqConfig": {
- "additionalProperties": false,
- "description": "Connection settings for the IBM MQ Queue Manager.",
- "properties": {
- "channel": {
- "description": "Required. Server Connection (SVRCONN) Channel name defined on the QM.",
- "type": "string"
- },
- "disable_status_inq": {
- "default": false,
- "description": "If false, attempt to open the queue with INQUIRE permissions to fetch queue depth for status checks. Defaults to false.",
- "type": "boolean"
- },
- "internal_buffer_size": {
- "default": null,
- "description": "Internal buffer size for the channel. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_message_size": {
- "default": 4194304,
- "description": "Maximum message size in bytes (default: 4MB). Optional.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "password": {
- "description": "Password for authentication. Optional; required if the channel enforces authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "queue": {
- "description": "Target Queue name for point-to-point messaging. Optional if `topic` is set; defaults to route name if omitted.",
- "type": [
- "string",
- "null"
- ]
- },
- "queue_manager": {
- "description": "Required. Name of the Queue Manager to connect to (e.g., `QM1`).",
- "type": "string"
- },
- "tls": {
- "$ref": "#/$defs/IbmTlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "cert_file": null,
- "cert_password": null,
- "cipher_spec": null,
- "required": false
- },
- "description": "TLS configuration settings (e.g., keystore paths). Optional."
- },
- "topic": {
- "description": "Target Topic string for Publish/Subscribe. If set, enables **Subscriber mode** (Consumer) or publishes to a topic (Publisher). Optional if `queue` is set.",
- "type": [
- "string",
- "null"
- ]
- },
- "url": {
- "description": "Required. Connection URL in `host(port)` format. Supports comma-separated list for failover (e.g., `host1(1414),host2(1414)`). If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Username for authentication. Optional; required if the channel enforces authentication",
- "type": [
- "string",
- "null"
- ]
- },
- "wait_timeout_ms": {
- "default": 1000,
- "description": "(Consumer only) Polling timeout in milliseconds (default: 1000ms). Optional.",
- "format": "int32",
- "type": "integer"
- }
- },
- "required": [
- "url",
- "queue_manager",
- "channel"
- ],
- "type": "object"
- },
- "IbmTlsConfig": {
- "additionalProperties": false,
- "description": "TLS configuration for the IBM MQ native client.\n\nThe IBM MQ client doesn't consume PEM files, so this uses MQ-native field\nnames rather than the generic [`TlsConfig`] used by the other endpoints.",
- "properties": {
- "accept_invalid_certs": {
- "default": false,
- "description": "If true, disable server certificate verification (insecure).",
- "type": "boolean"
- },
- "cert_file": {
- "description": "For IBM MQ this is the CMS key repository stem (e.g. `/path/to/tls` for `tls.kdb`/`tls.sth`),\nnot a PEM file. Exposed as `cert_file` for config parity with the generic `TlsConfig`;\nthe MQ-native name `key_repository` is still accepted.",
- "type": [
- "string",
- "null"
- ]
- },
- "cert_password": {
- "description": "Password unlocking the key repository. Requires an IBM MQ client/server at 9.3.0.0+.\nExposed as `cert_password` for parity with `TlsConfig`; alias `key_repository_password`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "cipher_spec": {
- "description": "TLS CipherSpec (e.g., `ANY_TLS12`). Required for encrypted connections. IBM MQ-specific.",
- "type": [
- "string",
- "null"
- ]
- },
- "key_repository": {
- "description": "MQ-native alias for `cert_file`: the CMS key repository stem (e.g. `/path/to/tls` for `tls.kdb`/`tls.sth`).",
- "type": [
- "string",
- "null"
- ]
- },
- "key_repository_password": {
- "description": "MQ-native alias for `cert_password`: password unlocking the key repository. Requires an IBM MQ client/server at 9.3.0.0+.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "required": {
- "default": false,
- "description": "If true, enable TLS/SSL.",
- "type": "boolean"
- }
- },
- "type": "object"
- },
- "IdHash": {
- "description": "Hash that folds a key into a deterministic 128-bit `message_id`.",
- "oneOf": [
- {
- "const": "fnv1a",
- "description": "FNV-1a, 128-bit. Default: the ids earlier releases produced, so existing dedup state keeps matching.",
- "type": "string"
- },
- {
- "const": "sha256",
- "description": "First 128 bits of SHA-256. Collision-resistant, but every id changes, so dedup state from `fnv1a` no longer matches.",
- "type": "string"
- }
- ]
- },
- "InputErrorPolicy": {
- "description": "What an input middleware does with a message it cannot decode.",
- "oneOf": [
- {
- "const": "fail",
- "description": "Stop the route; the message stays uncommitted.",
- "type": "string"
- },
- {
- "const": "drop",
- "description": "Log and acknowledge the message, then carry on with the rest.",
- "type": "string"
- }
- ]
- },
- "KafkaConfig": {
- "additionalProperties": false,
- "description": "General Kafka connection configuration.",
- "properties": {
- "consumer_options": {
- "default": null,
- "description": "(Consumer only) Additional librdkafka consumer configuration options (key-value pairs).",
- "items": {
- "maxItems": 2,
- "minItems": 2,
- "prefixItems": [
- {
- "type": "string"
- },
- {
- "type": "string"
- }
- ],
- "type": "array"
- },
- "type": [
- "array",
- "null"
- ]
- },
- "create_topic": {
- "default": true,
- "description": "(Publisher only) Create the topic at start if it is missing (default: true).",
- "type": [
- "boolean",
- "null"
- ]
- },
- "delayed_ack": {
- "default": false,
- "description": "(Publisher only) If true, do not wait for an acknowledgement when sending to broker. Defaults to false.",
- "type": "boolean"
- },
- "group_id": {
- "description": "(Consumer only) Consumer group ID.\nIf not provided, the consumer acts in **Subscriber mode**: it generates a unique, ephemeral group ID and starts consuming from the latest offset.",
- "type": [
- "string",
- "null"
- ]
- },
- "partition_key": {
- "default": null,
- "description": "(Publisher only) Name of a metadata field whose value is used as the Kafka record\nkey (drives partitioning/ordering). Unset, or absent on a given message, falls back\nto the message id. Default unset.",
- "type": [
- "string",
- "null"
- ]
- },
- "partitions": {
- "default": 6,
- "description": "(Publisher only) Partition count used when auto-creating the topic (default: 6).\nHigher values raise write/consume parallelism; ordering is only guaranteed per\npartition key (message_id), not across the whole topic. Ignored if the topic exists.",
- "format": "int32",
- "minimum": 1,
- "type": [
- "integer",
- "null"
- ]
- },
- "password": {
- "description": "Optional password for SASL authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "producer_options": {
- "default": null,
- "description": "(Publisher only) Additional librdkafka producer configuration options (key-value pairs).",
- "items": {
- "maxItems": 2,
- "minItems": 2,
- "prefixItems": [
- {
- "type": "string"
- },
- {
- "type": "string"
- }
- ],
- "type": "array"
- },
- "type": [
- "array",
- "null"
- ]
- },
- "shared": {
- "default": true,
- "description": "(Publisher only) Share one producer per connection (default: true); false gives a dedicated producer.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only) Include authoritative `mqb.src.kafka_*` source positions. Defaults to false.",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "topic": {
- "default": null,
- "description": "The topic to produce to or consume from. A source also takes a list (array or comma-separated) or a `^regex`.",
- "items": {
- "type": "string"
- },
- "type": [
- "string",
- "array",
- "null"
- ]
- },
- "url": {
- "description": "Comma-separated list of Kafka broker URLs. If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username for SASL authentication.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "LimiterMiddleware": {
- "additionalProperties": false,
- "description": "Throughput limiter middleware configuration.\n\nApplies a best-effort pacing delay so an endpoint does not exceed the configured\nmessage rate. For batch operations the limiter accounts for the number of messages\nin the batch, not just the batch count.",
- "properties": {
- "messages_per_second": {
- "description": "Target throughput in messages per second. Must be greater than zero.",
- "format": "double",
- "type": "number"
- }
- },
- "required": [
- "messages_per_second"
- ],
- "type": "object"
- },
- "LookupEntry": {
- "additionalProperties": false,
- "description": "One lookup of a `lookup` middleware.",
- "properties": {
- "from": {
- "$ref": "#/$defs/Endpoint",
- "description": "The endpoint to ask; it must answer with a response."
- },
- "into": {
- "description": "Dotted payload path the response is written to, e.g. `customer.profile`.",
- "type": "string"
- },
- "metadata": {
- "additionalProperties": {
- "type": "string"
- },
- "default": {},
- "description": "Request metadata templates, e.g. `http_path: \"/users/${payload:user_id}\"`.",
- "type": "object"
- },
- "payload": {
- "description": "Request payload template. Defaults to the message's own payload and metadata.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "from",
- "into"
- ],
- "type": "object"
- },
- "LookupMiddleware": {
- "additionalProperties": false,
- "description": "Lookup (enrich) middleware configuration.\n\nAsks a request-capable endpoint (HTTP, NATS/Memory `request_reply`, MongoDB `find`,\nSQLx/ClickHouse `lookup_query`) about each message and writes the response into the\npayload. A query keyed with `IN (…)` / `$in` answers a whole batch at once; other\nendpoints get one request per message. Several `entries` run in parallel.",
- "properties": {
- "concurrency": {
- "default": 16,
- "description": "Requests in flight per entry when the endpoint cannot answer a batch at once.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "entries": {
- "description": "Further lookups, run in parallel with `from`.",
- "items": {
- "$ref": "#/$defs/LookupEntry"
- },
- "type": "array"
- },
- "from": {
- "anyOf": [
- {
- "$ref": "#/$defs/Endpoint"
- },
- {
- "type": "null"
- }
- ],
- "description": "The endpoint to ask; it must answer with a response. Use `entries` for several."
- },
- "into": {
- "description": "Dotted payload path the response is written to, e.g. `customer.profile`.",
- "type": [
- "string",
- "null"
- ]
- },
- "metadata": {
- "additionalProperties": {
- "type": "string"
- },
- "default": {},
- "description": "Request metadata templates, e.g. `http_path: \"/users/${payload:user_id}\"`.",
- "type": "object"
- },
- "payload": {
- "description": "Request payload template. Defaults to the message's own payload and metadata.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "MappingRule": {
- "anyOf": [
- {
- "description": "Shorthand: just the source path.",
- "type": "string"
- },
- {
- "$ref": "#/$defs/DetailedMappingRule",
- "description": "Full form with a fallback value and/or a presence requirement."
- }
- ],
- "description": "How one output field is produced from the input document.\n\nEither a bare path string (`\"$.first_name\"`) or an object with a `path` plus an\noptional `default` and `required` flag."
- },
- "MeilisearchConfig": {
- "additionalProperties": false,
- "description": "The `meilisearch` endpoint: documents into one index, or a read of it.",
- "properties": {
- "api_key": {
- "description": "API key sent with every request.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "checkpoint_store": {
- "description": "(Consumer only) Where the read position is saved, e.g. `file:///var/lib/mqb/cursors.json`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Compression of request bodies."
- },
- "connect_timeout_ms": {
- "description": "Connection timeout in milliseconds. Defaults to 10000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "cursor_id": {
- "description": "(Consumer only) Cursor id that keys the saved read position.",
- "type": [
- "string",
- "null"
- ]
- },
- "delete_values": {
- "description": "Operation values that mean delete. Defaults to `delete` and `d`.",
- "items": {
- "type": "string"
- },
- "type": [
- "array",
- "null"
- ]
- },
- "fields": {
- "description": "(Consumer only) Comma-separated document fields to read. Default: all.",
- "type": [
- "string",
- "null"
- ]
- },
- "index": {
- "description": "Index UID. Defaults to the route name.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_polling_interval_ms": {
- "description": "(Consumer only) If set, the wait doubles after each empty page up to this value.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_request_bytes": {
- "description": "Largest request body in bytes; a bigger batch is split. Defaults to 90 MB.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "operation": {
- "description": "Template for a message's operation, e.g. `${metadata:postgres.operation}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "polling_interval_ms": {
- "description": "(Consumer only) Wait in milliseconds after an empty page. Defaults to 1000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "primary_key": {
- "description": "The document field Meilisearch keys documents by.",
- "type": [
- "string",
- "null"
- ]
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Address of the server, e.g. `meilisearch://localhost:7700`.",
- "type": "string",
- "x-mqb-uri": "url"
- }
- },
- "required": [
- "url"
- ],
- "title": "Meilisearch",
- "type": "object",
- "x-mqb-default-batch-size": 50000
- },
- "MemoryConfig": {
- "additionalProperties": false,
- "anyOf": [
- {
- "required": [
- "topic"
- ]
- },
- {
- "required": [
- "url"
- ]
- }
- ],
- "properties": {
- "capacity": {
- "description": "The capacity of the channel. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "enable_nack": {
- "default": false,
- "description": "(Consumer only) If true, enables NACK support (re-queuing), which requires cloning messages.\nDefaults to false for memory:// transports, automatically true for IPC transports (ipc://, unix://, pipe://).",
- "type": "boolean"
- },
- "request_reply": {
- "default": false,
- "description": "(Publisher only) If true, send() waits for a response.",
- "type": "boolean"
- },
- "request_timeout_ms": {
- "description": "(Publisher only) Timeout for request-reply operations in milliseconds. Defaults to 30000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "subscribe_mode": {
- "default": false,
- "description": "(Consumer only) If true, act as a **Subscriber** (fan-out). Defaults to false (queue).",
- "type": "boolean"
- },
- "topic": {
- "description": "The topic name or transport URL. Can be:\n- Simple name: \"my-topic\" (defaults to memory://my-topic)\n- Memory URL: \"memory://my-topic\"\n- IPC URL: \"ipc://my-queue\" or \"ipc:///path/to/socket\"\n- Unix socket: \"unix:///path/to/socket\" (Unix only)\n- Named pipe: \"pipe://my-pipe\" (Windows only)\n\nEither `topic` or `url` can be specified (they are serde aliases).",
- "minLength": 1,
- "type": "string"
- },
- "url": {
- "description": "Alias for `topic`. Use either `topic` or `url`.",
- "minLength": 1,
- "type": "string"
- }
- },
- "type": "object"
- },
- "MetricsMiddleware": {
- "additionalProperties": false,
- "description": "Metrics middleware configuration.\n\nEnables collection and reporting of message processing metrics such as throughput,\nlatency, and error rates. The presence of this middleware in the configuration\nenables metrics collection for the endpoint.\n\nMetrics are typically exported via Prometheus or similar monitoring systems.",
- "type": "object"
- },
- "Middleware": {
- "description": "An enumeration of all supported middleware types.",
- "oneOf": [
- {
- "additionalProperties": false,
- "description": "Renders a template into the `mqb.id` metadata key, e.g. `id: \"${payload:order_id}\"`.\nConsumer-only; list it *after* anything that reads `mqb.id`, since the last consumer\nmiddleware in the list runs first.",
- "properties": {
- "id": {
- "type": "string"
- }
- },
- "required": [
- "id"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "deduplication": {
- "$ref": "#/$defs/DeduplicationMiddleware"
- }
- },
- "required": [
- "deduplication"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "metrics": {
- "$ref": "#/$defs/MetricsMiddleware"
- }
- },
- "required": [
- "metrics"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Per-message OpenTelemetry spans, continuing the `traceparent` metadata. Input and output.",
- "properties": {
- "otel": {
- "$ref": "#/$defs/OtelMiddleware"
- }
- },
- "required": [
- "otel"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Asks other endpoints per message and writes their responses into the payload. Input and output.",
- "properties": {
- "lookup": {
- "$ref": "#/$defs/LookupMiddleware"
- }
- },
- "required": [
- "lookup"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Keeps a running state per key (sums, counters, moving averages) and writes it into the payload. Input and output.",
- "properties": {
- "aggregate": {
- "$ref": "#/$defs/AggregateMiddleware"
- }
- },
- "required": [
- "aggregate"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "dlq": {
- "$ref": "#/$defs/DeadLetterQueueMiddleware"
- }
- },
- "required": [
- "dlq"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "retry": {
- "$ref": "#/$defs/RetryMiddleware"
- }
- },
- "required": [
- "retry"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "random_panic": {
- "$ref": "#/$defs/RandomPanicMiddleware"
- }
- },
- "required": [
- "random_panic"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "delay": {
- "$ref": "#/$defs/DelayMiddleware"
- }
- },
- "required": [
- "delay"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Bounds each send; one that does not finish in time fails as retryable. Output-only.",
- "properties": {
- "timeout": {
- "$ref": "#/$defs/TimeoutMiddleware"
- }
- },
- "required": [
- "timeout"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "weak_join": {
- "$ref": "#/$defs/WeakJoinMiddleware"
- }
- },
- "required": [
- "weak_join"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "limiter": {
- "$ref": "#/$defs/LimiterMiddleware"
- }
- },
- "required": [
- "limiter"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "buffer": {
- "$ref": "#/$defs/BufferMiddleware"
- }
- },
- "required": [
- "buffer"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "cookie_jar": {
- "$ref": "#/$defs/CookieJarMiddleware"
- }
- },
- "required": [
- "cookie_jar"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "transform": {
- "$ref": "#/$defs/TransformMiddleware"
- }
- },
- "required": [
- "transform"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "encryption": {
- "$ref": "#/$defs/EncryptionConfig"
- }
- },
- "required": [
- "encryption"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "compression": {
- "$ref": "#/$defs/CompressionMiddleware"
- }
- },
- "required": [
- "compression"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Combines a publish batch into one physical transport message. Output-only;\npair it with `unpack` on the reading route's input.",
- "properties": {
- "pack": {
- "$ref": "#/$defs/PackMiddleware"
- }
- },
- "required": [
- "pack"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Splits a packed physical message back into its logical messages. Input-only.",
- "properties": {
- "unpack": {
- "$ref": "#/$defs/UnpackMiddleware"
- }
- },
- "required": [
- "unpack"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "description": "Keeps only messages matching an expression, e.g. `filter: \"amount > 100\"`.\nReads payload fields by name and metadata as `meta.<key>`. Input and output.\nAs a map, `on_error: drop` drops a message the expression cannot evaluate.",
- "properties": {
- "filter": {
- "$ref": "#/$defs/FilterMiddleware"
- }
- },
- "required": [
- "filter"
- ],
- "type": "object"
- },
- {
- "additionalProperties": false,
- "properties": {
- "custom": {
- "properties": {
- "config": true,
- "name": {
- "type": "string"
- }
- },
- "required": [
- "name",
- "config"
- ],
- "type": "object"
- }
- },
- "required": [
- "custom"
- ],
- "type": "object"
- }
- ]
- },
- "MongoConsume": {
- "description": "How a MongoDB endpoint consumes a collection. One intent-named selector — the bridge picks the\nunderlying mechanism (change stream vs. polling) automatically. Defaults to `capture_all`.",
- "oneOf": [
- {
- "const": "delete",
- "description": "**Queue** — competing consumers: claim, process, delete, so each document goes to exactly one\nreader. Destructive and intended for jobs, not bulk reads. Also accepted as `consumer`.",
- "type": "string"
- },
- {
- "const": "snapshot",
- "description": "**One-shot read** — page documents by `_id`, then end the route. Non-destructive and\nnon-resumable; needs no replica set and reads arbitrary collections. This is not a\npoint-in-time snapshot: separate page queries can observe concurrent inserts and deletes,\nwhile inserts below the current `_id` high-water mark can be missed.",
- "type": "string"
- },
- {
- "const": "capture_new",
- "description": "**Watch existing collection** — capture changes from now on (insert/update/delete), resuming\nunder `cursor_id`. Reads an existing collection non-destructively; never ends on drain.",
- "type": "string"
- },
- {
- "const": "capture_all",
- "description": "**Watch existing collection** — read the existing documents first, then capture changes.\nNon-destructive and the fastest read mode; use this for bulk reads and ETL. Default.\nNeeds a replica set; on a standalone `mongod` use `snapshot` or `delete`.",
- "type": "string"
- }
- ]
- },
- "MongoDbConfig": {
- "additionalProperties": false,
- "description": "General MongoDB connection configuration.",
- "properties": {
- "capped_size_bytes": {
- "description": "(Publisher only) If set, creates a capped collection with this size in bytes.",
- "format": "int64",
- "type": [
- "integer",
- "null"
- ]
- },
- "change_stream": {
- "default": false,
- "description": "(Consumer only) **Deprecated** — use `consume: capture_new`. Kept for compatibility.",
- "type": "boolean"
- },
- "checkpoint_store": {
- "description": "(Consumer only) Where to persist the resume cursor in `capture_new`/`capture_all` mode. A URL\nselects the backend; a bare name (or `/name`) reuses the **source** database with that name:\n- absent → source database, collection `mqb_cursors_<source_collection>` (auto-unique)\n- `/my_cursors` → source database, collection `my_cursors`\n- `file:///var/lib/mqb/cursors.json` → local JSON file (read-only / write-restricted sources)\n- `mongodb://host/db/collection` → external MongoDB collection (collection optional)\n- `postgres://user@host/db/table` or `mysql://host/db/table` → external SQL table (table optional)\n- `s3://bucket/prefix` (also `gs://`, `az://`, `abfs://`) → cloud object store; creds via env\n\nWhen no collection/table is named, it defaults to `mqb_cursors_<source_collection>`.\nMay embed connection credentials, so it is treated as a secret.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "collection": {
- "description": "The MongoDB collection name.",
- "type": [
- "string",
- "null"
- ]
- },
- "consume": {
- "anyOf": [
- {
- "$ref": "#/$defs/MongoConsume"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Consumer only) How to consume the collection: `capture_all` (default, read existing\ndocuments first, then watch for changes — non-destructive, for bulk reads and ETL),\n`capture_new` (watch an existing collection for changes only), `snapshot` (one-shot\nnon-destructive read that ends on drain, the option without a replica set), or\n`delete` (competing-consumers work queue — destructive and intended for jobs; also `consumer`).\nThe bridge selects the underlying mechanism automatically. If unset, the deprecated\n`change_stream` boolean is honored for backward compatibility."
- },
- "cursor_id": {
- "description": "The ID used for the cursor in sequenced mode. If not provided, consumption starts from the current sequence (ephemeral).",
- "type": [
- "string",
- "null"
- ]
- },
- "database": {
- "description": "The database name.",
- "type": "string"
- },
- "find": {
- "description": "(Publisher only) Extended-JSON filter template for `lookup`, e.g. `{\"_id\": \"${payload:id}\"}`.\nAnswers with the first match. `{\"_id\": {\"$in\": [\"${payload:id}\"]}}` answers a batch at once.",
- "type": [
- "string",
- "null"
- ]
- },
- "format": {
- "$ref": "#/$defs/MongoDbFormat",
- "default": "normal",
- "description": "Format for storing messages. Defaults to Normal."
- },
- "id_field": {
- "description": "(Publisher only) Top-level payload field whose value becomes the document `_id`, or a\nreplay-stable `${...}` template such as `${metadata:mqb.id}`. Enables idempotent inserts\nthrough MongoDB's unique `_id` index. Sink collections only.",
- "type": [
- "string",
- "null"
- ]
- },
- "meta_collection": {
- "description": "**Deprecated** — unused since the publisher no longer writes `seq`. Kept for compatibility.",
- "type": [
- "string",
- "null"
- ]
- },
- "password": {
- "description": "Optional password. Takes precedence over any credentials embedded in the `url`.\nUse embedded URL credentials for simple one-off connections but prefer explicit username/password fields (or environment-sourced secrets) for clarity and secret management in production.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "polling_interval_ms": {
- "description": "(Consumer only) Polling interval in milliseconds for the consumer (when not using Change Streams). Defaults to 100ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "receive_query": {
- "description": "(Consumer only) Optional custom MongoDB query to filter messages. Provided as a JSON string (e.g., '{\"type\": \"notification\"}').",
- "type": [
- "string",
- "null"
- ]
- },
- "reply_polling_ms": {
- "description": "(Publisher only) Polling interval in milliseconds for the publisher when waiting for a reply. Defaults to 50ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "report_outcome": {
- "default": false,
- "description": "(Publisher only) Return the message with metadata `mongodb.outcome` = `inserted`/`existed`\n(dup-key) so a `request`+`switch` can branch. Sink collections only; pair with `id_field`.",
- "type": "boolean"
- },
- "request_reply": {
- "default": false,
- "description": "(Publisher only) If true, the publisher will wait for a response in a dedicated collection. Defaults to false.",
- "type": "boolean"
- },
- "request_timeout_ms": {
- "description": "(Publisher only) Timeout for request-reply operations in milliseconds. Defaults to 30000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "shared": {
- "default": true,
- "description": "Share one MongoDB client per connection (default: true); false forces a dedicated client.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only, `capture_new`/`capture_all`) Include authoritative `mqb.src.mongodb_*`\nsource positions. Defaults to false.",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "ttl_seconds": {
- "description": "(Publisher only) TTL in seconds for documents created by the publisher. If set, a TTL index is created.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "url": {
- "description": "MongoDB connection string URI. Can contain a comma-separated list of hosts for a replica set. If it contains userinfo, it will be treated as a secret.\nCredentials provided via the separate `username` and `password` fields take precedence over any credentials embedded in the URL.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username. Takes precedence over any credentials embedded in the `url`.\nUse embedded URL credentials for simple one-off connections but prefer explicit username/password fields (or environment-sourced secrets) for clarity and secret management in production.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url",
- "database"
- ],
- "type": "object"
- },
- "MongoDbFormat": {
- "description": "MongoDB message storage format.\n\nDetermines how messages are stored and retrieved from MongoDB collections.",
- "oneOf": [
- {
- "const": "normal",
- "type": "string"
- },
- {
- "const": "json",
- "type": "string"
- },
- {
- "const": "text",
- "type": "string"
- },
- {
- "const": "raw",
- "type": "string"
- },
- {
- "const": "envelope",
- "description": "Same as `normal`.",
- "type": "string",
- "x-alias-of": "normal"
- },
- {
- "const": "envelope_json",
- "description": "Same as `json`.",
- "type": "string",
- "x-alias-of": "json"
- },
- {
- "const": "envelope_text",
- "description": "Same as `text`.",
- "type": "string",
- "x-alias-of": "text"
- },
- {
- "const": "payload",
- "description": "Same as `raw`.",
- "type": "string",
- "x-alias-of": "raw"
- }
- ]
- },
- "MqttConfig": {
- "additionalProperties": false,
- "description": "General MQTT connection configuration.",
- "properties": {
- "clean_session": {
- "default": false,
- "description": "(Consumer only) If true, start with a clean session. Defaults to false (persistent session). Setting this to true effectively enables **Subscriber mode** (ephemeral).",
- "type": "boolean"
- },
- "client_id": {
- "description": "Optional client ID. If not provided, one is generated or derived from route name.",
- "type": [
- "string",
- "null"
- ]
- },
- "delayed_ack": {
- "default": false,
- "description": "(Consumer only) If true, messages are acknowledged immediately upon receipt (auto-ack).\nIf false (default), messages are acknowledged after processing (manual-ack).\nNote: For QoS 1/2 the publisher always waits for end-to-end broker\nconfirmation (PUBACK/PUBCOMP) before reporting success, independent of\nthis setting; QoS 0 remains fire-and-forget.",
- "type": "boolean"
- },
- "keep_alive_seconds": {
- "description": "Keep-alive interval in seconds. Defaults to 20.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_inflight": {
- "description": "Maximum number of inflight messages.",
- "format": "uint16",
- "maximum": 65535,
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "password": {
- "description": "Optional password for authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "protocol": {
- "$ref": "#/$defs/MqttProtocol",
- "default": "v5",
- "description": "MQTT protocol version (V3 or V5). Defaults to V5."
- },
- "qos": {
- "description": "Quality of Service level (0, 1, or 2). Defaults to 1.",
- "format": "uint8",
- "maximum": 255,
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "queue_capacity": {
- "description": "Capacity of the internal channel for incoming messages. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "session_expiry_interval": {
- "description": "Session expiry interval in seconds (MQTT v5 only).",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "topic": {
- "description": "The MQTT topic.",
- "type": [
- "string",
- "null"
- ]
- },
- "url": {
- "description": "MQTT broker URL (e.g., \"tcp://localhost:1883\"). Does not support multiple hosts. If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username for authentication.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "MqttProtocol": {
- "description": "MQTT protocol version.\n\nSpecifies which version of the MQTT protocol to use for connections.",
- "enum": [
- "v5",
- "v3"
- ],
- "type": "string"
- },
- "NameBy": {
- "description": "How a `file` or `object_store` sink names what it writes.\n\nThe name decides everything downstream: a name derived from the source position sorts by\nsource order and repeats exactly on a replay (so a re-write is a no-op), while a name\nderived from the write time sorts by write order and is unique per write.\n\n`auto` resolves against the input for `object_store` only. There the two schemes write the\nsame thing under a different name, so the choice is free; on `file` they are different sink\nstructures — one appended file versus a directory of part files — and deriving that from the\ninput would change what `path` means without the operator asking for it.",
- "oneOf": [
- {
- "const": "auto",
- "description": "object_store: `source_position` when the input carries one, else `write_time`. file: always `write_time`.",
- "type": "string"
- },
- {
- "const": "source_position",
- "description": "`part-<source>-<start>-<end>.<ext>`. Needs a replayable input; on `file` this turns `path` into a directory.",
- "type": "string"
- },
- {
- "const": "write_time",
- "description": "object_store: `<uuidv7>.<ext>` under an optional `YYYY/MM/DD/` prefix. file: appends to `path`.",
- "type": "string"
- }
- ]
- },
- "NatsConfig": {
- "additionalProperties": false,
- "description": "General NATS connection configuration.",
- "properties": {
- "deduplicate": {
- "default": false,
- "description": "(Publisher only, JetStream) If true, publish a `Nats-Msg-Id` header (from the message id) so\nJetStream deduplicates redeliveries within the stream's duplicate window. Defaults to false.",
- "type": "boolean"
- },
- "delayed_ack": {
- "default": false,
- "description": "(Publisher only) If true, do not wait for an acknowledgement when sending to broker. Defaults to false.",
- "type": "boolean"
- },
- "deliver_policy": {
- "anyOf": [
- {
- "$ref": "#/$defs/NatsDeliverPolicy"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Consumer only) The delivery policy for the consumer. Defaults to \"all\"."
- },
- "no_jetstream": {
- "default": false,
- "description": "If no_jetstream: true, use Core NATS (fire-and-forget) instead of JetStream. Defaults to false.",
- "type": "boolean"
- },
- "password": {
- "description": "Optional password for authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "prefetch_count": {
- "description": "(Consumer only) Number of messages to prefetch from the consumer. Defaults to 10000.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "request_reply": {
- "default": false,
- "description": "(Publisher only) If true, the publisher uses the request-reply pattern.\nIt sends a request and waits for a response (using `core_client.request_with_headers()`).\nDefaults to false.",
- "type": "boolean"
- },
- "request_timeout_ms": {
- "description": "(Publisher only) Timeout for request-reply operations in milliseconds. Defaults to 30000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "shared": {
- "default": true,
- "description": "Share one NATS client per connection (default: true); false forces a dedicated connection.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only) Include authoritative `mqb.src.nats_*` source positions. Defaults to false.",
- "type": "boolean"
- },
- "stream": {
- "description": "The JetStream stream name. Required for Consumers, even with\n`no_jetstream: true` (unused there, but still validated).",
- "type": [
- "string",
- "null"
- ]
- },
- "stream_max_bytes": {
- "description": "(Publisher only) Maximum total bytes in the stream (if created by the bridge). Defaults to 1GB.",
- "format": "int64",
- "type": [
- "integer",
- "null"
- ]
- },
- "stream_max_messages": {
- "description": "(Publisher only) Maximum number of messages in the stream (if created by the bridge). Defaults to 1,000,000.",
- "format": "int64",
- "type": [
- "integer",
- "null"
- ]
- },
- "subject": {
- "description": "The NATS subject to publish to or subscribe to. If a stream is\nauto-created, it's scoped to `{stream}.>`, so prefix accordingly.",
- "type": [
- "string",
- "null"
- ]
- },
- "subscriber_mode": {
- "default": false,
- "description": "(Consumer only) If true, use ephemeral **Subscriber mode**. Defaults to false (durable consumer).",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration."
- },
- "token": {
- "description": "Optional token for authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "url": {
- "description": "Comma-separated list of NATS server URLs (e.g., \"nats://localhost:4222,nats://localhost:4223\"). If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username for authentication.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "NatsDeliverPolicy": {
- "enum": [
- "all",
- "last",
- "new",
- "last_per_subject"
- ],
- "type": "string"
- },
- "ObjectStoreConfig": {
- "additionalProperties": false,
- "description": "Configuration for a local or cloud object-store endpoint.\n\nAs a **sink**, each flushed batch is written as one immutable object under `url`,\nnamed `<prefix>/[YYYY/MM/DD/]<uuidv7>.<ext>`. As a **source**, objects under `url`\nare listed in key order, fetched, split by `delimiter`, and emitted as messages;\nprogress is persisted to `checkpoint_store` (the last processed object key) so a\nrestart resumes without re-emitting. Objects are never mutated or deleted in place.",
- "properties": {
- "checkpoint_store": {
- "description": "(Source only) Durable resume store URL recording the last processed object key, e.g.\n`file:///var/lib/mqb/obj.json`, `s3://bucket/cursors`, or `postgres://…`. Without it\nevery restart re-lists and re-emits all objects.",
- "type": [
- "string",
- "null"
- ]
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Whole-object compression (`none`, `gzip`, `lz4`, `zstd`). Requires the `compression` feature."
- },
- "csv": {
- "$ref": "#/$defs/CsvConfig",
- "description": "CSV dialect (separator, quote, header, columns); only read with `format: csv`."
- },
- "cursor_id": {
- "description": "(Source only) Cursor id namespacing the checkpoint key; enables durable resume.",
- "type": [
- "string",
- "null"
- ]
- },
- "date_partition": {
- "default": null,
- "description": "(Sink only) Prepend a `YYYY/MM/DD/` path (write time, UTC) to each object key. Applies to\n`write_time` naming only; defaults to on. Purely for readability / lifecycle rules.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "date_partition_style": {
- "$ref": "#/$defs/DatePartitionStyle",
- "default": "nested",
- "description": "(Sink only) Date folder layout: `nested` (`YYYY/MM/DD/`) or `hive` (`year=YYYY/month=MM/day=DD/`)."
- },
- "delimiter": {
- "description": "Record delimiter within an object. Defaults to newline (\"\\n\"). Can be a string or a\nhex sequence (e.g. \"0x00\").",
- "type": [
- "string",
- "null"
- ]
- },
- "encryption": {
- "anyOf": [
- {
- "$ref": "#/$defs/EncryptionConfig"
- },
- {
- "type": "null"
- }
- ],
- "default": null,
- "description": "At-rest AEAD encryption applied after compression. Requires the `encryption` feature."
- },
- "extension": {
- "description": "(Sink only) Extension for written objects, without the dot. Defaults to a value derived\nfrom `format`, `compression` and `encryption` (e.g. `jsonl`, `csv`, `bin`, `jsonl.gz`,\n`jsonl.lz4`, `jsonl.gz.enc`); encrypted objects get a trailing `.enc` since they are\nciphertext, not a directly decompressible `.gz`.",
- "type": [
- "string",
- "null"
- ]
- },
- "format": {
- "$ref": "#/$defs/FileFormat",
- "default": "normal",
- "description": "Record encoding within an object, shared with the file endpoint. Defaults to\n`normal` (one JSON `CanonicalMessage` per line). Parquet needs the `parquet` feature."
- },
- "idempotency": {
- "description": "Deprecated: use `name_by`. true = `source_position`, false = `write_time`; ignored when `name_by` is set.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "max_object_bytes": {
- "description": "(Source only) Maximum size in bytes of a single object to fetch into memory. An object\nlarger than this fails the read (surfaced as a consumer error) instead of being\nbuffered whole. Unset means no limit (the whole object is materialized).",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "name_by": {
- "$ref": "#/$defs/NameBy",
- "default": "auto",
- "description": "(Sink only) `auto`, `write_time` (uuidv7 name) or `source_position` (name carries the source range)."
- },
- "on_decode_error": {
- "$ref": "#/$defs/DecodeErrorPolicy",
- "default": "skip",
- "description": "(Source only) An object that cannot be decoded: `skip` it (default) or `fail` the route.\n`skip` also stops the route while no object has been decoded yet."
- },
- "polling_interval_ms": {
- "description": "(Source only) Idle poll interval in milliseconds when no new objects are found.\nDefaults to 1000.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "url": {
- "description": "Object-store URL, e.g. `file:///var/lib/mqb/incoming`, `s3://bucket/prefix`,\n`gs://bucket/prefix`, or `az://container/prefix`. Credentials are resolved\nfrom the environment by the `object_store` crate (same mechanism as the checkpoint\nbackend); Azure takes the account from `AZURE_STORAGE_ACCOUNT_NAME`, R2 uses `s3://`\nplus a custom `AWS_ENDPOINT_URL`.",
- "type": "string"
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "OpenAiChatConfig": {
- "additionalProperties": false,
- "description": "The `openai_chat` endpoint: one chat completion per message, or per `messages_per_request`.",
- "properties": {
- "api_key": {
- "description": "API key sent as a bearer token.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "auth": {
- "description": "`oauth2` or `aws_sigv4` credentials, as on `http_bulk`; JSON in a URI."
- },
- "concurrency": {
- "description": "Requests in flight for one batch sent to this endpoint. Defaults to 8.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "json": {
- "default": false,
- "description": "Ask for a JSON object and parse the answer.",
- "type": "boolean"
- },
- "max_request_bytes": {
- "description": "With `messages_per_request`: largest request body in bytes. Defaults to 10 MiB.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "messages_per_request": {
- "description": "Messages answered by one request, as numbered inputs. Unset = one request each.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "model": {
- "description": "Model name, e.g. `gpt-4o-mini`.",
- "type": "string"
- },
- "options": {
- "description": "Further request fields, e.g. `{\"temperature\": 0}`; JSON in a URI."
- },
- "prompt": {
- "description": "Template of the user message, e.g. `Summarise: ${payload:body}`.",
- "type": "string"
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "system": {
- "description": "Template of the system message.",
- "type": [
- "string",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Base URL of the API, e.g. `http://localhost:11434`. Defaults to `https://api.openai.com/v1`.",
- "type": [
- "string",
- "null"
- ],
- "x-mqb-uri": "url"
- }
- },
- "required": [
- "model",
- "prompt"
- ],
- "title": "OpenAI chat completion",
- "type": "object"
- },
- "OpenAiEmbeddingsConfig": {
- "additionalProperties": false,
- "description": "The `openai_embeddings` endpoint: one embeddings request per batch.",
- "properties": {
- "api_key": {
- "description": "API key sent as a bearer token.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "auth": {
- "description": "`oauth2` or `aws_sigv4` credentials, as on `http_bulk`; JSON in a URI."
- },
- "dimensions": {
- "description": "Length of the vectors, for models that can shorten them.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "input": {
- "description": "Template of the text to embed, e.g. `${payload:body}`.",
- "type": "string"
- },
- "max_request_bytes": {
- "description": "Largest request body in bytes; a bigger batch is split. Defaults to 10 MiB.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "model": {
- "description": "Model name, e.g. `text-embedding-3-small`.",
- "type": "string"
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Base URL of the API, e.g. `http://localhost:11434`. Defaults to `https://api.openai.com/v1`.",
- "type": [
- "string",
- "null"
- ],
- "x-mqb-uri": "url"
- }
- },
- "required": [
- "model",
- "input"
- ],
- "title": "OpenAI embeddings",
- "type": "object"
- },
- "OtelMiddleware": {
- "additionalProperties": false,
- "description": "OpenTelemetry middleware configuration.\n\nOpens a span per message and carries the W3C `traceparent` in metadata. Inactive unless\nthe host installed an OpenTelemetry tracer provider before the route started.",
- "type": "object"
- },
- "PackFormat": {
- "description": "Batch envelope used by the `pack` / `unpack` middlewares.",
- "oneOf": [
- {
- "const": "mqb",
- "description": "mq-bridge's own envelope: payload, metadata and `message_id` per record,\nlength-prefixed, with an optional compressed body.",
- "type": "string"
- },
- {
- "const": "benthos_binary",
- "description": "The layout Redpanda Connect's `archive: binary` writes. Payloads only —\nmetadata and ids are dropped — and no room for an inner codec.",
- "type": "string"
- }
- ]
- },
- "PackMiddleware": {
- "additionalProperties": false,
- "description": "Transport batching middleware configuration (`pack`, output side).\n\nCombines the messages of one publish batch into a single physical message, so a\nthousand rows cost one transport operation instead of a thousand. `unpack` on the\nreading side reverses it. Output only.\n\nCompression is a separate middleware: list `compression` before `pack` on the\noutput and after `unpack` on the input.",
- "properties": {
- "drop_message_id": {
- "default": false,
- "description": "Leave each message's `message_id` out of the envelope, saving 16 bytes per\nrecord. Off by default, so ids survive a pack/unpack round trip.",
- "type": "boolean"
- },
- "format": {
- "$ref": "#/$defs/PackFormat",
- "default": "mqb",
- "description": "Envelope format: `mqb` (default) or `benthos_binary`."
- },
- "max_bytes": {
- "default": 4194304,
- "description": "Body-size threshold that closes a physical message, in bytes. Defaults to 4 MiB.\nA single logical message larger than this is sent on its own and exceeds it,\nrather than being dropped.",
- "format": "uint",
- "minimum": 1,
- "type": "integer"
- },
- "max_messages": {
- "default": 1000,
- "description": "Maximum logical messages per physical message. Defaults to 1000.",
- "format": "uint",
- "minimum": 1,
- "type": "integer"
- }
- },
- "type": "object"
- },
- "PostgresCdcConfig": {
- "additionalProperties": false,
- "description": "Postgres logical-replication CDC source (pgoutput). Source-only.",
- "properties": {
- "checkpoint_store": {
- "description": "Local file (path or `file://`) for resume state; unset, the slot alone tracks the stream.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "consume": {
- "anyOf": [
- {
- "$ref": "#/$defs/PostgresConsume"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Consumer only) What to read: `capture_new` (default, changes only), `capture_all`\n(backfill the tables first, then changes) or `snapshot` (backfill only)."
- },
- "create_publication": {
- "default": false,
- "description": "Create the `publication` if missing (default false; leave off if it pre-exists).\nNeeds table ownership for `publication_tables`, or superuser when none are set (`FOR ALL TABLES`).",
- "type": "boolean"
- },
- "create_slot": {
- "default": true,
- "description": "Create the replication slot if it does not exist.",
- "type": "boolean"
- },
- "cursor_id": {
- "description": "Checkpoint key for the confirmed LSN and backfill progress; defaults to `slot_name`.",
- "type": [
- "string",
- "null"
- ]
- },
- "id_hash": {
- "$ref": "#/$defs/IdHash",
- "default": "fnv1a",
- "description": "Hash for change `message_id`s: `fnv1a` (default, keeps existing dedup state) or `sha256`."
- },
- "publication": {
- "description": "Publication name (must already exist; defines which tables are captured).",
- "type": "string"
- },
- "publication_tables": {
- "default": [],
- "description": "Tables to include when managing the publication (`create_publication`); may be `schema.table`.\nMissing ones are added to an existing publication (never removed). Empty = `FOR ALL TABLES` (needs superuser).",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "slot_name": {
- "default": "mq_bridge_slot",
- "description": "Replication slot name; created if missing when `create_slot` is true.",
- "type": "string"
- },
- "source_metadata": {
- "default": false,
- "description": "Include authoritative `mqb.src.postgres_*` source positions. Defaults to false.",
- "type": "boolean"
- },
- "split_key_change": {
- "description": "An update that changes the primary key becomes a `delete` of the old key, then the `update`.",
- "type": "boolean"
- },
- "status_interval_ms": {
- "default": 10000,
- "description": "Standby-status-update interval in ms; must be shorter than the server's `wal_sender_timeout`.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "temporary_slot": {
- "default": false,
- "description": "Ephemeral run: drop the slot when the route stops. Not restart-safe; a hard crash leaks it.",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration for the replication connection."
- },
- "url": {
- "description": "Connection URL, e.g. `postgres://user:pass@host:5432/dbname`.",
- "format": "password",
- "type": "string"
- }
- },
- "required": [
- "url",
- "publication"
- ],
- "type": "object"
- },
- "PostgresConsume": {
- "description": "How a `postgres_cdc` endpoint consumes its publication. The intent-named counterpart of\n[`MongoConsume`], so the two CDC endpoints read alike.\n\n`delete` has no Postgres meaning — a replication slot is not a work queue; use a `sqlx`\nendpoint for competing consumers.",
- "oneOf": [
- {
- "const": "capture_new",
- "description": "**Stream changes only** — capture what happens from the slot's position onward. Reads\nnothing that already exists; never ends on drain. Default, because it is what\n`postgres_cdc` has always done.",
- "type": "string"
- },
- {
- "const": "capture_all",
- "description": "**Backfill, then stream** — page the publication's tables by primary key first, then\ncapture changes, with no gap between the two. The slot is created before the backfill\nstarts, so the server retains WAL throughout. Delivery across the handover is\nat-least-once, so the sink must upsert. Each table needs a single-column primary key.",
- "type": "string"
- },
- {
- "const": "snapshot",
- "description": "**One-shot backfill** — page the publication's tables by primary key, then end the\nroute. Non-destructive, creates no slot, and captures no changes.",
- "type": "string"
- }
- ]
- },
- "QdrantConfig": {
- "additionalProperties": false,
- "description": "The `qdrant` endpoint: points into one collection, or a scroll through it.",
- "properties": {
- "api_key": {
- "description": "API key sent with every request.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "checkpoint_store": {
- "description": "(Consumer only) Where the read position is saved, e.g. `file:///var/lib/mqb/cursors.json`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "collection": {
- "description": "Collection the points are written to. It must exist.",
- "type": "string",
- "x-mqb-uri": "path"
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Compression of request bodies."
- },
- "cursor_id": {
- "description": "(Consumer only) Cursor id that keys the saved read position.",
- "type": [
- "string",
- "null"
- ]
- },
- "operation": {
- "description": "Template for a message's operation, e.g. `${metadata:postgres.operation}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Address of the server, e.g. `qdrant://localhost:6333`.",
- "type": "string",
- "x-mqb-uri": "origin"
- },
- "wait": {
- "description": "Answer after the points are stored, not when they are queued. Defaults to true.",
- "type": [
- "boolean",
- "null"
- ]
- }
- },
- "required": [
- "url",
- "collection"
- ],
- "title": "Qdrant",
- "type": "object"
- },
- "RandomPanicMiddleware": {
- "additionalProperties": false,
- "description": "Middleware for fault injection testing.\n\nAllows testing error handling and recovery mechanisms by injecting faults\nat specific points in the message processing pipeline.\n\n# Examples\n\n```yaml\nrandom_panic:\n mode: panic\n trigger_on_message: 3 # Trigger on the 3rd message\n```",
- "properties": {
- "enabled": {
- "default": true,
- "description": "Enable/disable the fault injection without removing the configuration.",
- "type": "boolean"
- },
- "mode": {
- "$ref": "#/$defs/FaultMode",
- "default": "panic",
- "description": "The type of fault to inject."
- },
- "trigger_on_message": {
- "default": null,
- "description": "Trigger the fault on the Nth message (1-indexed). None = trigger on every message.",
- "format": "uint",
- "minimum": 1,
- "type": [
- "integer",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "RedisStreamsConfig": {
- "additionalProperties": false,
- "description": "Configuration for a Redis Streams endpoint.\n\nPublishers `XADD` to the stream; consumers read via a consumer group\n(`XREADGROUP` + `XACK`) by default, or ephemerally via `XREAD` from new\nmessages when `subscriber_mode` is set.",
- "properties": {
- "approx_trim": {
- "description": "(Publisher) Use approximate (`~`) trimming when `maxlen` is set. Defaults to true.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "block_ms": {
- "description": "(Consumer) Block timeout in milliseconds for each read. Defaults to 5000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "consumer_name": {
- "description": "(Consumer) Consumer name within the group. Defaults to a unique per-instance id.",
- "type": [
- "string",
- "null"
- ]
- },
- "group": {
- "description": "(Consumer) Group name. Defaults to `{APP_NAME}-{stream}`; ignored in `subscriber_mode`.",
- "type": [
- "string",
- "null"
- ]
- },
- "internal_buffer_size": {
- "description": "Internal buffer size for the consumer channel. Defaults to 128.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "maxlen": {
- "description": "(Publisher) If set, cap the stream length with `XADD MAXLEN`.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "password": {
- "description": "Optional password for authentication.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "read_from_start": {
- "default": false,
- "description": "(Consumer) On group creation, start from the stream beginning (\"0\") not \"$\". Default false.",
- "type": "boolean"
- },
- "reader_connections": {
- "description": "(Consumer) Parallel `XREADGROUP` reader connections fanned out across the group. Default 1.\nIgnored in `subscriber_mode`.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "redelivery_timeout_ms": {
- "description": "(Consumer) Redeliver entries pending ≥ this long via `XAUTOCLAIM`; 0 disables. Default 60000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "stream": {
- "description": "The stream key to publish to or read from. Defaults to the route name.",
- "type": [
- "string",
- "null"
- ]
- },
- "subscriber_mode": {
- "default": false,
- "description": "(Consumer) Read ephemerally via `XREAD` from new messages (no group/acks). Default false.",
- "type": "boolean"
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS: a private CA in `ca_file`, a client certificate in `cert_file` and `key_file`."
- },
- "url": {
- "description": "Redis URL, `redis://` or `rediss://` for TLS. Userinfo is treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "description": "Optional username for authentication (Redis ACL).",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "RequestForwardConfig": {
- "additionalProperties": false,
- "description": "Sends each message to a request-capable endpoint and forwards its response elsewhere.\n\nTurns a request/reply exchange (HTTP, or a request_reply NATS/Mongo/Memory endpoint) into\na one-way flow whose response lands on `forward_to` — e.g. IBM MQ → HTTP → IBM MQ. On\nrequest error/timeout the original message is forwarded instead (unchanged). Successful\nresponses carry the transport-native status (e.g. `http_status_code`), so a `switch` on\n`forward_to` can route them by status; a failed request forwards the original message with\nno status key, so catch failures on the switch's default branch.",
- "properties": {
- "forward_to": {
- "$ref": "#/$defs/Endpoint",
- "description": "Where the response (or, on error, the original message) is forwarded."
- },
- "to": {
- "$ref": "#/$defs/Endpoint",
- "description": "The request-capable endpoint to send each message to (e.g. an `http` client)."
- }
- },
- "required": [
- "to",
- "forward_to"
- ],
- "type": "object"
- },
- "ResponseConfig": {
- "additionalProperties": false,
- "type": "object"
- },
- "RetryMiddleware": {
- "additionalProperties": false,
- "description": "Retry middleware configuration.\n\nImplements exponential backoff retry logic for failed message processing.\nFailed messages are automatically retried with increasing delays between attempts.",
- "properties": {
- "initial_interval_ms": {
- "default": 100,
- "description": "Initial retry interval in milliseconds. Defaults to 100ms.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "max_attempts": {
- "default": 3,
- "description": "Maximum number of retry attempts. Defaults to 3.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "max_interval_ms": {
- "default": 5000,
- "description": "Maximum retry interval in milliseconds. Defaults to 5000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "multiplier": {
- "default": 2,
- "description": "Multiplier for exponential backoff. Defaults to 2.0.",
- "format": "double",
- "type": "number"
- }
- },
- "type": "object"
- },
- "Route": {
- "additionalProperties": false,
- "description": "Defines a single message processing route from an input to an output.",
- "properties": {
- "allow_fault_injection": {
- "default": false,
- "description": "Allows fault-injection middleware such as random_panic. Disabled by default.",
- "type": "boolean"
- },
- "batch_size": {
- "default": 512,
- "description": "(Optional) Maximum number of messages to process in a single batch. The consumer waits for at least one message\nand then attempts to fetch more if available. Increasing this improves throughput but also increases\nthe potential impact of a single batch processing failure. Defaults to 512.",
- "format": "uint",
- "minimum": 1,
- "type": "integer"
- },
- "commit_concurrency_limit": {
- "default": 4096,
- "description": "(Optional) The maximum number of batch commits queued or running at once. Once reached,\nthe route stops reading until a commit finishes. Defaults to 4096.",
- "format": "uint",
- "minimum": 1,
- "type": "integer"
- },
- "concurrency": {
- "default": 1,
- "description": "(Optional) Number of concurrent processing tasks for this route. While it improves throughput for high-latency\nhandlers, it adds synchronization overhead for ordered commits and may lead to out-of-order processing\nin the handler. Above 1, whole batches may also reach the sink out of source order (rows keep their order\nwithin a batch) unless the sink declares itself order-sensitive, as `file` does.\nDefaults to 1.",
- "format": "uint",
- "minimum": 1,
- "type": "integer"
- },
- "description": {
- "description": "A human-readable description of the route's purpose. Defaults to an empty string.",
- "type": "string"
- },
- "empty_batch_delay_ms": {
- "default": 10,
- "description": "Delay after an empty receive batch to avoid hot polling. Set to 0 to only yield. Defaults to 10ms.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "exit_on_empty": {
- "description": "If true, the route exits gracefully once the source yields an empty batch\n(drain-then-exit). Off by default — routes normally poll indefinitely.\nA drain that fails before it delivered anything ends at once; later it gives up after ten reconnects.",
- "type": "boolean"
- },
- "input": {
- "$ref": "#/$defs/Endpoint",
- "description": "The input/source endpoint for the route."
- },
- "output": {
- "anyOf": [
- {
- "$ref": "#/$defs/Endpoint"
- },
- {
- "type": "null"
- }
- ],
- "default": {
- "middlewares": [],
- "null": null
- },
- "description": "The output/sink endpoint for the route."
- },
- "reconnect_interval_ms": {
- "default": 5000,
- "description": "Time to wait before reconnecting after a transient route failure. Defaults to 5000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- },
- "required_delivery": {
- "anyOf": [
- {
- "$ref": "#/$defs/DeliveryGuarantee"
- },
- {
- "type": "null"
- }
- ],
- "description": "Fail the route at startup unless its inferred delivery guarantee is at least this."
- },
- "startup_timeout_ms": {
- "default": 5000,
- "description": "Time to wait for a route to establish connections before startup fails. Defaults to 5000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "input"
- ],
- "type": "object"
- },
- "SequenceConfig": {
- "additionalProperties": false,
- "description": "Reads several input endpoints one after another: each is drained before the next\nbegins, and the last one streams until the route stops.",
- "properties": {
- "checkpoint_store": {
- "description": "(Optional) Where to persist the phase marker. A local path, `file://`, or any URL\nthe shared checkpoint stores accept. Without it, every restart begins at phase 1.",
- "type": [
- "string",
- "null"
- ]
- },
- "cursor_id": {
- "description": "(Optional) Identifies the durable phase marker, so a restart resumes at the phase\nit had reached instead of replaying the earlier ones. Needs `checkpoint_store`.",
- "type": [
- "string",
- "null"
- ]
- },
- "endpoints": {
- "description": "The inputs to read, in order. At least one is required.",
- "items": {
- "$ref": "#/$defs/Endpoint"
- },
- "minItems": 1,
- "type": "array"
- }
- },
- "required": [
- "endpoints"
- ],
- "type": "object"
- },
- "SledConfig": {
- "additionalProperties": false,
- "description": "General Sled database configuration",
- "properties": {
- "delete_after_read": {
- "default": false,
- "description": "(Consumer only) If true, delete messages after processing (Queue mode).",
- "type": "boolean"
- },
- "path": {
- "description": "Path to the Sled database directory.",
- "type": "string"
- },
- "read_from_start": {
- "default": false,
- "description": "(Consumer only) If true, start reading from the beginning of the tree.",
- "type": "boolean"
- },
- "tree": {
- "description": "The tree name to use as a queue. Defaults to \"default\".",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "path"
- ],
- "type": "object"
- },
- "SpoolClaim": {
- "description": "What a `dir_spool` endpoint does when a second instance opens the same directory in the\nsame role — a second producer writing it, or a second draining consumer emptying it.\n\nNeither is supported by the endpoint's design: two producers seeded from the same\nhighest sequence number overwrite each other's chunks, and two draining consumers each\ndeliver every chunk they win the race to read. So each role takes a pid lock on its own\nfile — `producer_file` or `consumer_file` — created when the endpoint opens and removed\nwhen it closes. A producer and a consumer sharing a directory do not conflict:\nthat is the whole point of the endpoint.\n\nThe locks say \"someone is running\", not \"production is finished\" — several producers\nmay fill one spool in turn, and `done_file` is what marks the end of that.\n\nA lock whose owner is no longer running is broken and retaken, so a crash does not\nwedge a restart. That check is by process id, which is only meaningful on the machine\nand in the pid namespace that wrote it: a spool shared between hosts or containers can\nhave a live holder's lock broken by a second starter that cannot see its process.",
- "oneOf": [
- {
- "const": "exclusive",
- "description": "Refuse to start when the role is already claimed, naming the holder (default).",
- "type": "string"
- },
- {
- "const": "warn",
- "description": "Log a warning and run anyway. For a spool deliberately shared by several\nproducers, which needs `{message_id}` in `naming_pattern` to keep names unique.",
- "type": "string"
- },
- {
- "const": "off",
- "description": "Take no claim and check for none.",
- "type": "string"
- }
- ]
- },
- "SpoolDone": {
- "description": "When a `dir_spool` publisher writes its `done_file`, marking production finished.\n\nThe distinction is *how* the route ended, which the publisher learns from\n[`DisconnectOutcome`](crate::traits::DisconnectOutcome): every teardown path closes the\npublisher, so being closed says nothing on its own.",
- "oneOf": [
- {
- "const": "never",
- "description": "Never write it (default). Something else decides when production is finished.",
- "type": "string"
- },
- {
- "const": "success",
- "description": "Write it only when the route reached the natural end of its input and every chunk\nthis producer accepted was written.\n\nA route that is shut down, that fails, or that reconnects does *not* write it —\nproduction did not finish, so a `stop_on_done` consumer keeps waiting. Note that a\ncontinuously running producer never reaches a natural end, so `success` on one only\never means \"not yet\": use `end` there.",
- "type": "string"
- },
- {
- "const": "end",
- "description": "Write it whenever the producer closes, however the route ended — a clean finish, a\nshutdown, or a failure.\n\nThis says \"nothing more is coming from here\", not \"everything worked\". Use it for a\nconsumer that must not wait forever on a producer that may have died, at the price\nof it treating a truncated stream as complete.",
- "type": "string"
- }
- ]
- },
- "SpoolFsync": {
- "description": "How hard a `dir_spool` endpoint works to make its writes survive a power loss.\n\nThis is the endpoint's dominant cost. `chunk`, the default, is two fsyncs per message —\nthe payload and its sidecar — plus one directory fsync per batch, and on a spinning disk\nor a conservative SSD that sets the throughput ceiling long before anything in\nmq-bridge does. Dropping the sidecar (`metadata_extension: \"\"`) halves it; `off` removes\nit.\n\nIndependent of `atomic`, which decides whether a reader can see a half-written chunk,\nnot whether a crash can lose one. The once-per-run fsyncs — the lock, the sentinel — are\nnot affected by either.",
- "oneOf": [
- {
- "const": "chunk",
- "description": "fsync every chunk file before it is renamed into place, and the directory once per\nbatch (default). An acknowledged batch has reached the disk.",
- "type": "string"
- },
- {
- "const": "off",
- "description": "Do not fsync chunks at all, and let the operating system flush when it chooses.\n\nA crash or a power loss can then lose recent chunks *or* leave one present but\ntruncated, which a consumer would deliver as a short message — so this trades the\nendpoint's crash-safety claim, not merely its tail. Reasonable when the spool is a\nbuffer between two processes on one machine and a reboot means starting over anyway.",
- "type": "string"
- }
- ]
- },
- "SqlColumns": {
- "description": "How a SQL sink maps a record to the columns of its table.",
- "oneOf": [
- {
- "const": "auto",
- "description": "Each top-level JSON field goes into the table column of the same name.",
- "type": "string"
- }
- ]
- },
- "SqlTimestamps": {
- "description": "How a SQL source renders timestamp columns.",
- "oneOf": [
- {
- "const": "text",
- "description": "The database's own text form (`2026-10-04 09:15:23.923277+00`).",
- "type": "string"
- },
- {
- "const": "rfc3339",
- "description": "RFC 3339 in UTC (`2026-10-04T09:15:23.923277Z`); `timestamp` without zone has no offset.",
- "type": "string"
- }
- ]
- },
- "SqlxConfig": {
- "additionalProperties": false,
- "description": "General SQLx connection configuration.",
- "properties": {
- "acquire_timeout_ms": {
- "description": "Timeout for acquiring a connection from the pool in milliseconds. Defaults to 30000ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "auto_create_table": {
- "default": false,
- "description": "(Publisher only) If true, automatically create the table and indexes if they don't exist. Defaults to false.",
- "type": "boolean"
- },
- "bulk_copy": {
- "default": false,
- "description": "(Publisher only) PostgreSQL only. Bulk-load batches via `COPY FROM STDIN` (much faster than multi-row INSERT). Requires a token-based `insert_query`; no `ON CONFLICT`/`RETURNING`.",
- "type": "boolean"
- },
- "checkpoint_store": {
- "description": "(Consumer only) Where to persist the resume cursor in `cursor_column` mode. A URL selects the\nbackend; a bare name (or `/name`) reuses the **source** datastore with that table name:\n- absent → source datastore, table `mqb_cursors_<source_table>` (auto-unique)\n- `/my_cursors` → source datastore, table `my_cursors`\n- `file:///var/lib/mqb/cursors.json` → local JSON file (read-only / write-restricted sources)\n- `postgres://user@host/db/table` or `mysql://host/db/table` → external SQL table (table optional)\n- `mongodb://host/db/collection` → external MongoDB collection (collection optional)\n- `s3://bucket/prefix` (also `gs://`, `az://`, `abfs://`) → cloud object store; creds via env\n\nWhen no table/collection is named, it defaults to `mqb_cursors_<source_table>`.\nMay embed connection credentials, so it is treated as a secret.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "columns": {
- "anyOf": [
- {
- "$ref": "#/$defs/SqlColumns"
- },
- {
- "type": "null"
- }
- ],
- "description": "(Publisher only) `auto`: write each JSON field into the table column of the same name. The table must exist."
- },
- "create_publication": {
- "default": false,
- "description": "(Consumer only, CDC) When `publication` is set, create it if missing (default false).\nNeeds table-owner privilege: it is auto-published `FOR TABLE {table}`.",
- "type": "boolean"
- },
- "cursor_column": {
- "description": "(Consumer only) Read an existing table **non-destructively** and resumably, paging by this\nmonotonic column (`SELECT * FROM {table} WHERE {cursor_column} > $last ORDER BY {cursor_column} ASC LIMIT n`)\nand persisting the last read value under `cursor_id`. Does not delete/lock source rows.\nMutually exclusive with `delete_after_read`.",
- "type": [
- "string",
- "null"
- ]
- },
- "cursor_id": {
- "description": "(Consumer only) Cursor id used to key the persisted resume position. Recommended when\n`cursor_column` is set: without it, progress is not persisted and every restart re-copies\nfrom the beginning.",
- "type": [
- "string",
- "null"
- ]
- },
- "delete_after_read": {
- "default": false,
- "description": "(Consumer only) If true, delete messages after processing.",
- "type": "boolean"
- },
- "extra_column": {
- "description": "(Publisher only, with `columns`) Column that takes the fields without a column of their own, as one JSON object.",
- "type": [
- "string",
- "null"
- ]
- },
- "idle_timeout_ms": {
- "description": "Maximum idle time for a connection in milliseconds. Defaults to 600000ms (10 minutes).",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "insert_query": {
- "description": "(Publisher only) Optional. A custom SQL INSERT query. Use `?` as a placeholder for the payload.\nIf not provided, a default `INSERT INTO {table} (payload) VALUES (?)` is used.\n\nFor multi-column inserts, embed explicit source tokens directly in the query:\n`${metadata:<key>}` binds `message.metadata[\"<key>\"]`, and `${payload:<field>}`\nbinds the top-level JSON field `<field>` of the payload (types preserved:\nnumbers/bools stay numeric/bool). There is no fallback between the two: an\nabsent metadata key, non-JSON payload, or missing/non-scalar field binds SQL NULL.\nExample: `INSERT INTO orders (customer_id, sku, qty) VALUES (${metadata:customer_id}, ${payload:sku}, ${payload:qty})`.\nA query with no `${...}` tokens behaves exactly as before (whole payload bound once).\n`auto_create_table` is not supported together with a token-based query.\n\nTokens bind as text/number/bool; Postgres won't implicitly cast text into a\n`numeric`/`timestamptz` column (these arrive as JSON strings from a sql source).\nAdd an explicit cast next to the token — it is preserved verbatim in the SQL:\n`VALUES (${payload:amount}::numeric, ${payload:created_at}::timestamptz)`.",
- "type": [
- "string",
- "null"
- ]
- },
- "key": {
- "description": "(Publisher only, with `columns`) Key column(s), comma-separated: a row with the same key is updated, not inserted again.",
- "type": [
- "string",
- "null"
- ]
- },
- "lookup_query": {
- "description": "(Publisher only) SELECT for `lookup`, answering with the first row. Write `WHERE id IN\n(${payload:id})` to answer a whole batch in one query, matching rows by `id`.",
- "type": [
- "string",
- "null"
- ]
- },
- "max_connections": {
- "description": "Maximum number of connections in the pool. Defaults to 10.",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_lifetime_ms": {
- "description": "Maximum lifetime of a connection in milliseconds. Defaults to 1800000ms (30 minutes).",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "max_polling_interval_ms": {
- "description": "(Consumer only) If set, the poll interval backs off exponentially from `polling_interval_ms`\nup to this value while drained, resetting on new rows. Unset = constant interval.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "min_connections": {
- "description": "Minimum number of connections to keep in the pool. Defaults to 0.",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "password": {
- "default": null,
- "description": "Optional password. Takes precedence over any credentials embedded in the `url`.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "polling_interval_ms": {
- "description": "(Consumer only) Polling interval in milliseconds. Defaults to 100ms.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "publication": {
- "description": "(Consumer only, PostgreSQL) If set, consume via logical-replication CDC instead of cursor\npolling: streams inserts/updates/deletes from this publication. Requires the `postgres-cdc`\nfeature and a Postgres URL. For full control use the dedicated `postgres_cdc` endpoint.",
- "type": [
- "string",
- "null"
- ]
- },
- "select_query": {
- "description": "(Consumer only) Optional. A custom SQL SELECT query to fetch messages. This is only supported for PostgreSQL and Microsoft SQL Server.\nThe query must include a placeholder for the batch size (`$1` for PostgreSQL, `@p1` for SQL Server).\nThe bridge will bind the route's `batch_size` to this placeholder.",
- "type": [
- "string",
- "null"
- ]
- },
- "shared": {
- "default": true,
- "description": "Share one connection pool per connection (default: true); false forces a dedicated pool.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "slot_name": {
- "description": "(Consumer only, CDC) Replication slot name; created if missing. Defaults to `mq_bridge_slot`.",
- "type": [
- "string",
- "null"
- ]
- },
- "source_metadata": {
- "default": false,
- "description": "(Consumer only) Include authoritative `mqb.src.sqlx_*` source positions; `cursor_column` must then be a unique integer. Defaults to false.",
- "type": "boolean"
- },
- "table": {
- "description": "The table to interact with.",
- "type": "string"
- },
- "test_before_acquire": {
- "description": "Ping each pooled connection before handing it out (default false). Costs a round-trip per acquire.",
- "type": [
- "boolean",
- "null"
- ]
- },
- "timestamps": {
- "$ref": "#/$defs/SqlTimestamps",
- "description": "(Consumer only, `cursor_column` mode, PostgreSQL) How timestamp columns are rendered: `text` (default) or `rfc3339`."
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS configuration for the database connection."
- },
- "url": {
- "description": "Database connection URL. If it contains userinfo, it will be treated as a secret.",
- "format": "password",
- "type": "string"
- },
- "username": {
- "default": null,
- "description": "Optional username. Takes precedence over any credentials embedded in the `url`.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "required": [
- "url",
- "table"
- ],
- "type": "object"
- },
- "StaticConfig": {
- "description": "Configuration for the `static` endpoint. Accepts either a bare string (the response body, JSON-encoded for backward compatibility) or a map where only `body` is required and `raw` / `metadata` are optional.",
- "oneOf": [
- {
- "description": "The response body, JSON-encoded as a string.",
- "type": "string"
- },
- {
- "additionalProperties": false,
- "properties": {
- "body": {
- "description": "The static response body.",
- "type": "string"
- },
- "metadata": {
- "additionalProperties": {
- "type": "string"
- },
- "description": "Extra metadata entries attached to the produced message.",
- "type": "object"
- },
- "raw": {
- "default": false,
- "description": "Send the body verbatim instead of JSON-encoding it as a string.",
- "type": "boolean"
- }
- },
- "required": [
- "body"
- ],
- "type": "object"
- }
- ]
- },
- "StreamBufferConfig": {
- "additionalProperties": false,
- "description": "Configuration for the correlated in-process stream response buffer.",
- "properties": {
- "capacity": {
- "description": "Capacity of each correlation partition. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "correlation_id": {
- "description": "Consumer-only correlation id partition to read from.\n\nLeave this unset for the publisher endpoint configured in\n`HttpConfig::stream_response_to`. Set it on consumers so a reader only\nreceives messages belonging to one request or response stream.",
- "type": [
- "string",
- "null"
- ]
- },
- "idle_ttl_secs": {
- "description": "Seconds before a partition with no attached consumer is discarded. 0 disables. Defaults to 3600.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "topic": {
- "description": "Shared buffer topic used by both the publisher and correlated consumers.",
- "type": "string"
- }
- },
- "required": [
- "topic"
- ],
- "type": "object"
- },
- "SwitchCase": {
- "additionalProperties": false,
- "description": "One predicate case of a `switch` in `when` mode.",
- "properties": {
- "if": {
- "description": "Expression over payload fields and `meta.<key>`, e.g. `amount > 100`.",
- "type": "string"
- },
- "to": {
- "$ref": "#/$defs/Endpoint",
- "description": "Where a matching message goes."
- }
- },
- "required": [
- "if",
- "to"
- ],
- "type": "object"
- },
- "SwitchConfig": {
- "additionalProperties": false,
- "properties": {
- "cases": {
- "additionalProperties": {
- "$ref": "#/$defs/Endpoint"
- },
- "default": {},
- "description": "Value-lookup mode: a map of metadata values to endpoints.",
- "type": "object"
- },
- "default": {
- "anyOf": [
- {
- "$ref": "#/$defs/Endpoint"
- },
- {
- "type": "null"
- }
- ],
- "description": "The default endpoint if no case matches."
- },
- "metadata_key": {
- "default": "",
- "description": "Value-lookup mode: the metadata key whose value picks the case.",
- "type": "string"
- },
- "when": {
- "default": [],
- "description": "Predicate mode: ordered cases, first match wins. Needs the `filter` feature.",
- "items": {
- "$ref": "#/$defs/SwitchCase"
- },
- "type": "array"
- }
- },
- "type": "object"
- },
- "TimeoutMiddleware": {
- "additionalProperties": false,
- "description": "Send timeout middleware configuration.\n\nFails a `send`/`send_batch` that has not finished within `timeout_ms` with a\nretryable error instead of letting it block the route. The sink may still have\naccepted the batch, so a retry after a timeout can deliver it twice.",
- "properties": {
- "timeout_ms": {
- "description": "Longest a single send may take, in milliseconds.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "timeout_ms"
- ],
- "type": "object"
- },
- "TlsConfig": {
- "additionalProperties": false,
- "description": "TLS configuration for secure connections.\n\nConfigures Transport Layer Security (TLS/SSL) for encrypted communication.\nSupports both client certificate (mutual TLS) and server certificate validation.\n\n# Examples\n\n```\nuse mq_bridge::models::TlsConfig;\n\nlet tls = TlsConfig {\n required: true,\n ca_file: Some(\"/path/to/ca.pem\".to_string()),\n cert_file: Some(\"/path/to/cert.pem\".to_string()),\n key_file: Some(\"/path/to/key.pem\".to_string()),\n ..Default::default()\n};\n```",
- "properties": {
- "accept_invalid_certs": {
- "default": false,
- "description": "If true, disable server certificate verification (insecure).",
- "type": "boolean"
- },
- "ca_file": {
- "description": "Path to the CA certificate file.",
- "type": [
- "string",
- "null"
- ]
- },
- "cert_file": {
- "description": "Path to the client certificate file (PEM).",
- "type": [
- "string",
- "null"
- ]
- },
- "cert_password": {
- "description": "Password for the private key (if encrypted).",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "key_file": {
- "description": "Path to the client private key file (PEM).",
- "type": [
- "string",
- "null"
- ]
- },
- "required": {
- "default": false,
- "description": "If true, enable TLS/SSL.",
- "type": "boolean"
- }
- },
- "type": "object"
- },
- "TransformErrorPolicy": {
- "description": "Action taken on a message that fails to transform.",
- "oneOf": [
- {
- "const": "reject",
- "description": "Reject the message: non-retryable failure on output, dropped from the batch on input.",
- "type": "string"
- },
- {
- "const": "pass_through",
- "description": "Forward the original payload unchanged with the error recorded in metadata.",
- "type": "string"
- }
- ]
- },
- "TransformMiddleware": {
- "additionalProperties": false,
- "description": "JSON transform middleware configuration.\n\nApplies `mapping`, a Zen `expression`, then an optional `schema`.\n\nOn an output endpoint a rejected message becomes a non-retryable failure, so a `dlq`\nmiddleware listed *after* this one captures it (publisher middlewares are wrapped in\nlist order, so the last entry is the outermost layer). On an input endpoint a rejected\nmessage is dropped from the batch and acknowledged, which is how invalid input is kept\nout of the route.",
- "not": {
- "properties": {
- "schema_file": {
- "type": "string"
- }
- },
- "required": [
- "schema",
- "schema_file"
- ]
- },
- "properties": {
- "apply_defaults": {
- "default": true,
- "description": "Insert `default` values from the schema for missing fields. Defaults to true.",
- "type": "boolean"
- },
- "coerce": {
- "default": true,
- "description": "Coerce safely convertible types (e.g. `\"42\"` -> `42`) instead of rejecting. Defaults to true.",
- "type": "boolean"
- },
- "coerce_empty_as_null": {
- "default": false,
- "description": "Read an empty string as `null`, so a nullable field or its default wins. Defaults to false.",
- "type": "boolean"
- },
- "expression": {
- "description": "Zen Expression whose result replaces the mapped payload.",
- "type": [
- "string",
- "null"
- ]
- },
- "mapping": {
- "additionalProperties": {
- "$ref": "#/$defs/MappingRule"
- },
- "description": "Output field name -> source path (e.g. `firstName: \"$.first_name\"`). Dots nest the output.",
- "type": "object"
- },
- "on_error": {
- "$ref": "#/$defs/TransformErrorPolicy",
- "default": "reject",
- "description": "What to do with a message that fails to transform. Defaults to `reject`."
- },
- "schema": {
- "description": "Inline JSON Schema subset (type, properties, required, default, items, nullable, enum)."
- },
- "schema_file": {
- "description": "Path to a JSON Schema file. Read once at startup; never re-read per message.",
- "type": [
- "string",
- "null"
- ]
- }
- },
- "type": "object"
- },
- "TypesenseConfig": {
- "additionalProperties": false,
- "description": "The `typesense` endpoint: documents into one Typesense collection.",
- "properties": {
- "api_key": {
- "description": "API key sent with every request.",
- "format": "password",
- "type": [
- "string",
- "null"
- ]
- },
- "collection": {
- "description": "Collection the documents are written to. It must exist.",
- "type": "string",
- "x-mqb-uri": "path"
- },
- "compression": {
- "$ref": "#/$defs/Compression",
- "default": "none",
- "description": "Compression of request bodies."
- },
- "operation": {
- "description": "Template for a message's operation, e.g. `${metadata:postgres.operation}`.",
- "type": [
- "string",
- "null"
- ]
- },
- "request_timeout_ms": {
- "description": "Request timeout in milliseconds. Unset = 300 s without a response byte; `0` = none.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "description": "TLS options as on `http_bulk` (`ca_file`, `cert_file`, `key_file`, `required`); JSON in a URI."
- },
- "url": {
- "description": "Address of the server, e.g. `typesense://localhost:8108`.",
- "type": "string",
- "x-mqb-uri": "origin"
- }
- },
- "required": [
- "url",
- "collection"
- ],
- "title": "Typesense",
- "type": "object"
- },
- "UnpackMiddleware": {
- "additionalProperties": false,
- "description": "Transport batching middleware configuration (`unpack`, input side).\n\nSplits one physical message back into the logical messages `pack` put in it.\nInput only.",
- "properties": {
- "format": {
- "$ref": "#/$defs/PackFormat",
- "default": "mqb",
- "description": "Envelope format: `mqb` (default) or `benthos_binary`. Must match the sender."
- },
- "max_messages": {
- "default": null,
- "description": "Reject a batch declaring more messages than this. Unset means no limit.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "on_error": {
- "$ref": "#/$defs/InputErrorPolicy",
- "default": "fail",
- "description": "What to do with a message that is not a valid envelope. Defaults to `fail`."
- }
- },
- "type": "object"
- },
- "WeakJoinAck": {
- "description": "When a weak join acknowledges the source messages it buffers.",
- "oneOf": [
- {
- "const": "on_join",
- "description": "Ack members once their joined message is committed; a crash redelivers open groups.",
- "type": "string"
- },
- {
- "const": "on_receive",
- "description": "Ack on receipt; open groups are lost on a crash. For sources whose prefetch limit is\nsmaller than the messages held in open groups.",
- "type": "string"
- }
- ]
- },
- "WeakJoinMiddleware": {
- "additionalProperties": false,
- "description": "Weak Join middleware configuration.\n\nCorrelates messages by a metadata key and joins them within a timeout window.\nCount mode (default) waits for `expected_count` messages and emits a JSON array.\nBranch mode (set `branch_by`) waits for named branches and emits a branch-keyed object.",
- "properties": {
- "ack": {
- "$ref": "#/$defs/WeakJoinAck",
- "default": "on_join",
- "description": "When source messages are acknowledged: after the joined message is committed (default) or on receipt."
- },
- "branch_by": {
- "default": null,
- "description": "Metadata key naming each message's branch; enables branch mode when set.",
- "type": [
- "string",
- "null"
- ]
- },
- "expected_count": {
- "description": "The number of messages (count mode) or distinct branches (branch mode) to wait for.",
- "format": "uint",
- "minimum": 0,
- "type": "integer"
- },
- "group_by": {
- "description": "The metadata key to group messages by (e.g., \"correlation_id\").",
- "type": "string"
- },
- "on_timeout": {
- "$ref": "#/$defs/WeakJoinTimeout",
- "default": "fire",
- "description": "What to do with an incomplete group when the timeout expires."
- },
- "required": {
- "default": [],
- "description": "Branch names that must all arrive before firing (branch mode; overrides expected_count).",
- "items": {
- "type": "string"
- },
- "type": "array"
- },
- "timeout_ms": {
- "description": "Timeout in milliseconds.",
- "format": "uint64",
- "minimum": 0,
- "type": "integer"
- }
- },
- "required": [
- "group_by",
- "expected_count",
- "timeout_ms"
- ],
- "type": "object"
- },
- "WeakJoinTimeout": {
- "description": "Action taken on an incomplete weak-join group when its timeout expires.",
- "oneOf": [
- {
- "const": "fire",
- "description": "Emit the partial join (current behavior).",
- "type": "string"
- },
- {
- "const": "discard",
- "description": "Drop the incomplete group without emitting.",
- "type": "string"
- }
- ]
- },
- "WebSocketConfig": {
- "additionalProperties": false,
- "description": "WebSocket connection configuration.",
- "properties": {
- "backlog": {
- "description": "(Consumer only) TCP listen backlog (pending-connection queue depth) for the accept socket.\nRaise this if high-concurrency handshake bursts are being dropped/reset before `accept()`\ncan keep up. Defaults to 4096, which is higher than the OS/tokio default of 1024.",
- "format": "uint32",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "basic_auth": {
- "description": "HTTP Basic credentials (username, password): a consumer requires them, a publisher sends them.",
- "format": "password",
- "maxItems": 2,
- "minItems": 2,
- "prefixItems": [
- {
- "type": "string"
- },
- {
- "type": "string"
- }
- ],
- "type": [
- "array",
- "null"
- ]
- },
- "execution_mode": {
- "$ref": "#/$defs/WebSocketExecutionMode",
- "default": "auto",
- "description": "(Consumer only) Selects whether WebSocket routes run directly or through the routed pipeline."
- },
- "max_connections": {
- "description": "(Consumer only) Maximum number of concurrent connections; additional connections receive 503. Unset = no limit.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "message_id_header": {
- "description": "(Consumer only) Header key to extract the message ID from the WebSocket handshake. Defaults to \"message-id\".",
- "type": [
- "string",
- "null"
- ]
- },
- "path": {
- "description": "(Consumer only) Optional request path filter. If set, only upgrade requests whose URI path matches exactly are delivered to this consumer.",
- "type": [
- "string",
- "null"
- ]
- },
- "routed_queue_capacity": {
- "description": "(Consumer only) Queue capacity for the routed adapter. Direct response routes do not use this queue. Defaults to 100.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "tls": {
- "$ref": "#/$defs/TlsConfig",
- "default": {
- "accept_invalid_certs": false,
- "ca_file": null,
- "cert_file": null,
- "cert_password": null,
- "key_file": null,
- "required": false
- },
- "description": "TLS: a consumer listens with `cert_file` and `key_file`; a publisher uses it for a `wss://` URL."
- },
- "url": {
- "description": "For consumers, the listen address (e.g. \"0.0.0.0:9000\"). For publishers, the target URL.",
- "format": "password",
- "type": "string"
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "WebSocketExecutionMode": {
- "description": "WebSocket route execution strategy.",
- "oneOf": [
- {
- "const": "auto",
- "description": "Use direct per-connection handling for simple `websocket -> response` routes and fall back\nto the routed adapter with a warning when route semantics need the normal pipeline.",
- "type": "string"
- },
- {
- "const": "direct_only",
- "description": "Require direct per-connection handling. Startup fails if the route cannot run directly.",
- "type": "string"
- },
- {
- "const": "routed",
- "description": "Always use the normal routed consumer/worker/disposition pipeline.",
- "type": "string"
- }
- ]
- },
- "ZeroMqBackend": {
- "description": "ZeroMQ backend implementation.\n\n`omq` uses `omq-tokio` (omq.rs) — much faster on the per-message `raw`/`raw_framed`\npath and adds CURVE/PLAIN security — and requires the `zeromq-omq` build feature\n(MSRV 1.93). `zmq` uses the `zeromq` crate (pure-Rust zmq.rs) and covers every\nsocket type.\n\n`try_omq` is the default: it picks `omq` when that feature is compiled in and the\nconfigured socket type is supported there, and falls back to `zmq` otherwise, so a\nbuild without `zeromq-omq` still runs. Name a backend explicitly to make the choice\na hard requirement instead — a missing feature is then a startup error.",
- "oneOf": [
- {
- "enum": [
- "zmq",
- "omq"
- ],
- "type": "string"
- },
- {
- "const": "try_omq",
- "description": "Prefer `omq`, fall back to `zmq` when it is unavailable for this build or socket.",
- "type": "string"
- }
- ]
- },
- "ZeroMqConfig": {
- "additionalProperties": false,
- "properties": {
- "backend": {
- "$ref": "#/$defs/ZeroMqBackend",
- "default": "try_omq",
- "description": "Backend: `try_omq` (default, prefer `omq` and fall back to `zmq`), `zmq` (the `zeromq` crate) or `omq` (the `omq-tokio` backend). `omq` needs the `zeromq-omq` build feature."
- },
- "bind": {
- "default": false,
- "description": "If true, bind to the address. If false, connect.",
- "type": "boolean"
- },
- "format": {
- "$ref": "#/$defs/ZeroMqFormat",
- "default": "raw_framed",
- "description": "Wire format: `json` wraps the CanonicalMessage; `raw` sends payload bytes per frame; `raw_framed` adds a JSON metadata frame. Default `raw_framed`.\nREQ/REP replies are the exception: a REP peer always answers with a JSON array of\ncanonical messages and a REQ publisher always decodes one, whatever `format` is set to."
- },
- "internal_buffer_size": {
- "default": null,
- "description": "Internal buffer size for the channel. Defaults to 128. `zmq` backend only — `omq`\napplies HWM backpressure on the socket itself and ignores this.",
- "format": "uint",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "request_timeout_ms": {
- "default": null,
- "description": "(REQ publisher only) Timeout in ms for one request/reply exchange before it is reported as failed. Defaults to 30000.",
- "format": "uint64",
- "minimum": 0,
- "type": [
- "integer",
- "null"
- ]
- },
- "socket_type": {
- "anyOf": [
- {
- "$ref": "#/$defs/ZeroMqSocketType"
- },
- {
- "type": "null"
- }
- ],
- "default": null,
- "description": "The socket type (PUSH, PULL, PUB, SUB, REQ, REP)."
- },
- "topic": {
- "description": "Topic: the prefix a SUB socket filters on, or the topic frame a PUB socket sends first.",
- "type": [
- "string",
- "null"
- ]
- },
- "url": {
- "description": "The ZeroMQ URL (e.g., \"tcp://127.0.0.1:5555\").",
- "format": "password",
- "type": "string"
- }
- },
- "required": [
- "url"
- ],
- "type": "object"
- },
- "ZeroMqFormat": {
- "description": "ZeroMQ wire format.\n\n`json` wraps each message as a JSON CanonicalMessage (batched into one frame);\n`raw` sends/receives the payload bytes directly, one frame per message (metadata\nis not transmitted); `raw_framed` (default) sends a two-frame message — a JSON metadata\nframe followed by the raw payload frame — keeping the payload binary-safe while still\ncarrying headers. Use `raw`/`raw_framed` for binary feeds such as JPEG, Avro or Protobuf.",
- "enum": [
- "json",
- "raw",
- "raw_framed"
- ],
- "type": "string"
- },
- "ZeroMqSocketType": {
- "description": "ZeroMQ socket type.\n\nDefines the messaging pattern for ZeroMQ connections.\nDifferent patterns support different communication paradigms (request-reply, publish-subscribe, etc.).",
- "enum": [
- "push",
- "pull",
- "pub",
- "sub",
- "req",
- "rep"
- ],
- "type": "string"
- }
-}
