@xemahq/event-hub-client

NestJS client library for integrating microservices with Event Hub API.

Features

• Automatic Registration — Background registration with exponential backoff
• Event Publishing — Fire-and-forget and confirmed publishing
• Webhook Guard — JWT validation for incoming events
• Token Provider — Pluggable token provider for service authentication
• Event Queue — In-memory queue when Event Hub is unavailable
• Non-Blocking Startup — Services start without waiting for Event Hub
• Schema Versioning — Track event schema versions per producer
• Event Migration — Optional migration functions for handling breaking changes
• Field-Level Encryption — Per-service AES-256 encryption for sensitive event payloads

Installation

This package is hosted on GitHub Packages. You need to configure npm to use GitHub Packages registry for the @neuralchowder scope.

1. Configure npm Registry

Create an .npmrc file in your project root:

@xemahq:registry=https://npm.pkg.github.com
//npm.pkg.github.com/:_authToken=${NPM_TOKEN}

2. Authentication

For local development, authenticate interactively:

npm login --scope=@xemahq --registry=https://npm.pkg.github.com

Use your GitHub username and a Personal Access Token (PAT) with read:packages permission as the password.

For CI/CD (GitHub Actions), add this step before npm install:

⚠️ Important: The default GITHUB_TOKEN does not have permission to read packages from other repositories in the organization. You must create a Personal Access Token (PAT) with read:packages scope and store it as a repository secret named GH_PACKAGES_TOKEN.

Creating the secret:

1. Go to GitHub → Settings → Developer settings → Personal access tokens
2. Generate a new token with read:packages scope
3. In your repository, go to Settings → Secrets and variables → Actions
4. Create a new repository secret named GH_PACKAGES_TOKEN with the PAT value

Workflow configuration:

- name: Setup npm for GitHub Packages
  run: |
    echo "@xemahq:registry=https://npm.pkg.github.com" >> .npmrc
    echo "//npm.pkg.github.com/:_authToken=${{ secrets.GH_PACKAGES_TOKEN }}" >> .npmrc

3. Install the Package

npm install @xemahq/event-hub-client

Quick Start

1. Import the Module

Services authenticate with Event Hub using a JWT access token provided by a tokenProvider function. This can be any OIDC client_credentials flow or a managed identity service.

// app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { IdentityBootstrapModule, IdentityBootstrapService } from '@xemahq/identity-api-client';
import { EventHubModule } from '@xemahq/event-hub-client';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),

    // 1. Register with identity-api to get a Keycloak service account
    IdentityBootstrapModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService) => ({
        identityApiUrl: config.getOrThrow('IDENTITY_API_URL'),
        identityApiInternalToken: config.getOrThrow('IDENTITY_API_INTERNAL_TOKEN'),
        serviceName: 'my-service',
        serviceDisplayName: 'My Service',
      }),
      inject: [ConfigService],
    }),

    // 2. Register with event-hub for publishing/subscribing
    EventHubModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService, identityBootstrap: IdentityBootstrapService) => ({
        eventHubUrl: config.getOrThrow('EVENT_HUB_URL'),
        serviceId: 'my-service',
        serviceName: 'My Service',
        schemaVersion: '1.0.0',
        dataTypes: ['order', 'payment'],
        eventTypes: ['user.*', 'organization.*'],
        webhookUrl: 'https://my-service.example.com/webhooks/events',
        tokenProvider: () => identityBootstrap.getAccessToken(),
      }),
      inject: [ConfigService, IdentityBootstrapService],
    }),
  ],
})
export class AppModule {}

2. Environment Variables

# Event Hub
EVENT_HUB_URL=http://event-hub.svc.cluster.local:3128

# Identity API (for service account registration)
IDENTITY_API_URL=http://identity-api.svc.cluster.local:3053
IDENTITY_API_INTERNAL_TOKEN=your-identity-api-token

3. Publish Events

import { Injectable } from '@nestjs/common';
import { EventPublisherService } from '@xemahq/event-hub-client';

@Injectable()
export class OrdersService {
  constructor(private readonly eventPublisher: EventPublisherService) {}

  async createOrder(dto: CreateOrderDto): Promise<Order> {
    const order = await this.ordersRepository.create(dto);

    // Fire-and-forget (doesn't throw on failure)
    this.eventPublisher.emit('order.created', 'order', {
      id: order.id,
      customerId: order.customerId,
      total: order.total,
    });

    // Or wait for confirmation (throws on failure)
    await this.eventPublisher.publish('order.created', 'order', {
      id: order.id,
      customerId: order.customerId,
      total: order.total,
    });

    return order;
  }
}

4. Receive Events

import { Controller, Post, Body, UseGuards } from '@nestjs/common';
import {
  EventHubWebhookGuard,
  EventPayload,
} from '@xemahq/event-hub-client';

@Controller('webhooks')
export class WebhooksController {
  @Post('events')
  @UseGuards(EventHubWebhookGuard)
  async handleEvent(@Body() event: EventPayload) {
    switch (event.eventType) {
      case 'user.created':
        await this.handleUserCreated(event.payload);
        break;
      case 'user.updated':
        await this.handleUserUpdated(event.payload);
        break;
    }
    return { status: 'processed' };
  }
}

API Reference

EventHubModule

Dynamic NestJS module for Event Hub integration.

EventHubModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: (config: ConfigService, identityBootstrap: IdentityBootstrapService) => ({
    eventHubUrl: string;              // Event Hub base URL
    serviceId: string;                // Your service ID (becomes OAuth2 client_id)
    serviceName: string;              // Your service display name
    schemaVersion: string;            // Schema version of your events (semver, e.g. '1.0.0')
    tokenProvider: () => Promise<string>; // Token provider (typically from IdentityBootstrapService)
    dataTypes?: string[];             // Data types you own (producer)
    eventTypes?: string[];            // Event patterns to subscribe to (consumer)
    webhookUrl?: string;              // Your webhook endpoint for receiving events
    eventDomain?: string;             // Event domain prefix (e.g. 'backlog' for 'backlog.item.created')
    timeout?: number;                 // Request timeout (default: 10000ms)
    maxQueueSize?: number;            // Event queue size (default: 1000)
    eventMigration?: (event, fromVersion, toVersion) => Record<string, unknown>; // Optional migration function
  }),
  inject: [ConfigService, IdentityBootstrapService],
});

Note: You don't need to configure Keycloak URL/realm. Event Hub returns all necessary OAuth2 configuration (tokenEndpoint, jwksUrl, issuer) in the registration response.

Token Provider

The tokenProvider is a required function () => Promise<string> that returns a valid JWT access token. It is called on every request that needs authentication.

Option A — OIDC client_credentials (standard):

// Minimal OIDC client_credentials token provider
async function getToken(): Promise<string> {
  const res = await fetch(process.env.OIDC_TOKEN_URL, {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
    body: new URLSearchParams({
      grant_type: 'client_credentials',
      client_id: process.env.OIDC_CLIENT_ID,
      client_secret: process.env.OIDC_CLIENT_SECRET,
    }),
  });
  const json = await res.json();
  return json.access_token;
}

// In the module config:
tokenProvider: getToken,

Option B — identity-api (@xemahq/identity-api-client):

tokenProvider: () => identityBootstrap.getAccessToken()

IdentityBootstrapService handles token caching, refresh at 90% of lifetime, and exponential backoff.

Required Roles

Event Hub validates that services have the micro-service role in their Keycloak token. This role is automatically assigned by identity-api when creating the service account.

interface RegistrationResponse {
  data: {
    id: string;
    serviceId: string;
    serviceName: string;
    dataTypes: string[];
    isActive: boolean;
    authInfo?: { tokenEndpoint: string; jwksUrl: string; issuer: string };
    encryption?: { serviceKey: string };
    roles: string[];  // ['micro-service']
  };
}

System Events (Transparent Handling)

The client library transparently handles system events from Event Hub:

| Event Type | Handled By | Description | |------------|------------|-------------| | system.refresh_registration | EventHubBootstrapService | Triggers automatic re-registration |

When Event Hub broadcasts a system.refresh_registration event:

1. The EventHubWebhookGuard detects it as a system event
2. EventHubBootstrapService.handleSystemEvent() is called automatically
3. The service re-registers, refreshing credentials and encryption keys
4. Your webhook handler still receives the event (for custom handling if needed)

Use cases:

• Credential refresh across all instances
• Secret rotation
• Configuration updates

// You can also manually trigger re-registration
@Injectable()
class MyService {
  constructor(private readonly bootstrapService: EventHubBootstrapService) {}

  async rotateCredentials() {
    await this.bootstrapService.forceReregistration();
  }
}

EventHubClient

Main API client for direct Event Hub operations.

@Injectable()
class MyService {
  constructor(private readonly eventHub: EventHubClient) {}

  async example() {
    // Check if ready to publish
    if (this.eventHub.isReady()) {
      await this.eventHub.publishEvent({
        eventType: 'order.shipped',
        dataType: 'order',
        payload: { orderId: '123' },
      });
    }

    // Check if event was already processed
    const processed = await this.eventHub.isEventProcessed('event-id');

    // Acknowledge event as processed
    await this.eventHub.acknowledgeEvent('event-id');

    // Trigger replay after downtime
    await this.eventHub.triggerReplay();
  }
}

Service-to-Service Authentication

For calling other services, use IdentityBootstrapService directly from @xemahq/identity-api-client:

import { IdentityBootstrapService } from '@xemahq/identity-api-client';

@Injectable()
class MyService {
  constructor(
    private readonly identityBootstrap: IdentityBootstrapService,
    private readonly httpService: HttpService,
  ) {}

  async callOtherService() {
    // Get a valid access token (automatically cached and refreshed)
    const token = await this.identityBootstrap.getAccessToken();

    // Use it for inter-service API calls
    const response = await firstValueFrom(
      this.httpService.get('http://other-service/resource', {
        headers: { Authorization: `Bearer ${token}` },
      }),
    );

    return response.data;
  }
}

The token is:

• Obtained using OAuth2 client_credentials grant
• Automatically cached until near expiry (90% of lifetime)
• Valid for calling any service in the same Keycloak realm
• Concurrent refresh requests are deduplicated

EventPublisherService

Simplified event publishing service.

// Fire-and-forget (logs errors, doesn't throw)
eventPublisher.emit('order.created', 'order', payload);

// Wait for confirmation (throws on failure)
await eventPublisher.publish('order.created', 'order', payload);

// Publish with sensitive fields (auto-encrypted)
await eventPublisher.publish('user.created', 'user', {
  id: 'user-123',
  email: '[email protected]',
  profile: {
    ssn: '123-45-6789',
    address: { street: '123 Main St', city: 'Anytown' },
  },
}, ['email', 'profile.ssn', 'profile.address.street']);

// Check status
eventPublisher.isReady();           // Has credentials?
eventPublisher.isEncryptionReady(); // Has encryption key?
eventPublisher.getQueueSize();      // Events waiting to be sent

EventHubBootstrapService

Handles automatic registration. Usually you don't interact with this directly.

// Check registration status
bootstrapService.isEventHubRegistered();

// Force re-registration
await bootstrapService.forceReregistration();

EventHubWebhookGuard

JWT validation guard for incoming webhooks from Event Hub.

@Post('events')
@UseGuards(EventHubWebhookGuard)
async handleEvent(@Body() event: EventPayload) {
  // JWT has been validated, `azp` claim is 'event-hub-api'
  // Sensitive fields are automatically decrypted
  console.log(event.payload.email); // Already decrypted
}

Field-Level Encryption

Event Hub provides transparent field-level encryption for sensitive data. Each service receives a unique AES-256 encryption key during registration, and the client library handles encryption/decryption automatically.

How It Works

1. At Registration: Event Hub generates a unique 256-bit key for your service
2. When Publishing: Client encrypts specified fields before sending
3. Event Hub Storage: Data is re-encrypted with master key at rest
4. When Delivering: Event Hub re-encrypts with subscriber's key
5. When Receiving: Guard automatically decrypts before your handler runs

Publishing Sensitive Data

Specify sensitive fields using dot-notation paths:

// Basic usage
await eventPublisher.publish('user.created', 'user', {
  id: 'user-123',
  email: '[email protected]',
  name: 'John Doe',
}, ['email']);

// Nested fields
await eventPublisher.publish('payment.processed', 'payment', {
  id: 'pay-456',
  amount: 99.99,
  card: {
    last4: '4242',
    number: '4242424242424242',
    cvv: '123',
  },
  customer: {
    name: 'John',
    ssn: '123-45-6789',
  },
}, ['card.number', 'card.cvv', 'customer.ssn']);

// Fire-and-forget with encryption
eventPublisher.emit('user.updated', 'user', payload, ['email', 'phone']);

Receiving Encrypted Data

The EventHubWebhookGuard automatically decrypts fields:

@Post('events')
@UseGuards(EventHubWebhookGuard)
async handleEvent(@Body() event: EventPayload) {
  // Sensitive fields are already decrypted
  const email = event.payload.email; // Plaintext, not encrypted
  const ssn = event.payload.profile.ssn; // Also plaintext
  
  // You can still see which fields were encrypted
  console.log(event.sensitiveAttributes); // ['email', 'profile.ssn']
}

Manual Encryption (Advanced)

Access the encryption service directly for custom scenarios:

import { EventHubEncryptionService } from '@xemahq/event-hub-client';

@Injectable()
class MyService {
  constructor(private readonly encryption: EventHubEncryptionService) {}

  encryptManually(value: string): string {
    return this.encryption.encrypt(value);
  }

  decryptManually(encrypted: string): string {
    return this.encryption.decrypt(encrypted);
  }
}

Security Notes

• Key Isolation: Each service has a unique key - your key cannot decrypt other services' data
• Automatic Key Setup: Key is configured automatically during registration
• No Key Sharing: Services never see each other's encryption keys
• At-Rest Protection: Events stored in Event Hub are encrypted with a master key

Idempotency

Events may be delivered more than once. Handle this with:

1. Event Hub Acknowledgment (recommended for expensive operations):

   const processed = await eventHub.isEventProcessed(event.id);
   if (!processed) {
     await handleEvent(event);
     await eventHub.acknowledgeEvent(event.id);
   }

2. Natural Idempotency (recommended for CRUD):

   await prisma.user.upsert({
     where: { externalId: payload.id },
     create: { ... },
     update: { ... },
   });

Schema Versioning & Event Migration

Every producer declares its schemaVersion (semver) when registering. Event Hub attaches this version to every webhook delivery, enabling consumers to detect and handle breaking changes.

How It Works

1. Producer registers with schemaVersion: '1.0.0' in the EventHubModule config
2. Event Hub stores the version alongside the producer record
3. On delivery, Event Hub includes schemaVersion in every webhook payload
4. Consumer receives the version and can compare it against its own expected version
5. If versions differ, the optional eventMigration function transforms the payload

Webhook Payload

interface EventPayload {
  id: string;
  sequenceId: string;
  eventType: string;
  dataType: string;
  schemaVersion: string;  // Version from the producer that published this event
  payload: Record<string, unknown>;
  timestamp: string;
  sensitiveAttributes?: string[];
}

Event Migration Function

Configure an optional eventMigration function in your module config to automatically transform incoming events when schema versions differ:

EventHubModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: (config: ConfigService) => ({
    eventHubUrl: config.getOrThrow('EVENT_HUB_URL'),
    serviceId: 'my-service',
    serviceName: 'My Service',
    schemaVersion: '2.0.0',
    dataTypes: ['order'],
    eventTypes: ['user.*'],
    webhookUrl: config.getOrThrow('WEBHOOK_URL'),

    // Handle events from producers running older schema versions
    eventMigration: (payload, fromVersion, toVersion) => {
      if (fromVersion === '1.0.0' && toVersion === '2.0.0') {
        return {
          ...payload,
          // v2 renamed 'userName' to 'displayName'
          displayName: payload.userName ?? payload.displayName,
        };
      }
      return payload;
    },
  }),
  inject: [ConfigService],
});

The EventHubWebhookGuard automatically calls your migration function before your handler runs. If no migration function is configured, or versions match, the payload is delivered as-is.

Version Strategy

• Use semantic versioning for schemaVersion
• Bump the major version when making breaking changes to event payloads
• All events from a producer share the same version (it's per-service, not per-event)
• Existing producers default to '1.0.0' if they registered before this feature

Error Handling

• Return 2xx when you've processed or queued the event
• Return non-2xx if you can't process (Event Hub will retry)
• Retry schedule: immediate → 5s → 30s → dead-letter

Startup Behavior

The client handles startup ordering automatically:

1. Service starts immediately (doesn't wait for Event Hub)
2. Background registration begins with exponential backoff
3. Events are queued in memory while registration is pending
4. Queue is flushed automatically after successful registration

License

MIT