NatsPubsub

Declarative Pub/Sub messaging for NATS JetStream

A production-ready pub/sub library with a familiar, declarative API. Features declarative subscribers, middleware support, and battle-tested reliability patterns including Inbox/Outbox, DLQ, and automatic retries with backoff.

Implementations for and with full interoperability

• Documentation
• Quick Start
• Packages
• Features
• Real-World Examples
• Performance
• Development
• Contributing
• License

📚 View Full Documentation Site →

Comprehensive documentation is available in the docs/ directory:

• Introduction - What is NatsPubsub and why use it
• Installation - Setup for JavaScript and Ruby
• JavaScript Quick Start - Get running in 5 minutes
• Ruby Quick Start - Get running in 5 minutes
• Core Concepts - Topics, subscribers, reliability patterns

• Publishing Messages - Complete publishing guide
• Creating Subscribers - Subscriber patterns and best practices
• Middleware System - Add cross-cutting concerns
• Testing Strategies - Unit, integration, and E2E testing
• Deployment Guide - Docker, Kubernetes, production setup
• Performance Tuning - Optimization strategies

• Inbox/Outbox Pattern - Guaranteed delivery and exactly-once processing
• Dead Letter Queue - Handle failed messages
• Schema Validation - Validate messages with Zod
• Event Sourcing - Build event-sourced systems

• Ruby on Rails - Rails integration guide
• Express.js - Express integration
• NestJS - NestJS integration
• Databases - PostgreSQL, MySQL, SQLite

• JavaScript API - Complete TypeScript/JavaScript API
• Ruby API - Complete Ruby API
• Configuration - All configuration options
• CLI Reference - Command-line tools

• Architecture - System design and components
• Internals - How NatsPubsub works
• Custom Repositories - Implement custom storage
• Security - Security best practices

• Common Issues - Solutions to frequent problems
• Debugging Guide - Debug message flow
• FAQ - Frequently asked questions

• Example Projects - Complete working examples
• Microservices Example - Multi-service architecture
• JavaScript Examples - TypeScript/JavaScript examples
• Ruby Examples - Ruby examples

📚 Start with the Documentation Index →

Start the full development environment with Docker Compose:

git clone https://github.com/attaradev/nats_pubsub.git
cd nats-pubsub
docker compose up -d

This starts NATS, PostgreSQL, Prometheus, and Grafana with preconfigured monitoring.

For package-specific setup:

• Ruby Setup →
• JavaScript Setup →

Rails-integrated Pub/Sub library with Web UI, Inbox/Outbox, and ActiveRecord support.

gem "nats_pubsub", "~> 0.1"

📖 Full Ruby Docs →

Node.js Pub/Sub library with full TypeScript support and enterprise monitoring.

pnpm add nats-pubsub

📖 Full JavaScript Docs →

• 🎯 Topic-Based Messaging - Flexible pub/sub with hierarchical topics (e.g., order.created, notification.email)
• 🔌 Declarative Subscriber API - Clean, decorator-based subscription patterns
• 🌲 Hierarchical Topics - Organize messages with dot notation and wildcard subscriptions (* for single level, > for multiple levels)
• 🧨 Dead Letter Queue (DLQ) - Automatic handling of failed messages
• ⚙️ Durable Pull Consumers - Reliable message delivery with exponential backoff
• 🎭 Middleware System - Extensible processing pipeline for cross-cutting concerns
• 🔄 Auto-Topology Management - Automatic JetStream stream and consumer creation

• 🛡️ Inbox/Outbox reliability patterns
• 📊 Web UI for monitoring
• 🔗 ActiveRecord integration
• 🚂 Rails generators

• 📊 Prometheus metrics
• ❤️ Health check endpoints
• 📦 Batch publishing API
• 🚀 Full TypeScript support

Both implementations use identical event formats, enabling seamless interoperability between Ruby and JavaScript services.

📖 Learn More:

• Ruby Package Docs - Full Ruby documentation with Rails integration
• JavaScript Package Docs - Full TypeScript/JavaScript documentation
• Performance Benchmarks - Throughput and latency metrics
• Contributing Guide - How to contribute to the project

Scenario: Process orders across multiple services (inventory, shipping, notifications)

Ruby Publisher (Order Service):

# app/services/order_service.rb
class OrderService
  def create_order(user_id, items)
    Order.transaction do
      order = Order.create!(user_id: user_id, status: 'pending')

      # Publish order created event
      NatsPubsub.publish(
        topic: 'order.created',
        message: {
          order_id: order.id,
          user_id: user_id,
          item: items,
          total: calculate_total(items)
        }
      )
    end
  end
end

JavaScript Subscriber (Inventory Service):

// subscribers/inventory-subscriber.ts
import { Subscriber, TopicMetadata } from "nats-pubsub";

class InventorySubscriber extends Subscriber {
  constructor() {
    super("production.order-service.order.created");
  }

  async handle(message: Record<string, unknown>, metadata: TopicMetadata) {
    const { order_id, item } = message;

    // Reserve inventory
    await this.reserveInventory(item as Array<any>);

    // Publish inventory reserved event
    await NatsPubsub.publish("inventory.reserved", {
      order_id,
      reserved_at: new Date(),
    });
  }

  private async reserveInventory(item: any[]) {
    // Inventory reservation logic
  }
}

Scenario: Send welcome emails, SMS, and push notifications when users sign up

JavaScript Publisher (Auth Service):

// services/auth.service.ts
async function createUser(userData: UserData) {
  const user = await User.create(userData);

  // Publish to multiple notification topics
  await NatsPubsub.publish({
    topics: ["notification.email", "notification.sms", "audit.user-created"],
    message: {
      user_id: user.id,
      email: user.email,
      phone: user.phone,
      name: user.name,
    },
  });

  return user;
}

Ruby Subscribers (Notification Service):

# app/subscribers/email_notification_subscriber.rb
class EmailNotificationSubscriber < NatsPubsub::Subscriber
  subscribe_to "notification.email"

  def handle(message, context)
    WelcomeMailer.welcome_email(
      email: message['email'],
      name: message['name']
    ).deliver_later
  end
end

# app/subscribers/sms_notification_subscriber.rb
class SmsNotificationSubscriber < NatsPubsub::Subscriber
  subscribe_to "notification.sms"

  def handle(message, context)
    TwilioService.send_welcome_sms(
      phone: message['phone'],
      name: message['name']
    )
  end
end

Scenario: Payment service notifies order and analytics services

// Payment Service (Publisher)
await NatsPubsub.publish("payment.completed", {
  payment_id: payment.id,
  order_id: payment.order_id,
  amount: payment.amount,
  currency: "USD",
});

// Order Service (Subscriber)
class PaymentCompletedSubscriber extends Subscriber {
  constructor() {
    super("production.payment-service.payment.completed");
  }

  async handle(message: Record<string, unknown>) {
    await Order.update(message.order_id, { status: "paid" });
  }
}

// Analytics Service (Subscriber with wildcard)
class PaymentAnalyticsSubscriber extends Subscriber {
  constructor() {
    super("production.payment-service.payment.*"); // Wildcard subscription
  }

  async handle(message: Record<string, unknown>, metadata: TopicMetadata) {
    await Analytics.track({
      event: metadata.topic,
      properties: message,
    });
  }
}

Scenario: Centralized audit log for all system events

# Any service publishing events
NatsPubsub.publish(
  topic: 'user.login',
  message: { user_id: user.id, ip: request.ip },
  trace_id: request_id
)

# Audit Service (Subscriber) - Wildcard to capture all events
class AuditLogSubscriber < NatsPubsub::Subscriber
  subscribe_to "production.*.>" # Subscribe to ALL topics from all services

  def handle(message, context)
    AuditLog.create!(
      event_type: context.topic,
      user_id: message['user_id'],
      data: message,
      trace_id: context.trace_id,
      occurred_at: context.occurred_at
    )
  end
end

NatsPubsub leverages NATS JetStream for high-performance messaging:

Typical Microservices Setup:

• Latency: 2-10ms (including network + processing)
• Throughput: 50K-200K msg/sec per service
• Concurrency: 10-50 concurrent message processors

Factors Affecting Performance:

• 🔸 Network latency between services
• 🔸 Message size and serialization
• 🔸 Subscriber processing time
• 🔸 Database operations in subscribers
• 🔸 JetStream storage type (memory vs file)

// JavaScript
NatsPubsub.configure({
  concurrency: 20, // Adjust based on workload
});

# Ruby
NatsPubsub.configure do |config|
  config.concurrency = 20
end

Guidelines:

• CPU-bound tasks: concurrency = CPU cores
• I/O-bound tasks: concurrency = 2-4x CPU cores
• Database-heavy: concurrency ≤ DB pool size

class BatchProcessor extends BaseSubscriber {
  private batch: any[] = [];
  private batchSize = 100;
  private flushInterval = 1000; // 1 second

  async call(message: Record<string, unknown>) {
    this.batch.push(message);

    if (this.batch.length >= this.batchSize) {
      await this.flush();
    }
  }

  private async flush() {
    if (this.batch.length === 0) return;

    // Process batch
    await Database.bulkInsert(this.batch);
    this.batch = [];
  }
}

// Reuse database connections
const pool = new Pool({
  max: 20, // Match or exceed concurrency
  min: 5,
});

// Keep messages small
await NatsPubsub.publish("user.created", {
  id: user.id, // ✅ Reference
  // user: fullUserObject  // ❌ Avoid embedding large objects
});

// Add timing middleware
class PerformanceMiddleware {
  async call(event: any, metadata: any, next: () => Promise<void>) {
    const start = Date.now();

    await next();

    const duration = Date.now() - start;
    if (duration > 1000) {
      console.warn(`Slow processing: ${metadata.subject} took ${duration}ms`);
    }
  }
}

NatsPubsub.use(new PerformanceMiddleware());

Note: Based on AWS/GCP pricing for comparable instances.

# Install dependencies (monorepo root)
pnpm install

# Or install per package
cd packages/ruby && bundle install
cd packages/javascript && pnpm install

# Run tests
cd packages/ruby && bundle exec rspec
cd packages/javascript && pnpm test

# Run with coverage
cd packages/javascript && pnpm test -- --coverage
cd packages/ruby && bundle exec rspec --format documentation

# Build all packages
pnpm build

# Lint all packages
pnpm lint

# Documentation website
pnpm start:website    # Start dev server
pnpm build:website    # Build for production

This repository uses Husky for Git hooks:

• pre-commit → runs lint-staged
• commit-msg → validates Conventional Commits
• pre-push → runs tests before pushing

Set up hooks after cloning:

pnpm install  # Automatically configures hooks

More guides:

• Ruby Development →
• JavaScript Development →
• CI/CD Setup →

We welcome contributions!

See CONTRIBUTING.md for details.

Quick Steps:

1. Fork the repo
2. Create a branch (git checkout -b feat/awesome-feature)
3. Add tests and implement changes
4. Commit using Conventional Commits
5. Open a Pull Request

MIT License - Copyright (c) 2025 Mike Attara