@nl-framework/microservices

Microservices module for nl-framework providing NestJS-style message patterns, event-driven architecture, and Dapr integration.

Features

• Message Patterns: Decorate controller methods with @MessagePattern and @EventPattern to handle incoming messages
• Client API: Publish events with emit() (fire-and-forget) or send requests with send() (request/response)
• Dapr Integration: Built-in transport for Dapr sidecar with pub/sub support
• Pluggable Transports: Abstract transport interface allows custom implementations
• Logger Integration: Uses @nl-framework/logger for structured logging
• Shared Decorators: Message handlers can reuse @UseGuards(), @UseInterceptors(), @UsePipes(), and @UseFilters() from @nl-framework/core, just like HTTP and GraphQL controllers

Installation

bun add @nl-framework/microservices

Quick Start

1. Configure Dapr Settings

Add to your config/default.yaml:

microservices:
  dapr:
    httpPort: 3500
    grpcPort: 50001
    componentsPath: ./dapr/components
  pubsub:
    name: redis-pubsub

2. Create a Message Handler Controller

import { Controller, Injectable } from '@nl-framework/core';
import { MessagePattern, EventPattern } from '@nl-framework/microservices';

@Controller()
@Injectable()
export class OrdersController {
  @MessagePattern('order.created')
  handleOrderCreated(context: MessageContext) {
    console.log('Order created:', context.data);
    return { status: 'processed' };
  }

  @EventPattern({ cmd: 'notify' })
  handleNotification(context: MessageContext) {
    console.log('Notification:', context.data);
  }
}

Decorator Tip: Import @UseGuards(), @UseInterceptors(), @UsePipes(), and @UseFilters() from @nl-framework/core to apply the same guard/interceptor/pipe/filter semantics to message handlers that you already use in HTTP or GraphQL controllers.

3. Register the Module

import { Module } from '@nl-framework/core';
import { createMicroservicesModule } from '@nl-framework/microservices';
import { OrdersController } from './orders.controller';

@Module({
  imports: [
    createMicroservicesModule({
      controllers: [OrdersController],
    }),
  ],
  controllers: [OrdersController],
})
export class AppModule {}

4. Publish Events

import { MicroserviceClient } from '@nl-framework/microservices';

@Injectable()
export class OrdersService {
  constructor(private readonly client: MicroserviceClient) {}

  async createOrder(data: unknown) {
    // Fire-and-forget event
    await this.client.emit('order.created', { orderId: 123, ...data });

    // Request/response over Dapr service invocation
    const result = await this.client.send('process.order', data, { appId: 'orders' });
  }
}

API Reference

Decorators

@MessagePatternDecorator(pattern)

Marks a method as a message handler for request/response patterns.

@MessagePatternDecorator('user.get')
async getUser(context: MessageContext) {
  return { id: 1, name: 'John' };
}

@EventPattern(pattern)

Marks a method as an event handler (fire-and-forget).

@EventPattern('user.created')
handleUserCreated(context: MessageContext) {
  console.log('New user:', context.data);
}

Client Methods

emit(pattern, data)

Publishes a fire-and-forget event.

await client.emit('order.created', { orderId: 123 });

send(pattern, data, options?)

Invokes a service over Dapr service invocation and awaits the typed response. Options: timeout (default 30s), appId (target app), signal. A non-2xx response throws MicroserviceInvocationException (carrying status + body).

const result = await client.send('process.payment', { amount: 100 }, {
  appId: 'payments',
  timeout: 5000,
});

Exception Filters

Microservice handlers use the shared @UseFilters() decorator and the MicroserviceExceptionFilter interface to handle errors consistently with HTTP and GraphQL.

Scoped filters

import { Controller, Injectable, UseFilters } from '@nl-framework/core';
import {
  MessagePattern,
  MicroserviceExceptionFilter,
  MicroserviceExceptionContext,
} from '@nl-framework/microservices';

@Injectable()
class LoggingFilter implements MicroserviceExceptionFilter {
  catch(exception: Error, context: MicroserviceExceptionContext) {
    context.logger?.error({ pattern: context.pattern, exception });
    return { ok: false };
  }
}

@Controller()
@UseFilters(LoggingFilter)
export class OrdersConsumer {
  @MessagePattern('orders.create')
  async create(order: OrderDto) {
    throw new Error('Not implemented');
  }
}

Method-level filters run before class-level filters. Return a value from catch() to override the handler response; return undefined to continue to the next filter.

Global filters

import { registerMicroserviceExceptionFilter } from '@nl-framework/microservices';

registerMicroserviceExceptionFilter(new LoggingFilter());

Use registerMicroserviceExceptionFilter() or registerMicroserviceExceptionFilters() during bootstrap to apply filters to every handler. The dispatcher resolves tokens through dependency injection if available, then instantiates classes or uses provided instances.

Dapr Setup

1. Create Component Definition

Create dapr/components/redis-pubsub.yaml:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: redis-pubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
    - name: redisHost
      value: localhost:6379

2. Run with Dapr Sidecar

dapr run --app-id my-service --app-port 3000 --dapr-http-port 3500 -- bun run start

Transport Options

DaprTransport

import { DaprTransport } from '@nl-framework/microservices';

const transport = new DaprTransport({
  daprHost: 'localhost',
  daprHttpPort: 3500,
  pubsubName: 'redis-pubsub',
  logger: myLogger,
});

Custom Transport

Implement the Transport interface:

import type { Transport, MessagePattern } from '@nl-framework/microservices';

export class CustomTransport implements Transport {
  async connect(): Promise<void> {
    // Initialize connection
  }

  async close(): Promise<void> {
    // Cleanup
  }

  async emit(pattern: MessagePattern, data: unknown): Promise<void> {
    // Publish event
  }

  async send<TResult>(pattern: MessagePattern, data: unknown): Promise<TResult> {
    // Send request, await response
  }
}

Roadmap

• [x] Guards, interceptors, and pipes execute in the dispatcher pipeline
• [x] Request/response pattern via Dapr service invocation (send())
• [x] Automatic subscription endpoint (GET /dapr/subscribe) for Dapr pub/sub
• [x] Sidecar health check on connect()
• [ ] Message serialization/deserialization strategies
• [ ] Retry policies and dead-letter queues
• [ ] Additional transport implementations (NATS, RabbitMQ, etc.)

License

MIT