# Exemple : charges Avro et registre de schémas

> Commandes Avro vérifiées sous schéma, lues en JSON ou telles quelles, format filaire du registre de schémas.

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

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

## Charges Avro sous schéma, registre de schémas

Une boutique publie ses commandes en **Avro** depuis HTTP : le corps est décodé sous `order.avsc`,
compilé au chargement, avant d'atteindre Kafka ; un tableau de bord les lit en SSE **en JSON**, un
service les lit en WebSocket **en Avro tel quel**. Le canal d'audit porte le **format filaire de
Confluent** : chaque identifiant de schéma est lu dans le registre, et doit y être enregistré sous le
sujet du tenant (`<tenant>.orders.audit-value`) - un autre tenant ne publie jamais sous ce schéma.

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

```yaml
# Avro orders through the asynchronous plane (1.6.0): an HTTP caller publishes an Avro
# body - checked under its schema, compiled at load -, a subscriber over SSE or a webhook reads it in
# JSON, a WebSocket that declares binary frames reads it as it is. The audit channel carries Confluent's
# wire format: each schema id is read in the registry and must be registered under the tenant's subject.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-brokers:
    protocol: kafka
    brokers: [kafka-0.internal:9092, kafka-1.internal:9092]
schema_registry:
  url: https://registry.internal:8081                # read, never written - the Confluent API
  auth:
    username: flexway
    password: ${REGISTRY_PASSWORD}                    # a reference, resolved at each request
  timeout: 2s                                         # a lookup holds the message that needs it
  max_schemas: 1000                                   # compiled once each, kept by id
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-binary/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Shop orders, version: "1.0.0" }
channels:
  orders:
    address: "{tenant}.orders"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      order:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
  audit:
    address: "{tenant}.orders.audit"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      audited:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
        bindings:
          kafka: { schemaIdLocation: payload, schemaIdPayloadEncoding: confluent, schemaLookupStrategy: TopicIdStrategy }
operations:
  publishOrder:
    action: send
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # decode every body under order.avsc - opt-in for a binary message
    x-flexway-key: /orderId             # read in the payload's JSON form
  receiveOrder:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
  publishAudit:
    action: send
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # the schema id in the registry, under <tenant>.orders.audit-value
  receiveAudit:
    # Records of an ordinary Kafka producer (a Confluent serializer, no CloudEvents header) are read
    # too: the channel is bound to the tenant and its only message is in the registry's wire format,
    # so the gateway builds each envelope - id <topic>/<partition>/<offset>, the record's timestamp.
    action: receive
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
```

```yaml
# examples/async-binary/openapi.yaml
openapi: 3.1.0
info: { title: Shop orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
security: [{ customers: [] }]
paths:
  /orders/events:
    post:
      operationId: emitOrder
      x-flexway-publish: { operation: publishOrder, idempotency: required }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published, after the broker's acknowledgement } }
  /orders/audit:
    post:
      operationId: emitAudit
      x-flexway-publish: { operation: publishAudit }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published } }
  /orders/stream:
    get:
      operationId: streamOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [sse] }
      responses: { "200": { description: the tenant's orders, in JSON } }
  /orders/frames:
    get:
      operationId: framesOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [websocket], frames: binary }
      responses: { "200": { description: the tenant's orders, the Avro payload in binary frames } }
  /orders/audit/stream:
    get:
      operationId: streamAudit
      x-flexway-subscribe: { operation: receiveAudit, transport: [sse] }
      responses: { "200": { description: the tenant's audit records in JSON, ordinary producers' included } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /orders/events`, corps Avro conforme | `202` après l'accusé ; clé Kafka lue en `/orderId` de la forme JSON du message |
| Corps qui ne se décode pas sous `order.avsc` | `422 invalid_request`, la règle seule, jamais une valeur du message |
| Autre `Content-Type` | `422 invalid_request` (la porte ne déclare que l'Avro) ; une porte sans `requestBody` répond `415 unsupported_media_type` |
| `POST /orders/audit`, identifiant enregistré sous `acme.orders.audit-value` | `202` ; schéma lu une fois par identifiant, gardé en mémoire (`max_schemas`) |
| Identifiant inconnu, ou enregistré sous un autre sujet | `422 invalid_request` |
| Registre injoignable ou en erreur | `503 upstream_unavailable` avec `Retry-After` : jamais un message publié sans validation ; coupe-circuit propre au registre |
| `GET /orders/stream` (SSE) | Chaque évènement en JSON (encodage JSON d'Avro), `datacontenttype: application/json` ; un message qui ne se décode pas n'est pas remis (`flexway_async_deliver_total{outcome="invalid"}`) |
| `GET /orders/frames` (WebSocket) | Une trame texte (les attributs, dont l'`id` à accuser), puis une trame **binaire** : la charge Avro telle quelle |
| Webhook (`x-flexway-deliver`) | Corps CloudEvent en JSON transcodé ; registre injoignable : le message n'est ni livré ni accusé, il revient |
| Protobuf | `schemaFormat: "application/vnd.google.protobuf;version=3"` et `schema: { $ref: "./reading.binpb#acme.Reading" }` - un jeu de descripteurs compilé (`protoc --descriptor_set_out --include_imports`, `buf build`), le message par son nom ; une source `.proto` en ligne est refusée |
| CBOR | `contentType: application/cbor` et un schéma JSON : la charge est lue dans le modèle JSON de CBOR (RFC 8949 §6.1) et vérifiée contre ce schéma |
| `GET /orders/audit/stream`, enregistrement d'un **producteur Kafka ordinaire** (format filaire, aucun en-tête CloudEvents) | Remis dans l'enveloppe que construit la passerelle : `id` = `acme.orders.audit/<partition>/<position>` (stable d'une réplique à l'autre, c'est l'`id` SSE et le `Last-Event-ID`, l'`id` à accuser en WebSocket, le `webhook-id`), `source` = `/flexway/orders-brokers/audit`, `type` = `audited`, `time` = l'horodatage Kafka, `datacontenttype: application/json` après transcodage ; **aucun** attribut `tenant` : le tenant est celui du canal lié, jamais un en-tête ni le contenu du message |
| Enregistrement sans attribut sur un autre canal, ou qui porte un attribut `ce_*` sans en être un | Pas un évènement, comme en `1.5.0` : non remis (`flexway_async_deliver_total{outcome="invalid"}`) |
| Schéma d'écriture du registre différent du schéma du contrat | **Résolu** vers le schéma du contrat par les règles standard d'Avro (champ ajouté avec défaut, champ retiré, `int` -> `long`, alias, symbole d'énumération par défaut) : même JSON qu'un message écrit sous le schéma du contrat ; résolution une fois par paire (identifiant, schéma du contrat), `max_schemas` paires gardées. Schémas identiques (même forme canonique) : aucun coût de plus |
| Schéma d'écriture qui ne se résout pas vers celui du contrat (champ sans défaut, type incompatible) | Refusé entier, jamais livré en partie : `422 invalid_request` à la publication ; à la consommation, non remis comme une charge qui ne se décode pas - SSE au plus une fois, WebSocket accusé, webhook mis au rebut -, la consommation continue ; compté `outcome="incompatible"` |
| `x-flexway-key` sur un message binaire | La clé est lue dans la forme **JSON** du message : **un décodage par publication**, même avec `x-flexway-validation: off` (une charge qui ne se décode pas est alors refusée en `422`) - un coût à peser sur un débit élevé ; sans clé, une charge non validée est relayée sans être lue |

> [!WARNING]
> Refusé au chargement : un `schemaFormat` inconnu (RAML, OpenAPI, Avro `2.x`), un schéma Avro
> invalide, récursif ou dont un tableau contient des éléments sans octet, un message absent du jeu de
> descripteurs, un `$ref` de schéma binaire hors du répertoire du document AsyncAPI, le format filaire
> sans `schema_registry` ou sur un courtier qui n'est pas Kafka, `x-flexway-validation: off` sur un
> message JSON (toujours vérifié), `frames: binary` sans WebSocket ou sur un message JSON, une
> consommation d'un canal au format filaire isolée par le seul attribut `tenant` (ni liée, ni publique) :
> un producteur ordinaire n'écrit aucun attribut, et le tenant n'est jamais lu dans un message.

### Mesurer

 `flexway_async_payloads_total{operation, format, step, outcome}` - `step` vaut `validate`
ou `transcode`, `outcome` `ok`, `invalid`, `not_attached`, `incompatible` ou `unavailable` ; jamais une valeur du
message ni un identifiant de schéma.

## 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-binary/` 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/...`, variables d'environnement), 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" `-e REGISTRY_PASSWORD` \
     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/).