# Exemples : NATS et JetStream

> NATS : publier dans JetStream depuis HTTP ou MCP, lire en SSE et WebSocket liés au tenant, livrer en webhook.

Exemple - Depuis la version 1.6.0 - Flexway v1.7.1 - https://flexway.io/docs/operations-examples-nats/

> Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : [guide produit](https://flexway.io/docs/product-overview/) ; l'appeler : [guide développeur](https://flexway.io/docs/developer-overview/).

## NATS JetStream : publier depuis HTTP ou un outil MCP

> [!NOTE]
> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un service HTTP ou un agent publie dans un flux JetStream **sans client NATS** ; `202` seulement après
le `PubAck` du flux, l'`Idempotency-Key` devenue `Nats-Msg-Id` - le flux dédoublonne un renvoi après un `504`.
La passerelle ne crée **aucun** flux : le sujet doit être capturé par un flux existant, sinon `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 } }
```

| Ce qui se passe | NATS JetStream |
| --- | --- |
| `POST /orders/events` | Admission, corps vérifié, CloudEvents en mode binaire (en-têtes `ce-*`), publication sur `orders.acme.created`, `202` après le `PubAck` |
| Le même appel en outil MCP (`emit_order_created`) | Même décision, même accusé |
| `jetstream: false` (NATS Core) | > [!WARNING]
> Refusé au chargement : Core n'accuse rien, un `202` mentirait |
> | Aucun flux ne capture le sujet | `502 upstream_error` (« no responders »), jamais un `202` |

## NATS : SSE et WebSocket liés au tenant, jokers, NATS Core

> [!NOTE]
> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un tableau de bord lit **tous** les évènements de commande de son tenant d'un seul flux, grâce à un joker
**inclus dans le canal** après le segment lié ; un opérateur lit les mesures de **son** site sur un NATS Core
(SSE seul, au plus une fois) ; un écran public lit l'état sans jeton. Chaque message reçu est comparé au motif du canal,
segment par segment : un courtier qui en enverrait trop ne fait jamais fuir un 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 } }
```

| Règle | Comportement |
| --- | --- |
| Joker | `*` et `>` seulement dans un canal lu, **après** les segments liés, `>` en dernier - `orders.>.{tenant}` est refusé ; une valeur liée ne contient jamais `.`, `*` ni `>`. Même règle pour `+` et `#` en MQTT, et pour `*` et `#` dans la clé de liaison AMQP 0-9-1 d'un échange (`audit.{tenant}.#`, servi depuis `1.5.0`) |
| SSE | Un consommateur ordonné JetStream (ou un abonnement Core) par canal et par réplique, quel que soit le nombre d'abonnés ; reprise `Last-Event-ID` dans l'historique du canal, `id` = séquence du flux |
| WebSocket | Un consommateur durable par abonné, `AckWait` = `ack_timeout` ; refusé au chargement sur NATS Core |
| CloudEvents | Binaire (`ce-*`) ou structuré selon le type de contenu ; l'attribut `tenant` filtré dans les en-têtes, jamais dans `data` |

## NATS JetStream : webhook depuis un consommateur durable

> [!NOTE]
> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un partenaire reçoit les factures de **son** tenant en webhook signé, sans accès à NATS : un consommateur
durable **partagé** par les répliques (une réplique par message), accusé après le `2xx`, rebut déclaré au bout
des réessais. Aucune porte HTTP : le contrat OpenAPI n'est pas nécessaire.

<!-- 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
```

Sur NATS Core, la même livraison est **refusée au chargement** : rien ne relivre un message que la cible n'a pas pris.

## Lancer

1. Enregistrer dans un répertoire le `flexway.yaml` montré et les contrats qu'il référence, à côté de lui (`examples/async-nats-jetstream/`, `examples/async-nats-streams/`, `examples/async-nats-webhook/` dans les sources de la release).
2. Valider hors ligne, sans courtier ni réseau - le validateur est dans l'image publiée :

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

3. Monter les secrets que le fichier référence (`file:/etc/flexway-secrets/...`), puis démarrer :

   ```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>
   ```

## Vérifier

1. `validate -v` sort en `0` et liste chaque opération résolue ; un refus nomme le champ fautif et sort en `1`.
2. Une fois démarrée, `curl -s localhost:9090/readyz` répond `200` `{"status":"ready","operations":<n>}` ; un secret absent fait refuser le démarrage, qui le nomme.
3. Chaque ligne du tableau des cas se rejoue avec le développeur : [Publier et s'abonner](https://flexway.io/docs/developer-async/), [Recevoir des webhooks signés](https://flexway.io/docs/developer-webhooks/).