Adding a custom event and handler

Evolve uses event-driven messaging for asynchronous communication between services. This guide walks through adding a new internal event type, publishing it from a service, and consuming it in a handler.

Define the event schema

Events are defined as JSON Schema files in the shared schemas/ directory at the repository root. Zod validators are generated from them per service:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "https://schemas.evolve.labdigital.nl/events/inventory-low",
  "title": "InventoryLowEvent",
  "type": "object",
  "allOf": [
    { "$ref": "https://schemas.evolve.labdigital.nl/base.schema.json" }
  ],
  "properties": {
    "type": { "type": "string", "const": "inventory-low" },
    "sku": { "type": "string" },
    "availableQuantity": { "type": "number" },
    "threshold": { "type": "number" },
    "storeKey": { "type": "string" }
  },
  "required": ["type", "sku", "availableQuantity", "threshold", "storeKey"]
}

The base.schema.json reference adds the shared envelope fields (type, origin, timestamp). Register the schema URI in the publishing service’s codegen.types.ts and run pnpm codegen — this generates the inventoryLowEventSchema Zod validator and InventoryLowEvent type into the service’s src/generated/events.ts via @evolve-framework/json-schema-to-zod.

Publish the event

Use resolvePublisher from @evolve-packages/cloud-adapters to obtain a MessagePublisher for the configured target. It inspects the target URL to select the correct cloud provider (EventBridge/SQS, Event Grid/Service Bus, Pub/Sub, or Redis for local development):

import { resolvePublisher } from "@evolve-packages/cloud-adapters";
import { inventoryLowEventSchema } from "#src/generated/events.ts";

const publisher = resolvePublisher(config.INTERNAL_EVENTS_TARGET);

const event = inventoryLowEventSchema.parse({
  type: "inventory-low",
  sku: variant.sku,
  availableQuantity: variant.quantity,
  threshold: 10,
  storeKey: storeContext.storeKey,
  origin: "catalog-commercetools",
  timestamp: new Date().toISOString(),
});

await publisher.publish(event, {
  type: event.type,
  origin: event.origin,
  messageGroupId: variant.sku,
});

The type and origin are passed as message attributes for routing. The messageGroupId ensures ordering per SKU on transports that support it.

Create the event handler

Add a handler function that processes the event. The handler receives the event payload directly (not wrapped in a .data property). Use a switch on event.type to dispatch:

import type { CloudEventsPayload } from "@commercetools/platform-sdk";

const handleInternalEvent = async (event: CloudEventsPayload) => {
  switch (event.type) {
    case "inventory-low":
      return handleInventoryLow(event);
    // ... other event types
  }
};

const handleInventoryLow = async (event) => {
  await notifyOpsTeam(event.sku, event.availableQuantity);
};

Wire the cloud trigger

Each cloud adapter package has a handler factory that wraps the trigger into the common callback signature:

AWS (SQS + Lambda):

import { createSQSHandler } from "@evolve-framework/cloud-adapter-aws";
import { lambdaHandlerFactory } from "@evolve-framework/core/observability/lambda";

const sqsHandler = createSQSHandler(handleInternalEvent);
export const handler = lambdaHandlerFactory(
  "Internal event handler",
  () => sqsHandler,
);

Azure (Service Bus):

import { createServiceBusHandler } from "@evolve-framework/cloud-adapter-azure";

app.serviceBusQueue(`${config.COMPONENT_NAME}-events`, {
  handler: createServiceBusHandler(handleInternalEvent),
  connection: "SERVICE_BUS_CONNECTION_STRING",
  queueName: config.QUEUE_NAME,
  cardinality: "one",
});

GCP (Pub/Sub):

import { createPubSubHandler } from "@evolve-framework/cloud-adapter-gcp";

const pubsubHandler = createPubSubHandler((event) => {
  return handleInternalEvent(event as CloudEventsPayload);
});

await pubsubHandler.start({
  path: "/internal-events",
  httpHost: config.HTTP_HOST,
  httpPort: config.HTTP_PORT,
});

Configure the queue in Terraform

Create the cloud queue and subscribe it to the internal event bus. For AWS:

resource "aws_sqs_queue" "inventory_events" {
  name = "${var.component_name}-inventory-events"
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.inventory_events_dlq.arn
    maxReceiveCount     = 4
  })
}

Always include a dead-letter queue for messages that fail after the maximum number of delivery attempts.

Test locally

For local development, the framework uses Redis as its transport. The startDevelopmentListener polls the queue with brPop so you can test the full publish/consume cycle without cloud infrastructure:

import { startDevelopmentListener } from "@evolve-framework/core/messaging";

startDevelopmentListener(
  config.REDIS_URL,
  async (event) => {
    await handleInternalEvent(event as CloudEventsPayload);
  },
  { continueOnError: true, dlq: `${config.COMPONENT_NAME}-dlq` },
);

Adding a custom event and handler

Further reading