Concepts

Contract-Driven Topology

Hoppity’s core idea: handlers and publish declarations ARE the topology. You declare what your service handles and what it sends. deriveTopology generates all Rascal config automatically — exchanges, queues, bindings, publications, and subscriptions.

defineDomain() → onEvent/onCommand/onRpc() → hoppity.service() → .build()

No topology files. No string-literal exchange/queue names scattered across services. The contracts carry all the naming conventions, and the topology is derived mechanically.

Domain Contracts

A domain groups related events, commands, and RPC operations with Zod schemas:

import { z } from "zod";
import { defineDomain } from "@apogeelabs/hoppity";

export const OrdersDomain = defineDomain("orders", {
    events: {
        orderCreated: z.object({ orderId: z.string(), total: z.number() }),
    },
    commands: {
        cancelOrder: z.object({ orderId: z.string() }),
    },
    rpc: {
        getOrderSummary: {
            request: z.object({ orderId: z.string() }),
            response: z.object({ orderId: z.string(), status: z.string() }),
        },
    },
});

Each contract object carries:

• Schema — Zod type for runtime validation and compile-time inference
• Exchange name — {domain} for events/commands, {domain}_rpc for RPC
• Routing key — {domain}.{type}.{snake_name} (e.g., orders.event.order_created)
• Publication/subscription names — derived mechanically from domain, type, and operation name

Contracts are shared between services. Both producer and consumer import the same contract objects.

Handlers

Handlers declare what your service processes. Each is bound to a contract and registered in ServiceConfig.handlers:

import { onEvent, onCommand, onRpc } from "@apogeelabs/hoppity";

// Event handler — reacts to broadcasts
const handler = onEvent(OrdersDomain.events.orderCreated, async (content, context) => {
    // content is typed from the contract's schema
});

// Command handler — processes directed work
const handler = onCommand(OrdersDomain.commands.cancelOrder, async ({ orderId }, { broker }) => {
    await broker.publishEvent(OrdersDomain.events.orderCancelled, { orderId });
});

// RPC handler — responds synchronously
const handler = onRpc(OrdersDomain.rpc.getOrderSummary, async ({ orderId }) => {
    return { orderId, status: "active" }; // return type enforced by response schema
});

All handlers accept optional HandlerOptions for queue type, redelivery limits, and dead-letter configuration:

const handler = onCommand(OrdersDomain.commands.cancelOrder, handlerFn, {
    queueType: "quorum", // defaults to "quorum"
    redeliveries: { limit: 10 }, // defaults to { limit: 5 }
    deadLetter: { exchange: "my-dlx" },
});

Seven Build Phases

When you call .build(), seven phases execute in order:

1. Topology Derivation

deriveTopology() generates a complete Rascal BrokerConfig from handler declarations and publish contracts. Every handler produces a queue, binding, and subscription. Every publish contract produces a publication (and the exchange, if not already declared by a handler).

2. Topology Merge

mergeTopology(rawTopology, derived) combines the optional raw topology from ServiceConfig (as base) with the derived config. This is the escape hatch — use it for one-off infrastructure that can’t be derived (DLX exchanges, shovel config, etc.).

3. Middleware Pipeline (synchronous)

Each middleware function runs sequentially. It receives:

• topology — the current BrokerConfig, after derivation and merge
• context — a shared MiddlewareContext with logger, mutable data store, and middleware name tracking

Each returns a MiddlewareResult:

{
    topology: BrokerConfig;                                        // Required — the (possibly modified) topology
    onBrokerCreated?: (broker: BrokerAsPromised) => Promise<void>; // Optional lifecycle hook
}

Middleware functions are synchronous. Async work goes in onBrokerCreated.

4. Broker Creation

The final accumulated topology is passed to Rascal’s BrokerAsPromised.create(). This is where the AMQP connection is established and all exchanges, queues, and bindings are declared in RabbitMQ.

5. Handler Wiring

wireHandlers() subscribes event and command handlers to their queues. wireRpcHandlers() subscribes RPC responders. Auto-ack on success, nack without requeue on error.

6. Outbound Wiring

wireOutbound() attaches publishEvent and sendCommand to the broker. If any RPC handlers or callers exist, the reply queue subscription is set up and request / cancelRequest are attached. broker.shutdown() is wrapped to drain pending RPC requests.

7. Callback Execution (async, sequential)

Each onBrokerCreated callback runs in the order its middleware was registered. If any callback throws, the broker is shut down before the error propagates.

MiddlewareContext

The shared context flows through the entire pipeline:

interface MiddlewareContext {
    data: Record<string, any>; // Mutable shared state for inter-middleware communication
    middlewareNames: string[]; // Names of executed middleware (in order)
    logger: Logger; // Logger instance (defaults to ConsoleLogger)
    serviceName?: string; // The service name passed to hoppity.service()
}

• context.data — Any middleware can read/write here. Useful for passing configuration between middleware.
• context.logger — All middleware should use this for logging. Set via logger in ServiceConfig — active before any middleware runs.
• context.serviceName — The service name, populated automatically by the builder.

Middleware

Middleware is for cross-cutting concerns only — things that apply across the service but aren’t specific to any one handler or contract. Observability, custom topology augmentation.

Handlers, subscriptions, RPC, and typed broker methods are all built into core. You do not need middleware for any of those.

Custom Logger

Pass your logger directly in ServiceConfig — it’s the simplest approach and has no ordering concerns:

const broker = await hoppity
    .service("order-service", {
        connection: { url: "amqp://localhost" },
        handlers: [cancelOrderHandler],
        logger: myLogger, // active before any middleware runs
    })
    .build();

Writing Custom Middleware

A middleware is a function (or a factory that returns one) matching the MiddlewareFunction signature:

import type { MiddlewareFunction } from "@apogeelabs/hoppity";

function withMyFeature(options: MyOptions): MiddlewareFunction {
    return (topology, context) => {
        const modified = structuredClone(topology);
        // Modify topology...

        context.data.myFeature = {
            /* share state downstream */
        };
        context.logger.info("My feature middleware executed");

        return {
            topology: modified,
            onBrokerCreated: async broker => {
                // Async setup after broker creation...
            },
        };
    };
}

Key rules:

• Clone before modifying — Use structuredClone(topology) to avoid mutating the input
• Synchronous body — The middleware function itself must be synchronous. Async work goes in onBrokerCreated
• Return topology — Always return { topology } even if you didn’t modify it
• Use context.logger — Not console.log. Respect the pipeline’s logger.

Interceptors

Interceptors wrap handler and publish execution for per-message cross-cutting concerns — tracing, metrics, header injection. They complement middleware: middleware runs once during setup, interceptors run for every message.

const withHandlerTiming: Interceptor = {
    name: "handler-timing",
    inbound: (handler, meta) => async (payload, ctx) => {
        const start = performance.now();
        try {
            return await handler(payload, ctx);
        } finally {
            console.log(
                `${meta.contract._name} handled in ${(performance.now() - start).toFixed(2)}ms`
            );
        }
    },
};

const broker = await hoppity
    .service("order-service", {
        connection: { url: "amqp://localhost" },
        handlers: [cancelOrderHandler],
        publishes: [OrdersDomain.events.orderCreated],
        interceptors: [withHandlerTiming],
    })
    .build();

For composition mechanics, metadata types, RPC support, and the full comparison with middleware, see the Interceptors guide.

Naming Conventions

Topology artifact names are derived mechanically from domain contracts. You don’t set them — the contracts carry them.

Artifact	Pattern	Example
Exchange (event/command)	{domain}	orders
Exchange (rpc)	{domain}_rpc	orders_rpc
Routing key (event)	{domain}.event.{snake_name}	orders.event.order_created
Routing key (command)	{domain}.command.{snake_name}	orders.command.cancel_order
Routing key (rpc)	{domain}.rpc.{snake_name}	orders.rpc.create_order
Queue	{service}_{domain}_{type}_{snake_name}	catalog-service_orders_event_order_created
Publication name	{domain}_{type}_{snake_name}	orders_event_order_created
Subscription name	{domain}_{type}_{snake_name}	orders_event_order_created
Reply queue	{service}_{instanceId}_reply	order-service_abc123_reply

camelCase operation names are converted to snake_case. Acronyms collapse to a single segment: getHTTPResponse becomes get_http_response.

The Escape Hatch

For services not using contracts, or for one-off infrastructure that can’t be derived, pass raw Rascal BrokerConfig directly:

const broker = await hoppity
    .service("legacy-service", {
        connection: { url: "amqp://localhost" },
        topology: existingRascalConfig, // raw BrokerConfig, merged as base
    })
    .build();

You can combine raw topology with derived topology. The raw config is the base; derived topology layers on top:

const broker = await hoppity
    .service("order-service", {
        connection: { url: "amqp://localhost" },
        handlers: [cancelOrderHandler],
        publishes: [OrdersDomain.events.orderCancelled],
        topology: {
            // DLX exchange not derived automatically — add it manually
            vhosts: {
                "/": {
                    exchanges: { "order-service-dlx": { type: "topic" } },
                },
            },
        },
    })
    .build();