# Exemple : Kafka

> Kafka derrière la passerelle : publier depuis HTTP ou MCP, diffuser en SSE ou WebSocket, livrer en webhook signé.

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

> 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/).

## Kafka : publier, diffuser, livrer

Le parcours complet de [Publier et consommer sur un courtier](https://flexway.io/docs/operations-async/) sur Kafka : une porte de publication, aussi
outil MCP ; un flux SSE en éventail ou WebSocket accusé, lié au tenant ; les évènements d'`acme` livrés en
webhook signé, le rebut dans `orders.dead`. La passerelle s'authentifie en SASL SCRAM sur TLS.

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

```yaml
# Kafka behind the gateway's controls (1.5.0): an HTTP door and an MCP tool publish, a
# browser streams its tenant's events in SSE or WebSocket, a partner receives them as signed webhooks.
# flexway.yaml lists no channel - they come from the AsyncAPI contract.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml                      # AsyncAPI 3.0.0 or 3.1.0
upstreams:
  events:
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093]
    tls: { server_name: kafka.internal }          # TLS, system roots; ca_file, cert_file and key_file for a private CA or mTLS
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 2s                                  # one publication's budget, the acknowledgement included
    max_in_flight: 512                           # bounded buffer: past it, 503 overloaded
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/orders
    secrets: ["file:/etc/flexway-secrets/hooks/current", "file:/etc/flexway-secrets/hooks/next"]
    retries: { max: 8, max_delay: 10m }
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every channel below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp
```

```yaml
# examples/async-kafka/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders events, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"           # one topic per tenant
    parameters:
      tenant: { x-flexway-bind: tenant }         # bound to the verified identity, never read from the request
    messages:
      created:
        contentType: application/json
        payload:
          type: object
          required: [orderId]
          properties:
            orderId: { type: string, minLength: 1 }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-key: /orderId                      # ordered per key, and per key only
  receiveOrderCreated:
    action: receive
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-deliver:                           # acme's events to its webhook; what fails ends in orders.dead
      - { upstream: partner-hook, tenant: acme, dead_letter: orders.dead }
```

```yaml
# examples/async-kafka/openapi.yaml
openapi: 3.1.0
info: { title: Orders events, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      summary: Publish an order-created event for the caller's tenant
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }
      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, minLength: 1 } } }
      responses: { "202": { description: Acknowledged by the broker } }
  /orders/stream:
    get:
      operationId: streamOrderCreated
      x-flexway-subscribe: { operation: receiveOrderCreated, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, buffer: 256, history: 64 }
      security: [ { customers: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's events } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /orders/events` ou l'outil `emit_order_created` | `202` après `acks=all` ; `Idempotency-Key` exigée (`428` sinon), devenue l'`id` ; ordre par `orderId` |
| `GET /orders/stream` en SSE | Un consommateur par canal et par réplique ; `Last-Event-ID` repris dans les 64 derniers messages, `412 precondition_failed` au-delà |
| `GET /orders/stream` en WebSocket | Groupe propre à l'abonné, `{"ack": "<id>"}` dans les 30 s, sinon fermeture `1008` et nouvelle livraison |
| Livraison à `partner-hook` | Groupe partagé par les répliques, Standard Webhooks, position validée après le `2xx`, rebut `orders.dead` |
| mTLS au lieu de SASL | `tls.cert_file` et `tls.key_file` montés, sans `sasl` |

## 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-kafka/` 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/).