# Examples: NATS and JetStream

> NATS: publish to JetStream from HTTP or MCP, read in SSE and WebSocket bound to the tenant, deliver by webhook.

Example - Since version 1.6.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-examples-nats/

> Operations guide: configure, deploy and observe the gateway. What the feature does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

## NATS JetStream: publish from HTTP or an MCP tool

> [!NOTE]
> Available since `1.6.0`; the example is validated by the binary and pinned by the tests.

An HTTP service or an agent publishes into a JetStream stream **without a NATS client**; `202` only after
the stream's `PubAck`, the `Idempotency-Key` becoming `Nats-Msg-Id` - the stream deduplicates a resend after a `504`.
The gateway creates **no** stream: an existing stream must capture the subject, otherwise `502 upstream_error`.

<!-- example: examples/async-nats-jetstream/flexway.yaml -->

```yaml
# NATS JetStream behind the gateway's controls (1.6.0): an HTTP door and an MCP tool publish
# to a stream without a NATS client - 202 only after the stream's PubAck, the Idempotency-Key as
# Nats-Msg-Id. The gateway creates no stream: a subject no stream captures answers 502.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222", "tls://nats-1.nats:4222"]   # nats:// or tls://, never a credential in the URL
    tls: { server_name: nats.internal }
    nats:
      credentials: "file:/etc/flexway-secrets/nats/flexway.creds"   # the gateway's NATS user: JWT and nkey seed
      jetstream: true                            # 202 after the stream's PubAck; false (NATS Core) refuses a publication
    timeout: 2s                                  # one publication's budget, the PubAck included
    max_in_flight: 512                           # bounded buffer: past it, 503 overloaded
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every subject below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp
```

```yaml
# examples/async-nats-jetstream/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"          # captured by the ORDERS stream (orders.>): the gateway creates no stream
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      created:
        contentType: application/json
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: orders-nats
```

```yaml
# examples/async-nats-jetstream/openapi.yaml
openapi: 3.1.0
info: { title: Orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }   # Idempotency-Key -> Nats-Msg-Id
      x-flexway-tool: { name: emit_order_created }   # the same door, offered to agents as an MCP tool
      security: [ { customers: [orders:write] } ]
      requestBody:
        required: true
        content:
          application/json: { schema: { type: object, required: [orderId], properties: { orderId: { type: string } } } }
      responses: { "202": { description: Acknowledged by the stream } }
```

| What happens | NATS JetStream |
| --- | --- |
| `POST /orders/events` | Admission, body checked, CloudEvents in binary mode (`ce-*` headers), published on `orders.acme.created`, `202` after the `PubAck` |
| The same call as an MCP tool (`emit_order_created`) | The same decision, the same acknowledgement |
| `jetstream: false` (NATS Core) | > [!WARNING]
> Refused at load: Core acknowledges nothing, a `202` would lie |
> | No stream captures the subject | `502 upstream_error` ("no responders"), never a `202` |

## NATS: SSE and WebSocket bound to the tenant, wildcards, NATS Core

> [!NOTE]
> Available since `1.6.0`; the example is validated by the binary and pinned by the tests.

A dashboard reads **every** order event of its tenant from one stream, thanks to a wildcard **inside the
channel** after the bound segment; an operator reads the metrics of **its own** site on a NATS Core server
(SSE only, at most once); a public screen reads the status without a token. Each message received is checked against the
channel's pattern, segment by segment: a broker that sent too much never leaks a tenant.

<!-- example: examples/async-nats-streams/flexway.yaml -->

```yaml
# NATS in server-sent events and WebSocket (1.6.0): a dashboard reads every order event of
# its tenant through a wildcard written in the channel after the bound segment; an operator reads its
# own site's readings from NATS Core (server-sent events only, at most once); a public status is read
# without a token. Two brokers, two identities.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      token: "file:/etc/flexway-secrets/nats/token"   # a token, by reference - over TLS only
      jetstream: true
  metrics-core:
    protocol: nats
    brokers: ["nats://nats-edge.plant:4222"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/nats-edge/password" }
    tls: { server_name: nats-edge.plant }
    nats:
      jetstream: false                           # NATS Core: server-sent events only - no publication, WebSocket nor webhook
auth:
  issuers:
    - name: operators
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-nats-streams/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders and metrics, version: 1.0.0 }
channels:
  orderFeed:
    address: "orders.{tenant}.>"                # the wildcard IN the channel, after the segment bound to the tenant
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      any:
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
  siteMetrics:
    address: "sites.{site}.metrics.*"           # bound to a claim of the token: an operator reads its own site
    parameters:
      site: { x-flexway-bind: "claim:site" }
    messages:
      # Written as application/cloudevents+json by a producer outside the gateway: read in structured mode.
      reading:
        payload: { type: object, required: [value], properties: { value: { type: number } } }
  status:
    address: status.public                       # binds nothing, filters nothing: declared public below
    messages:
      state:
        payload: { type: object, required: [state], properties: { state: { type: string } } }
operations:
  receiveOrders: { action: receive, channel: { $ref: "#/channels/orderFeed" }, x-flexway-upstream: orders-nats }
  receiveMetrics: { action: receive, channel: { $ref: "#/channels/siteMetrics" }, x-flexway-upstream: metrics-core }
  receiveStatus:
    action: receive
    channel: { $ref: "#/channels/status" }
    x-flexway-upstream: orders-nats
    x-flexway-public: true                       # read by anyone, anonymous included
```

```yaml
# examples/async-nats-streams/openapi.yaml
openapi: 3.1.0
info: { title: Orders and metrics, version: "1.0.0" }
components:
  securitySchemes:
    operators: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/stream:
    get:
      operationId: streamOrders
      # SSE: one ordered consumer per channel and replica; WebSocket: one durable consumer per subscriber, AckWait = ack_timeout
      x-flexway-subscribe: { operation: receiveOrders, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, history: 64 }
      security: [ { operators: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's orders } }
  /metrics/stream:
    get:
      operationId: streamMetrics
      x-flexway-subscribe: { operation: receiveMetrics, transport: [sse], history: 0 }   # NATS Core: SSE only
      security: [ { operators: [metrics:read] } ]
      responses: { "200": { description: The readings of the caller's site } }
  /status:
    get:
      operationId: streamStatus
      x-flexway-subscribe: { operation: receiveStatus, transport: [sse], max_anonymous_streams: 1000 }
      security: []
      responses: { "200": { description: The public status } }
```

| Rule | Behaviour |
| --- | --- |
| Wildcard | `*` and `>` only in a channel that is read, **after** the bound segments, `>` last - `orders.>.{tenant}` is refused; a bound value never holds `.`, `*` or `>`. The same rule for `+` and `#` in MQTT, and for `*` and `#` in the AMQP 0-9-1 binding key of an exchange (`audit.{tenant}.#`, served since `1.5.0`) |
| SSE | One JetStream ordered consumer (or one Core subscription) per channel and replica, whatever the number of subscribers; `Last-Event-ID` resumed from the channel's history, `id` = the stream sequence |
| WebSocket | A durable consumer per subscriber, `AckWait` = `ack_timeout`; refused at load on NATS Core |
| CloudEvents | Binary (`ce-*`) or structured by content type; the `tenant` attribute filtered in the headers, never in `data` |

## NATS JetStream: a webhook from a durable consumer

> [!NOTE]
> Available since `1.6.0`; the example is validated by the binary and pinned by the tests.

A partner receives **its** tenant's invoices as signed webhooks, with no access to NATS: a durable consumer
**shared** by the replicas (one replica per message), acknowledged after the `2xx`, a declared dead-letter
channel once the retries are spent. No HTTP door: no OpenAPI contract is needed.

<!-- example: examples/async-nats-webhook/flexway.yaml -->

```yaml
# NATS JetStream delivered by webhook (1.6.0): a partner receives its tenant's invoices as
# signed webhooks, without access to NATS - a durable consumer the replicas share (one replica per
# message), acknowledged after the 2xx, the declared dead letter once the retries are spent. No HTTP
# door: no OpenAPI contract. Refused on NATS Core, which redelivers nothing.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  billing-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      nkey_seed: "file:/etc/flexway-secrets/nats/flexway.nk"   # an nkey seed alone, by reference
      jetstream: true                            # a delivery needs JetStream: a durable consumer the replicas share
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/invoices
    secrets: ["file:/etc/flexway-secrets/hooks/current"]
    retries: { max: 8, max_delay: 10m }
```

```yaml
# examples/async-nats-webhook/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Invoices, version: 1.0.0 }
channels:
  invoiceIssued:
    address: "invoices.{tenant}.issued"
    parameters:
      tenant: { x-flexway-bind: tenant }       # a delivery has no caller: bound to the tenant only
    messages:
      issued:
        payload: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } }
operations:
  receiveInvoices:
    action: receive
    channel: { $ref: "#/channels/invoiceIssued" }
    x-flexway-upstream: billing-nats
    x-flexway-deliver:
      - { upstream: partner-hook, tenant: acme, dead_letter: invoices.dead }   # a subject a stream captures
```

On NATS Core, the same delivery is **refused at load**: nothing delivers again a message the target did not take.

## Run

1. Save the `flexway.yaml` shown and the contracts it references in one directory, next to it (`examples/async-nats-jetstream/`, `examples/async-nats-streams/`, `examples/async-nats-webhook/` in the release sources).
2. Validate offline, without a broker or a network - the validator ships in the published image:

   ```bash
   docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway:<version> validate -v
   ```

3. Mount the secrets the file references (`file:/etc/flexway-secrets/...`), then start:

   ```bash
   docker run --rm -p 8080:8080 -p 9090:9090 \
     -v "$PWD:/etc/flexway:ro" -v "$PWD/secrets:/etc/flexway-secrets:ro" \
     ghcr.io/flexwiz/flexway:<version>
   ```

## Verify

1. `validate -v` exits `0` and lists every resolved operation; a refusal names the faulty field and exits `1`.
2. Once started, `curl -s localhost:9090/readyz` answers `200` `{"status":"ready","operations":<n>}`; a missing secret makes the start refuse, naming it.
3. Each row of the case table replays from the developer side: [Publish and subscribe](https://flexway.io/en/docs/developer-async/), [Receive signed webhooks](https://flexway.io/en/docs/developer-webhooks/).