Echoes

Echoes in Orionjs provide a structured way to implement event-driven architecture and handle inter-service communication. Using the @Echoes() and @EchoRequest() or @EchoEvent() decorators, you can easily create handlers for both synchronous requests and asynchronous events.

Standalone core and Orion integration

@orion-js/echoes is framework-independent. Its required dependency graph does not include @orion-js/schema, @orion-js/services, @orion-js/http, @orion-js/logger, or any other Orion runtime package. Orion applications keep their existing behavior through @orion-js/echoes-orion. It connects Echoes to Orion schemas, dependency injection, async logger context, HTTP routes, and Orion error classes:

bun add @orion-js/echoes @orion-js/echoes-orion

Import the adapter once, before modules that declare Echoes controllers:

import '@orion-js/echoes-orion'

Applications using @orion-js/components do not need this explicit import because Components installs the adapter automatically.

Schema execution is still runtime behavior, not just TypeScript inference. For every handler, Echoes cleans and validates params before calling resolve, then cleans returns before sending the result. The Orion adapter delegates those operations to @orion-js/schema, so existing custom cleaners, defaults, validations, and ValidationError values keep working.

Creating Echo Controllers

An echoes controller is a class decorated with @Echoes() that contains methods decorated with @EchoRequest() or @EchoEvent():

import { EchoRequest, EchoEvent, Echoes, createEchoRequest, createEchoEvent } from '@orion-js/echoes'
import { Inject } from '@orion-js/services'
import { ExampleRepository } from '../repos/ExampleRepository'
import { schemaWithName, InferSchemaType } from '@orion-js/schema'

// Define schema for parameters and return type
export const ExampleSchema = schemaWithName('ExampleSchema', {
  _id: { type: String },
  name: { type: String },
  createdAt: { type: Date }
})

// Infer TypeScript type from schema
export type ExampleType = InferSchemaType<typeof ExampleSchema>

@Echoes()
export class GetDataEchoes {
  @Inject(() => ExampleRepository)
  private exampleRepository: ExampleRepository

  @EchoRequest()
  getDataById = createEchoRequest({
    params: {
      exampleId: { type: String }
    },
    returns: ExampleSchema,
    resolve: async (params) => {
      return await this.exampleRepository.getExampleById(params.exampleId)
    }
  })
}

Echo Request Handlers

Use the @EchoRequest() decorator with the createEchoRequest() function to define methods that handle synchronous requests from other services:

@EchoRequest()
getUserById = createEchoRequest({
  params: {
    userId: { type: String }
  },
  returns: UserSchema,
  resolve: async (params) => {
    return await this.userRepository.findById(params.userId)
  }
})

Echo Event Handlers

Use the @EchoEvent() decorator with the createEchoEvent() function to define methods that process asynchronous events:

@Echoes()
export class UserEventsEchoes {
  @Inject(() => EmailService)
  private emailService: EmailService

  @EchoEvent()
  userRegistered = createEchoEvent({
    params: {
      user: { type: UserSchema }
    },
    resolve: async (params) => {
      await this.emailService.sendWelcomeEmail(params.user.email)
    }
  })
}

Event Decorator Options

The createEchoEvent() function accepts options similar to createEchoRequest():

@EchoEvent()
processSomething = createEchoEvent({
  attemptsBeforeDeadLetter: 5,
  // Increase when changing persisted Pulse subscription settings
  configVersion: 1,
  params: {
    // schema definition
  },
  resolve: async (params) => {
    // implementation
  }
})

Pulse persists each consumerGroup + topic configuration. Increase the listener’s integer configVersion whenever persisted retry or offset settings change:

@EchoEvent()
processSomething = createEchoEvent({
  configVersion: 2,
  resolve: async params => {
    // implementation
  },
})

The highest version wins during a deploy, so an older replica cannot restore an earlier setting. Using the same version with different settings fails fast. Existing callbacks finish normally and new work converges on the winning configuration. Pulse listeners keep queue state, leases, retries, and completed attempts atomically on their delivery documents. Configure maxConcurrency on an event to override events.pulse.subscription.maxConcurrency for that receiver.

Batch event handlers with Pulse

Use @EchoBatchEvent() when one receiver invocation should process multiple events together:

@Echoes()
export class InvoiceEventsEchoes {
  @EchoBatchEvent({
    configVersion: 2,
    batchSize: 100,
    maxConcurrency: 2,
  })
  async invoiceCreated(events) {
    await invoices.insertMany(
      events.map(({params, context}) => ({
        eventId: context.eventId,
        ...params,
      })),
    )
  }
}

Each array element contains its cleaned params and individual context, including eventId, headers, creation time, and attempt number. batchSize is a maximum; Pulse immediately invokes the handler with whatever is available and never waits to fill the array. One resolveBatch invocation consumes one worker and one maxConcurrency slot. The complete array is one retry and acknowledgement unit. If the handler throws, Pulse retries the whole delivery, so bulk side effects should use each context.eventId as an idempotency key. Batch event receivers are Pulse-only and require a transaction-capable MongoDB deployment. Kafka consumption fails at startup when a batch receiver is configured. For a hot rollout, first deploy the new handler and packages with batchSize: 1. After every replica understands batch deliveries, increase configVersion and set the intended size. Processes using a normal handler can consume a multi-event delivery sequentially, while batch handlers naturally receive an older one-event delivery as an array of length one.

Making Requests

To make a request to another service:

import { request } from '@orion-js/echoes'

// In a service or resolver
async function getUserDetails(userId: string): Promise<UserDetails> {
  return await request({
    service: 'users',           // Target service name
    method: 'getUserById',      // Method name in the target service
    params: { userId },         // Parameters to pass
    timeout: 5000,              // Optional timeout in milliseconds
    retries: 3                  // Optional number of retries
  })
}

Publishing Events

To publish an event for other services to consume:

import { publish } from '@orion-js/echoes'

// In a service after creating a user
async function createUser(userData: UserInput): Promise<User> {
  const user = await this.userRepository.create(userData)
  
  // Publish an event
  await publish({
    topic: 'userRegistered',    // Event topic
    params: { user },           // Event payload
    acks: 1,                    // Optional: number of acknowledgments
    timeout: 3000               // Optional: timeout in milliseconds
  })
  
  return user
}

Starting the Echoes Service

To enable echoes in your application, you need to configure and start the echoes service:

import { startService } from '@orion-js/echoes'

// Start the echoes service
await startService({
  echoes,

  // Kafka client configuration (for events)
  client: {
    clientId: 'my-app',
    brokers: ['kafka:9092']
  },
  
  // Request configuration (for synchronous communication)
  requests: {
    key: 'shared-secret-key',      // Secret key for request signing
    handlerPath: '/echoes-services', // Path for HTTP handlers
    services: {
      users: 'http://users-service:3000',
      payments: 'http://payments-service:3000'
    }
  },
  
  // Advanced options
  readTopicsFromBeginning: true,    // Read missed messages when reconnecting
  partitionsConsumedConcurrently: 4 // Number of partitions to consume concurrently
})

Using Echoes without Orion

A standalone service can use a SimpleSchema-compatible object directly. Echoes recognizes objects with clean() and validate() methods and executes both at runtime. typedEchoesSchema() adds the value type when the schema library does not expose it in a form TypeScript can infer:

import SimpleSchema from 'simpl-schema'
import {
  createEchoRequest,
  startService,
  typedEchoesSchema,
} from '@orion-js/echoes'

const params = typedEchoesSchema<{email: string}>(
  new SimpleSchema({
    email: {type: String},
  }),
)

const echoes = {
  findUser: createEchoRequest({
    params,
    async resolve({email}) {
      return {email}
    },
  }),
}

For synchronous Echoes requests, connect the framework-neutral receiver to the HTTP server of your choice:

import express from 'express'

const app = express()

await startService({
  echoes,
  requests: {
    key: process.env.ECHOES_PASSWORD,
    registerHandler(definition) {
      app.post(definition.path, express.json({limit: definition.bodyLimit}), async (req, res) => {
        res.json(await definition.handle(req.body))
      })
    },
  },
})

Kafka and Pulse are optional transports and are loaded only when selected. Install kafkajs for Kafka, or @orion-js/pulse with its MongoDB peer dependency for Pulse. A service that only uses synchronous requests loads neither event transport.

Incremental migration from Kafka to Pulse

Echoes can listen through Kafka and Pulse at the same time while publishing through exactly one transport. This allows the migration to happen in three independent deployments, without dual-publishing an event. Install Pulse and its MongoDB peer dependency in the application:

bun add @orion-js/pulse mongodb

1. Prepare every listener

Keep the existing top-level Kafka configuration and add Pulse under events. Deploy this version to every instance before changing any publisher:

import {startService} from '@orion-js/echoes'

await startService({
  echoes,

  // Existing Kafka configuration remains valid.
  client: {
    clientId: 'billing',
    brokers: ['kafka:9092'],
  },
  consumer: {
    groupId: 'billing',
  },
  producer: {},

  events: {
    pulse: {
      connectionString: process.env.MONGO_URL,
      databaseName: 'events',
      // Replicas of the same service must share this value.
      consumerGroup: 'billing',
    },
    consumeFrom: ['kafka', 'pulse'],
    publishTo: 'kafka',
  },
})

At this point, handlers accept events from both systems, but publish() still sends only to Kafka. The event handler API does not change. The Pulse transport uses a MongoDB application pool of at most one connection per service instance by default. Callbacks remain concurrent because they do not retain the pool socket. Set events.pulse.maxPoolSize only when a specific high-throughput service has measured pool contention. events.pulse.maxIdleTimeMS defaults to 30 seconds and lets larger pools contract after bursts. events.pulse.discoveryLockTimeoutMs independently tunes discovery-leader failover; handler locks continue using events.pulse.lockTimeoutMs.

Pulse no longer supports events.pulse.changeStreams. Event discovery always uses polling. Remove the legacy field before deploying this configuration; Echoes fails during startup if the field is still present, including when its value is "disabled".

2. Switch publishers

Once every listener is running the dual-consumer configuration, change only publishTo:

events: {
  pulse: {
    connectionString: process.env.MONGO_URL,
    databaseName: 'events',
    consumerGroup: 'billing',
  },
  consumeFrom: ['kafka', 'pulse'],
  publishTo: 'pulse',
}

This can be rolled out gradually. Old application instances continue publishing to Kafka and new instances publish to Pulse; the listener fleet consumes both. For a controlled switch, publishTo can also come from an environment flag:

const publishTo = process.env.ECHOES_PUBLISH_TO === 'pulse' ? 'pulse' : 'kafka'

3. Remove Kafka consumption

After every publisher uses Pulse and Kafka has no pending events for the service, stop consuming from Kafka:

events: {
  pulse: {
    connectionString: process.env.MONGO_URL,
    databaseName: 'events',
    consumerGroup: 'billing',
  },
  consumeFrom: ['pulse'],
  publishTo: 'pulse',
}

The legacy client, consumer, and producer options can then be removed from that application. If only Pulse is configured, Echoes defaults both consumption and publishing to Pulse.

Echoes never publishes the same call to both transports. During a rolling publisher deployment, Kafka and Pulse form two independent delivery streams, so they do not provide a shared global sequence. Use a coordinated publisher cutover after draining Kafka when cross-transport ordering matters.

Pulse is an optional peer dependency and is loaded only when configured. Applications that stay on Kafka do not need to install or initialize it.

Error Handling

Echoes automatically handles errors in request and event handlers:

@EchoRequest()
processPayment = createEchoRequest({
  params: { type: PaymentParamsSchema },
  returns: PaymentResultSchema,
  resolve: async (params) => {
    try {
      const result = await this.paymentService.processPayment(params)
      return result
    } catch (error) {
      // Errors are automatically propagated back to the requester
      // with proper error classification (UserError, ValidationError, etc.)
      throw new Error(`Payment processing failed: ${error.message}`)
    }
  }
})

Custom Error Types

Orionjs handles special error types appropriately:

UserError: For expected application errors
ValidationError: For data validation errors

import { UserError } from '@orion-js/helpers'
import { ValidationError } from '@orion-js/schema'

@EchoRequest()
validateUser = createEchoRequest({
  params: {
    userId: { type: String }
  },
  resolve: async (params) => {
    const user = await this.userRepository.findById(params.userId)
    
    if (!user) {
      throw new UserError('USER_NOT_FOUND', 'User was not found')
    }
    
    if (!user.isActive) {
      throw new ValidationError({
        status: 'User account is inactive'
      })
    }
  }
})

Those errors will be automatically propagated back to the requester.

Type Safety

Using TypeScript with schema inference, you can ensure type safety for your echo handlers:

// Define schemas for strong typing
const CreateOrderParamsSchema = schemaWithName('CreateOrderParams', {
  customerId: { type: String },
  items: {
    type: [{ 
      productId: { type: String },
      quantity: { type: Number }
    }]
  }
})

const OrderResultSchema = schemaWithName('OrderResult', {
  orderId: { type: String },
  total: { type: Number },
  status: { type: String, allowedValues: ['pending', 'completed'] }
})

// Infer types from schemas
type CreateOrderParamsType = InferSchemaType<typeof CreateOrderParamsSchema>
type OrderResultType = InferSchemaType<typeof OrderResultSchema>

@EchoRequest()
createOrder = createEchoRequest({
  params: CreateOrderParamsSchema,
  returns: OrderResultSchema,
  resolve: async (params: CreateOrderParamsType): Promise<OrderResultType> => {
    // Implementation with full type safety
  }
})

Best Practices

Organize by Domain: Group related echo handlers in the same controller class.
Leverage Dependency Injection: Use @Inject(() => Service) to access repositories and services.
Keep Methods Focused: Each echo handler should have a clear, single responsibility.
Use Strong Typing: Define parameter and return types with schemas and infer TypeScript types.
Handle Errors Gracefully: Catch and properly categorize errors.
Idempotent Handlers: Design event handlers to be idempotent (safe to process the same event multiple times).
Timeout Configuration: Set appropriate timeouts for requests based on expected execution time.
Security: Use the shared key to secure inter-service communication.
Service Discovery: Keep the service registry updated when adding new services.
Monitoring: Implement proper logging and monitoring for echo handlers.

Get Started

Essentials

Controllers

Other Modules

Standalone core and Orion integration

Creating Echo Controllers

Echo Request Handlers

Echo Event Handlers

Event Decorator Options

Batch event handlers with Pulse

Making Requests

Publishing Events

Starting the Echoes Service

Using Echoes without Orion

Incremental migration from Kafka to Pulse

1. Prepare every listener

2. Switch publishers

3. Remove Kafka consumption

Error Handling

Custom Error Types

Type Safety

Best Practices

​Standalone core and Orion integration

​Creating Echo Controllers

​Echo Request Handlers

​Echo Event Handlers

​Event Decorator Options

​Batch event handlers with Pulse

​Making Requests

​Publishing Events

​Starting the Echoes Service

​Using Echoes without Orion

​Incremental migration from Kafka to Pulse

​1. Prepare every listener

​2. Switch publishers

​3. Remove Kafka consumption

​Error Handling

​Custom Error Types

​Type Safety

​Best Practices

Standalone core and Orion integration

Creating Echo Controllers

Echo Request Handlers

Echo Event Handlers

Event Decorator Options

Batch event handlers with Pulse

Making Requests

Publishing Events

Starting the Echoes Service

Using Echoes without Orion

Incremental migration from Kafka to Pulse

1. Prepare every listener

2. Switch publishers

3. Remove Kafka consumption

Error Handling

Custom Error Types

Type Safety

Best Practices