Message contracts in a monorepo: JSON Schema and DTOs

The problem

Our online shop lives in a monorepo. The order service publishes order.order.placed.v1, which stock and shipping consume. Producer, consumers and message format change in the same pull request, and the CI tests them together.

Let us rename customerId to clientId on both sides: every test passes. Yet in production, the change breaks in three places.

The rolling deployment. The old version of stock, still running, receives messages without customerId from the new version of order.
The messages already queued. The new version of stock reads messages written before the update, and finds no clientId in them.
The replays. A message quarantined last week is replayed today, in a format that no code reads any more.

Two format flaws add to this. Added up as floats then encoded as JSON, quantities of 0.1 and 0.2 kilo give 0.30000000000000004. And a field set to null does not say whether it is unknown, empty, erased or forgotten.

The options

What it solves: there is nothing to write, since Messenger applies serialize() to the shared classes by default.
What it costs: the format follows the code. Renaming a property breaks the queued messages, and another language cannot read them.

One JSON Schema per event, classes in each service

What it solves: the format is described outside the code.
What it costs: each service translates the schema in its own way, and nothing checks that the code complies with it.

A lightweight contracts package

What it solves: constants, DTOs and schemas change together, and tests check that they still agree.
What it costs: evolution rules to follow, and two descriptions of the same message to maintain.

A schema registry and generated code

What it solves: the registry refuses a version that is incompatible with the previous one, and the code is generated for each language.
What it costs: one more service to run, and a code generation chain to maintain.

Our recommendation

A shared app-contracts package, additive changes, validation in the producer, tolerant reading in the consumers and frozen fixtures.

The package holds the exchange name, the routing keys, one readonly DTO per event and the JSON Schemas, with no runtime dependency. Queues, binding keys and handlers stay with each consumer, as in the article on the RabbitMQ topology.

One common envelope, one payload per event

The envelope fields:

event_id: a UUID v7 set by the producer in the outbox;
event_name: identical to the routing key;
occurred_at: the moment of the fact, in RFC 3339 with a time offset;
producer: the publishing service;
schema_version: the schema revision, 1.0 then 1.1 after an addition;
correlation_id: links the messages of one flow; for an order, its identifier;
payload: the event data.

Additive changes, a v2 for everything else

Within a version, we only add optional fields. Removing a field, renaming it, changing its type or making it required creates a new event, order.order.placed.v2. It is published alongside the v1 during the transition.

The schemas never set additionalProperties: false. In JSON Schema 2020-12, leaving this keyword out accepts any extra property. A closed schema would turn every addition into a breaking change.

Decimal strings, absence rather than null

Quantities and amounts travel as decimal strings ("2.000"), whose shape the schema fixes. The consumer computes with BCMath, with no float. An optional field without a value is absent, never null: the schema refuses null, and the producer serializes without null values.

Strings, not PHP enums

A PHP enum rejects any unknown value: BackedEnum::from() throws a ValueError, and the Symfony Serializer refuses the data by default. A value added by a newer producer would make the message unreadable for older consumers. Our DTOs therefore carry strings, the known values are constants, and the consumer provides a default case.

Validate in the producer, tolerate in the consumer

The producer validates each message against its JSON Schema while writing its outbox row, in the order's transaction. An invalid message rolls the transaction back: the error shows up where it was caused. The consumer does not validate: it ignores unknown fields and reads only what it needs.

The trade-offs we accept

Two descriptions of the same message. The schema and the DTO change together, and the snapshot tests flag a mismatch. We write the DTOs by hand as long as the contracts remain few.

A versioning discipline. A v2 costs a double publication, then a clean-up. Code review has to spot every breaking change.

A consumer that trusts. A message published by hand, outside the outbox, is checked by nobody. We keep manual publishing for replays.

A format that depends on the library. In 2020-12, format is only an annotation by default. The opis/json-schema library checks it, but accepts a date-time without a time offset: occurred_at therefore also carries a pattern.

The signal that should trigger a review of this choice: repeated mismatches between schemas and DTOs, or a consumer written in another language. Code generation, or even a registry, then pays off.

Implementation

We use PHP 8.3 or later, Symfony 7.4 and opis/json-schema 2.6, which supports draft 2020-12. justinrainbow/json-schema 6.13, also maintained, stops at draft 2019-09.

1. Install the package

The package lives in app-contracts/, at the root of the monorepo. Its composer.json only requires PHP: validator, Serializer, PropertyAccess and PHPUnit are only used by the tests. Each service declares it as a path repository:

{
  "repositories": [{ "type": "path", "url": "../app-contracts" }],
  "require": { "app/contracts": "*@dev" }
}

Composer creates a symbolic link: a change to the package is visible right away. For a Docker image, COMPOSER_MIRROR_PATH_REPOS=1 copies the package into vendor/ instead.

2. Write the schemas

The envelope, in app-contracts/schemas/common/envelope.v1.json:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/app-contracts/schemas/common/envelope.v1.json",
  "type": "object",
  "required": [
    "event_id",
    "event_name",
    "occurred_at",
    "producer",
    "schema_version",
    "correlation_id",
    "payload"
  ],
  "properties": {
    "event_id": { "type": "string", "format": "uuid" },
    "event_name": { "type": "string" },
    "occurred_at": {
      "type": "string",
      "format": "date-time",
      "pattern": "(Z|[+-][0-9]{2}:[0-9]{2})$"
    },
    "producer": { "type": "string", "minLength": 1 },
    "schema_version": { "type": "string", "pattern": "^[0-9]+\\.[0-9]+$" },
    "correlation_id": { "type": "string", "minLength": 1 },
    "payload": { "type": "object" }
  }
}

The event, in app-contracts/schemas/order.order.placed.v1.json, applies the envelope through $ref and details the payload. In 2020-12, other keywords may sit next to $ref.

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/app-contracts/schemas/order.order.placed.v1.json",
  "$ref": "common/envelope.v1.json",
  "properties": {
    "event_name": { "const": "order.order.placed.v1" },
    "schema_version": { "pattern": "^1\\." },
    "payload": {
      "required": ["orderId", "customerId", "lines"],
      "properties": {
        "orderId": { "type": "string", "format": "uuid" },
        "customerId": { "type": "string", "format": "uuid" },
        "lines": {
          "type": "array",
          "minItems": 1,
          "items": {
            "type": "object",
            "required": ["productId", "quantity"],
            "properties": {
              "productId": { "type": "string", "format": "uuid" },
              "quantity": { "type": "string", "pattern": "^[0-9]+\\.[0-9]{3}$" }
            }
          }
        }
      }
    }
  }
}

3. The DTO and the catalogue

// app-contracts/src/Order/OrderPlacedV1.php
namespace AppContracts\Order;

final readonly class OrderPlacedV1
{
    public const SCHEMA_VERSION = '1.0';

    /**
     * @param list<array{productId: string, quantity: string}> $lines
     */
    public function __construct(
        public string $orderId,
        public string $customerId,
        public array $lines,
    ) {
    }
}

// app-contracts/src/EventCatalog.php
namespace AppContracts;

use AppContracts\Order\OrderCancelledV1;
use AppContracts\Order\OrderPlacedV1;

final class EventCatalog
{
    public const SCHEMA_BASE_URI = 'https://example.com/app-contracts/schemas/';
    public const SCHEMA_DIR = __DIR__ . '/../schemas';

    /** Event name => DTO. */
    public const CLASSES = [
        Routing::ORDER_PLACED_V1 => OrderPlacedV1::class,
        Routing::ORDER_CANCELLED_V1 => OrderCancelledV1::class,
    ];
}

4. Validate in the producer's transaction

// src/Messenger/ContractValidator.php (order service)
namespace App\Messenger;

use AppContracts\EventCatalog;
use Opis\JsonSchema\Errors\ErrorFormatter;
use Opis\JsonSchema\Validator;

final class ContractValidator
{
    private readonly Validator $validator;

    public function __construct()
    {
        $this->validator = new Validator();
        $this->validator->resolver()->registerPrefix(EventCatalog::SCHEMA_BASE_URI, EventCatalog::SCHEMA_DIR);
    }

    public function validate(string $eventName, string $json): void
    {
        // opis/json-schema expects objects: no associative array.
        $result = $this->validator->validate(
            json_decode($json, flags: \JSON_THROW_ON_ERROR),
            EventCatalog::SCHEMA_BASE_URI . $eventName . '.json',
        );

        if (!$result->isValid()) {
            throw new \UnexpectedValueException(json_encode((new ErrorFormatter())->format($result->error(), false)));
        }
    }
}

The order service calls it while writing its outbox row:

// src/Outbox/OutboxWriter.php (order service), excerpt of add()
$envelope = [
    'event_id' => $eventId->toRfc4122(),
    'event_name' => $eventName,
    'occurred_at' => $occurredAt->format(\DateTimeInterface::RFC3339_EXTENDED),
    'producer' => 'order',
    'schema_version' => $event::SCHEMA_VERSION,
    'correlation_id' => $correlationId,
    'payload' => $this->normalizer->normalize($event, 'json', [
        AbstractObjectNormalizer::SKIP_NULL_VALUES => true,
    ]),
];

// An invalid message throws: the transaction is rolled back.
$this->contractValidator->validate($eventName, json_encode($envelope, \JSON_THROW_ON_ERROR));

The outbox row then receives this envelope, and the relay publishes it without going through the DTO again.

5. A Messenger serializer for the envelope

On the consumer side, our serializer reads the envelope, finds the class from event_name and writes it again on a retry. Two readonly stamps, implementing StampInterface, carry the envelope: EventIdStamp for event_id, EventMetadataStamp for the rest.

// src/Messenger/EventSerializer.php (stock and shipping)
namespace App\Messenger;

use AppContracts\EventCatalog;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Serializer\Normalizer\AbstractObjectNormalizer;
use Symfony\Component\Serializer\Normalizer\DenormalizerInterface;
use Symfony\Component\Serializer\Normalizer\NormalizerInterface;

final class EventSerializer implements SerializerInterface
{
    public function __construct(
        private readonly NormalizerInterface $normalizer,
        private readonly DenormalizerInterface $denormalizer,
    ) {
    }

    public function encode(Envelope $envelope): array
    {
        $id = $envelope->last(EventIdStamp::class);
        $meta = $envelope->last(EventMetadataStamp::class);
        if (null === $id || null === $meta) {
            throw new \LogicException('Missing event stamps.');
        }

        $body = [
            // The same event_id on every retry: idempotency depends on it.
            'event_id' => $id->eventId,
            'event_name' => $meta->eventName,
            'occurred_at' => $meta->occurredAt,
            'producer' => $meta->producer,
            'schema_version' => $meta->schemaVersion,
            'correlation_id' => $meta->correlationId,
            // Absence rather than null: null fields are not written.
            'payload' => $this->normalizer->normalize($envelope->getMessage(), 'json', [
                AbstractObjectNormalizer::SKIP_NULL_VALUES => true,
            ]),
        ];

        return [
            'body' => json_encode($body, \JSON_THROW_ON_ERROR),
            'headers' => [
                'Content-Type' => 'application/json',
                // Without this counter, every retry would start again from zero.
                'X-Retry-Count' => (string) RedeliveryStamp::getRetryCountFromEnvelope($envelope),
            ],
        ];
    }

    public function decode(array $encodedEnvelope): Envelope
    {
        try {
            $data = json_decode($encodedEnvelope['body'], true, flags: \JSON_THROW_ON_ERROR);
            $class = EventCatalog::CLASSES[$data['event_name'] ?? '']
                ?? throw new \UnexpectedValueException('Unknown event name.');

            // Unknown fields are ignored: the Serializer's default behaviour.
            $message = $this->denormalizer->denormalize($data['payload'], $class, 'json');
            $stamps = [
                new EventIdStamp($data['event_id']),
                new EventMetadataStamp(
                    $data['event_name'],
                    $data['occurred_at'],
                    $data['producer'],
                    $data['schema_version'],
                    $data['correlation_id'],
                ),
            ];
        } catch (\Throwable $e) {
            throw new MessageDecodingFailedException($e->getMessage(), 0, $e);
        }

        $retryCount = (int) ($encodedEnvelope['headers']['X-Retry-Count'] ?? 0);
        if ($retryCount > 0) {
            $stamps[] = new RedeliveryStamp($retryCount);
        }

        return new Envelope($message, $stamps);
    }
}

Messenger counts the attempts with the RedeliveryStamp: a serializer that loses it retries a failing message forever. And on a MessageDecodingFailedException, the AMQP transport rejects the message without requeuing it: it is lost, unless the queue has a dead letter exchange.

Serializer and stamps depend on Messenger: they live in each consuming service, not in app-contracts. On the order side, the relay publishes the stored envelope with its own serializer. They are enabled per transport:

# config/packages/messenger.yaml (stock and shipping), excerpt
framework:
  messenger:
    transports:
      events:
        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
        serializer: App\Messenger\EventSerializer

6. Freeze one fixture per revision

The fixture app-contracts/tests/fixtures/order.order.placed.v1.0.json will not change again:

{
  "event_id": "01a0f6cc-9dc0-73c6-869a-45b14da4f9fc",
  "event_name": "order.order.placed.v1",
  "occurred_at": "2026-10-01T09:30:00.000+00:00",
  "producer": "order",
  "schema_version": "1.0",
  "correlation_id": "01a0f6cc-9dc0-7b8a-9ea5-f190656412a9",
  "payload": {
    "orderId": "01a0f6cc-9dc0-7b8a-9ea5-f190656412a9",
    "customerId": "019cf0d4-bdc0-7eaf-b33a-9c7f4a14876a",
    "lines": [{ "productId": "01a05c4d-d5c0-7051-a329-665966ceab36", "quantity": "2.000" }]
  }
}

// app-contracts/tests/ContractSnapshotTest.php
namespace AppContracts\Tests;

use AppContracts\EventCatalog;
use Opis\JsonSchema\Helper;
use Opis\JsonSchema\Validator;
use PHPUnit\Framework\Attributes\DataProvider;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Serializer\Normalizer\AbstractObjectNormalizer;
use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;
use Symfony\Component\Serializer\Serializer;

final class ContractSnapshotTest extends TestCase
{
    public static function fixtures(): iterable
    {
        foreach (glob(__DIR__ . '/fixtures/*.json') as $file) {
            yield basename($file) => [$file];
        }
    }

    #[DataProvider('fixtures')]
    public function testFrozenFixtureStillFits(string $file): void
    {
        $validator = new Validator();
        $validator->resolver()->registerPrefix(EventCatalog::SCHEMA_BASE_URI, EventCatalog::SCHEMA_DIR);
        $serializer = new Serializer([new ObjectNormalizer()]);

        $message = json_decode(file_get_contents($file), true, flags: \JSON_THROW_ON_ERROR);
        $schema = EventCatalog::SCHEMA_BASE_URI . $message['event_name'] . '.json';

        // 1. The fixture still validates against the schema.
        self::assertTrue($validator->validate(Helper::toJSON($message), $schema)->isValid());

        // 2. It still deserializes into the DTO.
        $dto = $serializer->denormalize($message['payload'], EventCatalog::CLASSES[$message['event_name']]);

        // 3. The re-serialized DTO gives back the same payload.
        $payload = $serializer->normalize($dto, null, [AbstractObjectNormalizer::SKIP_NULL_VALUES => true]);
        self::assertEquals($message['payload'], $payload);

        // 4. And it still validates against the schema.
        $message['payload'] = $payload;
        self::assertTrue($validator->validate(Helper::toJSON($message), $schema)->isValid());
    }
}

Renaming customerId in the DTO and the schema makes this test fail: the 1.0 fixture no longer validates. A fixture field forgotten in the DTO makes it fail at step 3.

7. Add a field, then publish a v2

For the delivery mode, the schema gains an optional deliveryMode property. The DTO moves to revision 1.1 (SCHEMA_VERSION = '1.1') with a last argument ?string $deliveryMode = null, and the known values become constants:

// app-contracts/src/Order/DeliveryMode.php
namespace AppContracts\Order;

final class DeliveryMode
{
    public const STANDARD = 'standard';
    public const EXPRESS = 'express';
}

The fixture order.order.placed.v1.1.json joins the previous one, which stays untouched. In shipping, a missing or unknown value falls back to the default case:

// src/MessageHandler/PrepareShipmentOnOrderPlaced.php (shipping service), excerpt
$shipment->setPriority(match ($event->deliveryMode) {
    DeliveryMode::EXPRESS => Shipment::PRIORITY_HIGH,
    default => Shipment::PRIORITY_NORMAL,
});

A breaking change follows another path:

Create the schema, the DTO, the constant and the fixtures of the v2.
Publish both versions, with two outbox rows in the same transaction.
Bind each consumer to the v2, handler included, then delete the v1 binding key in RabbitMQ: Messenger never removes one. In between, each fact arrives twice, under two event_ids: the handler drops the duplicate by orderId.
Remove the v1 when no queue receives it any more and no v1 message can be replayed.

Checklist

Shared constants, DTOs and schemas; the queues with each consumer.
An event_id set by the producer, in a common envelope.
Optional additions, no additionalProperties: false; a v2 otherwise.
Decimal strings, absence rather than null, no PHP enum.
Validation in the producer, in the outbox transaction.
A serializer that keeps event_id and the retry counter.
One frozen fixture per revision, tested on every build.

Sources

JSON Schema 2020-12, Core (additionalProperties, $ref) and Validation (format).
IETF, RFC 3339, section 5.6 and RFC 9562 (UUID v7).
opis/json-schema: version 2.6.0 of 17 October 2025, documentation and DateTimeFormats.
justinrainbow/json-schema: version 6.13.1 of 30 September 2026.
Symfony, custom Messenger serializer, SendFailedMessageForRetryListener, AmqpReceiver and Connection.
Symfony, Serializer, UID; PHP, BackedEnum::from.
Composer, path repositories and COMPOSER_MIRROR_PATH_REPOS.
Confluent, Schema Evolution and Compatibility.

Message contracts in a monorepo: JSON Schema and DTOs

The problem

The options

One JSON Schema per event, classes in each service

A lightweight contracts package

A schema registry and generated code

Our recommendation

One common envelope, one payload per event

Additive changes, a v2 for everything else

Decimal strings, absence rather than null

Strings, not PHP enums

Validate in the producer, tolerate in the consumer

The trade-offs we accept

Implementation

1. Install the package

2. Write the schemas

3. The DTO and the catalogue

4. Validate in the producer's transaction

5. A Messenger serializer for the envelope

6. Freeze one fixture per revision

7. Add a field, then publish a v2

Checklist

Sources

Related articles

RabbitMQ: one topic exchange and one queue per consuming service

The Outbox pattern with Symfony Messenger: publish without losing events

Idempotent consumer with Symfony Messenger: the processed_event table

Related articles

Event-driven architecture

RabbitMQ: one topic exchange and one queue per consuming service

Several services react to the same event? A single topic exchange, one queue per consuming service and versioned routing keys, with Symfony Messenger.

Updated on Oct 01, 20269 min read

Event-driven architecture

The Outbox pattern with Symfony Messenger: publish without losing events

Writing to the database then publishing can lose an event. The outbox writes it in the same transaction, a relay publishes it: Messenger, Doctrine, PostgreSQL.

Updated on Oct 02, 202611 min read

Event-driven architecture

Idempotent consumer with Symfony Messenger: the processed_event table

The same event can arrive twice. A processed_event table, written in the same transaction as the effect, discards the duplicate with Symfony Messenger.

Updated on Oct 02, 202611 min read

The problem

The options

Share the PHP classes, without a schema

One JSON Schema per event, classes in each service

A lightweight contracts package

A schema registry and generated code

Our recommendation

One common envelope, one payload per event

Additive changes, a v2 for everything else

Decimal strings, absence rather than null

Strings, not PHP enums

Validate in the producer, tolerate in the consumer

The trade-offs we accept

Implementation

1. Install the package

2. Write the schemas

3. The DTO and the catalogue

4. Validate in the producer's transaction

5. A Messenger serializer for the envelope

6. Freeze one fixture per revision

7. Add a field, then publish a v2

Checklist

Sources

Related articles

RabbitMQ: one topic exchange and one queue per consuming service

The Outbox pattern with Symfony Messenger: publish without losing events

Idempotent consumer with Symfony Messenger: the processed_event table