> ## Documentation Index
> Fetch the complete documentation index at: https://www.orionjs.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Echoes

> Event and request handlers for microservice communication in Orionjs

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.

## Creating Echo Controllers

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

```typescript theme={null}
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:

```typescript theme={null}
@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:

```typescript theme={null}
@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()`:

```typescript theme={null}
@EchoEvent()
processSomething = createEchoEvent({
  attemptsBeforeDeadLetter: 5,
  params: {
    // schema definition
  },
  resolve: async (params) => {
    // implementation
  }
})
```

## Making Requests

To make a request to another service:

```typescript theme={null}
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:

```typescript theme={null}
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:

```typescript theme={null}
import { startService } from '@orion-js/echoes'
import { app } from '@orion-js/http'

// Start the echoes service
await startService({
  // 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
})
```

## 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:

```bash theme={null}
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:

```typescript theme={null}
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 five connections per service
instance by default. Set `events.pulse.maxPoolSize` when a specific service needs a different
limit.

### 2. Switch publishers

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

```typescript theme={null}
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:

```typescript theme={null}
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:

```typescript theme={null}
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.

<Warning>
  Echoes never publishes the same call to both transports. During a rolling publisher deployment,
  Kafka and Pulse form two independent ordered streams, so they do not provide a shared total
  order. Topics that require strict ordering should use a coordinated publisher cutover after
  draining Kafka.
</Warning>

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:

```typescript theme={null}
@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

```typescript theme={null}
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:

```typescript theme={null}
// 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

1. **Organize by Domain**: Group related echo handlers in the same controller class.

2. **Leverage Dependency Injection**: Use `@Inject(() => Service)` to access repositories and services.

3. **Keep Methods Focused**: Each echo handler should have a clear, single responsibility.

4. **Use Strong Typing**: Define parameter and return types with schemas and infer TypeScript types.

5. **Handle Errors Gracefully**: Catch and properly categorize errors.

6. **Idempotent Handlers**: Design event handlers to be idempotent (safe to process the same event multiple times).

7. **Timeout Configuration**: Set appropriate timeouts for requests based on expected execution time.

8. **Security**: Use the shared key to secure inter-service communication.

9. **Service Discovery**: Keep the service registry updated when adding new services.

10. **Monitoring**: Implement proper logging and monitoring for echo handlers.