@solarios/hr-shared-events

The single source of truth for all RabbitMQ contracts across the HR microservices platform. Every service that communicates over RabbitMQ installs this package and imports from it — never hardcode queue names, routing keys, or message shapes in your service.

Installation

npm install solarios

What this package contains

| Export | Purpose | |--------|---------| | Exchanges | Exchange name constants | | RoutingKeys | Routing key string constants | | QueueNames | Queue name string constants | | RABBITMQ_DEFAULTS | Shared configuration (timeouts, prefetch, options) | | RabbitMessage<T> | Standard message envelope | | RpcResponse<T> | Standard RPC response envelope | | Event type interfaces | Typed payload shapes for every event | | RPC request/response interfaces | Typed shapes for every RPC contract |

Exchanges

Exchanges are the routing hubs. Every service declares the exchanges it uses.

import { Exchanges } from "solarios";

// Available exchanges
Exchanges.AUTH     // "auth"     — auth service events
Exchanges.PLATFORM // "platform" — platform service events
Exchanges.EDM      // "edm"      — employee data management events
Exchanges.PAYROLL  // "payroll"  — payroll processing events
Exchanges.AUDIT    // "audit"    — audit trail events

Rule: Publishers use the exchange that matches their service domain. If EDM publishes an event, it goes to Exchanges.EDM. If Auth publishes, it goes to Exchanges.AUTH.

Routing Keys

Routing keys tell the exchange which queues to deliver a message to. Pattern: <exchange>.<entity>.<action>

import { RoutingKeys } from "solarios";

// Auth events
RoutingKeys.auth.userLoggedIn    // "auth.user.logged_in"
RoutingKeys.auth.userLoggedOut   // "auth.user.logged_out"
RoutingKeys.auth.tokenRefreshed  // "auth.token.refreshed"
RoutingKeys.auth.sessionRevoked  // "auth.session.revoked"

// EDM events
RoutingKeys.edm.userCreated      // "edm.user.created"
RoutingKeys.edm.userUpdated      // "edm.user.updated"
RoutingKeys.edm.userDeleted      // "edm.user.deleted"
RoutingKeys.edm.userUpdated      // "edm.user.updated"
RoutingKeys.edm.employeeUpdated  // "edm.employee.updated"

// Payroll events
RoutingKeys.payroll.runInitiated  // "payroll.run.initiated"
RoutingKeys.payroll.runCompleted  // "payroll.run.completed"
RoutingKeys.payroll.runFailed     // "payroll.run.failed"

Queue Names

Queues are the mailboxes. Each service has its own queues. Queue names follow the pattern: q.<service>.<event> for pub/sub and rpc.<service>.<action>.request for RPC.

import { QueueNames } from "solarios";

// EDM queues (Auth consumes these)
QueueNames.edm.userCreated    // "q.edm.user.created"
QueueNames.edm.userUpdated    // "q.edm.user.updated"
QueueNames.edm.userDeleted    // "q.edm.user.deleted"
QueueNames.edm.sessionRevoked // "q.edm.session.revoked"

// Payroll queues
QueueNames.payroll.sessionRevoked  // "q.payroll.session.revoked"
QueueNames.payroll.runCompleted    // "q.payroll.run.completed"

// RPC queues
QueueNames.rpc.companyModules.request  // "rpc.company.modules.request"
QueueNames.rpc.tokenRefresh.request    // "rpc.auth.token.refresh.request"
QueueNames.rpc.employeeData.request    // "rpc.edm.employee.data.request"

Rule: Never define queue name strings in your service. Always import from QueueNames.

RABBITMQ_DEFAULTS

Shared configuration values every service should use:

import { RABBITMQ_DEFAULTS } from "solarios";

RABBITMQ_DEFAULTS.EXCHANGE_TYPE     // "topic"
RABBITMQ_DEFAULTS.EXCHANGE_OPTIONS  // { durable: true }
RABBITMQ_DEFAULTS.QUEUE_OPTIONS     // { durable: true }
RABBITMQ_DEFAULTS.PREFETCH_COUNT    // 10
RABBITMQ_DEFAULTS.RPC_TIMEOUT_MS    // 10000 (10 seconds)

Use these when asserting exchanges and queues:

await channel.assertExchange(
  Exchanges.EDM,
  RABBITMQ_DEFAULTS.EXCHANGE_TYPE,
  RABBITMQ_DEFAULTS.EXCHANGE_OPTIONS
);

await channel.assertQueue(
  QueueNames.edm.userCreated,
  RABBITMQ_DEFAULTS.QUEUE_OPTIONS
);

Message Envelope — RabbitMessage<T>

Every message published on the bus is wrapped in this envelope automatically by the publish() helper in your service's rabbitmq.service.ts.

import type { RabbitMessage } from "solarios";

// What every message looks like on the wire:
interface RabbitMessage<TPayload> {
  messageId:   string;  // UUID — use for deduplication and tracing
  messageType: string;  // mirrors the routing key
  timestamp:   string;  // ISO-8601 publish time
  payload:     TPayload; // your typed domain data
}

When consuming, unwrap like this:

const envelope = JSON.parse(
  msg.content.toString()
) as RabbitMessage<EdmUserCreatedEvent>;

const { messageId, timestamp, payload } = envelope;
// payload is fully typed as EdmUserCreatedEvent

Event Type Interfaces

Auth Events

import type {
  UserLoggedInEvent,
  UserLoggedOutEvent,
  TokenRefreshedEvent,
  SessionRevokedEvent,
} from "solarios";

// SessionRevokedEvent — published when user logs out or admin revokes session
interface SessionRevokedEvent {
  authSessionId: string;
  userId:        string;
  revokedAt:     string;
}

EDM Events

import type {
  EdmUserCreatedEvent,
  EdmUserUpdatedEvent,
  EdmUserDeletedEvent,
} from "solarios";

interface EdmUserCreatedEvent {
  userId:    string;  // EDM employee id e.g. "KACHA-EMP-1"
  companyId: string;
  email:     string;
  firstName: string;
  lastName:  string;
  createdAt: string;
}

interface EdmUserUpdatedEvent {
  userId:        string;
  companyId:     string;
  updatedFields: Record<string, unknown>; // only changed fields
  updatedAt:     string;
}

interface EdmUserDeletedEvent {
  userId:    string;
  companyId: string;
  deletedAt: string;
}

Payroll Events

import type {
  PayrollRunInitiatedEvent,
  PayrollRunCompletedEvent,
  PayrollRunFailedEvent,
} from "solarios";

Employee Sync Types (EDM ↔ Payroll RPC)

import type {
  EmployeeSyncRecord,
  EmployeeDataRequest,
  EmployeeDataResponse,
  EmployeeUpdatedEvent,
} from "solarios";

RPC Contract Interfaces

Company Modules (Auth → Platform)

import type {
  GetCompanyModulesRequest,
  GetCompanyModulesResponse,
  CompanyModuleDto,
} from "solarios";

// Request: Auth sends this
interface GetCompanyModulesRequest {
  companyId: string;
}

// Response: Platform replies with this
interface GetCompanyModulesResponse {
  companyId: string;
  modules:   CompanyModuleDto[];
}

Token Refresh (Payroll/EDM → Auth)

import type {
  TokenRefreshRequest,
  TokenRefreshResponse,
} from "solarios";

Employee Data (Payroll → EDM)

import type {
  EmployeeDataRequest,
  EmployeeDataResponse,
} from "solarios";

Making changes to this package

1. Edit the source files in src/
2. Run npm run build to compile to dist/
3. Bump the version in package.json
4. Run npm publish to push to npm
5. In each service that uses it: npm install solarios@latest
6. If you added a new queue name, also add it to hr-infrastructure/rabbitmq/definitions.json

Never make breaking changes to existing interfaces without coordinating with all service owners. Adding new optional fields is safe. Renaming or removing fields is a breaking change.

How to connect any service using solarios

This guide walks you step by step through adding RabbitMQ communication to a new service. By the end, your service will be able to publish events, consume events, and make RPC calls — all using solarios as the contract layer.

The two communication patterns

Before writing any code, decide which pattern your connection uses:

Pub/Sub — fire and forget

• One service publishes an event
• One or more services consume it independently
• Publisher does not wait for consumers
• Used for: employee created/updated/deleted, session revoked, payroll completed

RPC — request and reply

• One service sends a request and waits for a reply
• One service receives the request, processes it, and replies
• Used for: Auth asking Platform for company modules, Payroll asking EDM for employee data, token refresh

Step 1 — Install solarios

npm install solarios

Step 2 — Create the connection layer

Every service needs two files. These are identical across all services — copy them from any existing service.

src/integration/rabbitmq/rabbitmq.adapter.ts

Manages the TCP connection to RabbitMQ. Provides an in-memory fallback for local development without Docker.

import { connect, type Channel, type ChannelModel } from "amqplib";

function createFallbackChannel(): Channel { /* ... */ }

let connection: ChannelModel | null = null;
let channel: Channel | null = null;

export async function initializeRabbitMQ(): Promise<void> {
  try {
    connection = await connect(process.env.RABBITMQ_URL!);
    channel    = await connection.createChannel();
  } catch {
    channel = createFallbackChannel();
  }
}

export function getRabbitMQChannel(): Channel {
  if (!channel) throw new Error("Call initializeRabbitMQ() first");
  return channel;
}

export async function shutdownRabbitMQ(): Promise<void> {
  await channel?.close();
  await connection?.close();
}

src/integration/rabbitmq/rabbitmq.service.ts

High-level wrappers. Your business code never touches the raw channel.

import { v4 as uuidv4 } from "uuid";
import { RABBITMQ_DEFAULTS, type RabbitMessage } from "solarios";
import { getRabbitMQChannel } from "./rabbitmq.adapter";

// Declare which exchanges this service uses
export async function declareExchanges(): Promise<void> {
  const ch = getRabbitMQChannel();
  await ch.assertExchange("edm",  "topic", { durable: true });
  await ch.assertExchange("auth", "topic", { durable: true });
}

// Wrap payload in RabbitMessage envelope and publish
export function publish<T>(exchange: string, routingKey: string, payload: T): void {
  const envelope: RabbitMessage<T> = {
    messageId:   uuidv4(),
    messageType: routingKey,
    timestamp:   new Date().toISOString(),
    payload,
  };
  getRabbitMQChannel().publish(
    exchange,
    routingKey,
    Buffer.from(JSON.stringify(envelope)),
    { persistent: true }
  );
}

// Consume messages with automatic ack/nack
export async function consume<T>(
  queue:   string,
  handler: (payload: T) => Promise<void>
): Promise<void> {
  const ch = getRabbitMQChannel();
  await ch.prefetch(RABBITMQ_DEFAULTS.PREFETCH_COUNT);
  ch.consume(queue, async (msg) => {
    if (!msg) return;
    try {
      const { payload } = JSON.parse(msg.content.toString()) as RabbitMessage<T>;
      await handler(payload);
      ch.ack(msg);
    } catch {
      ch.nack(msg, false, false); // dead-letter on error
    }
  });
}

// Send directly to a queue (used for RPC replies)
export function sendToQueue(queue: string, payload: unknown, options?: object): void {
  getRabbitMQChannel().sendToQueue(
    queue,
    Buffer.from(JSON.stringify(payload)),
    { contentType: "application/json", ...options }
  );
}

Step 3 — Hook into your service bootstrap

In your main.ts or server.ts, add RabbitMQ initialization before HTTP:

import { initializeRabbitMQ, shutdownRabbitMQ } from "./integration/rabbitmq/rabbitmq.adapter";
import { declareExchanges }  from "./integration/rabbitmq/rabbitmq.service";
import { startAllConsumers } from "./integration/events/consumers/index";
import { startAllRpcServers } from "./integration/rpc/servers/index";

async function bootstrap() {
  // RabbitMQ MUST be ready before HTTP opens
  await initializeRabbitMQ();
  await declareExchanges();
  await startAllConsumers();
  await startAllRpcServers();

  // HTTP last
  app.listen(PORT);

  // Graceful shutdown
  process.on("SIGTERM", async () => {
    server.close(async () => {
      await shutdownRabbitMQ();
      process.exit(0);
    });
  });
}

Step 4a — Publishing an event (pub/sub)

Use this pattern when your service needs to notify other services that something happened.

Create a publisher file

// src/integration/events/publishers/employee-created.publisher.ts
import { Exchanges, RoutingKeys, type EdmUserCreatedEvent } from "solarios";
import { publish } from "../../rabbitmq/rabbitmq.service";

export function publishEmployeeCreated(event: EdmUserCreatedEvent): void {
  publish<EdmUserCreatedEvent>(
    Exchanges.EDM,           // your service's exchange
    RoutingKeys.edm.userCreated, // routing key from solarios
    event                    // typed payload
  );
}

Export it

// src/integration/events/publishers/index.ts
export { publishEmployeeCreated } from "./employee-created.publisher";

Call it from your business logic

import { publishEmployeeCreated } from "src/integration/events/publishers";

// After creating employee in DB:
publishEmployeeCreated({
  userId:    newEmployee.id,
  companyId: String(companyId),
  email:     "",
  firstName: "John",
  lastName:  "Doe",
  createdAt: new Date().toISOString(),
});
// Don't await — fire and forget

Important: Call publish after the database transaction succeeds. Never publish before saving to DB — if the DB write fails, you will have published an event for something that doesn't exist.

Step 4b — Consuming an event (pub/sub)

Use this pattern when your service needs to react to something another service published.

Create a consumer file

// src/integration/events/consumers/session-revoked.consumer.ts
import {
  Exchanges, QueueNames, RoutingKeys,
  RABBITMQ_DEFAULTS,
  type RabbitMessage, type SessionRevokedEvent,
} from "solarios";
import { getRabbitMQChannel } from "../../rabbitmq/rabbitmq.adapter";
import prisma from "../../config/database";

export async function startSessionRevokedConsumer(): Promise<void> {
  const channel = getRabbitMQChannel();

  // 1. Make sure the exchange exists
  await channel.assertExchange(Exchanges.AUTH, "topic", {
    ...RABBITMQ_DEFAULTS.EXCHANGE_OPTIONS,
  });

  // 2. Declare your service's queue
  await channel.assertQueue(QueueNames.edm.sessionRevoked, {
    ...RABBITMQ_DEFAULTS.QUEUE_OPTIONS,
  });

  // 3. Bind the queue to the exchange with the routing key
  await channel.bindQueue(
    QueueNames.edm.sessionRevoked,  // your queue
    Exchanges.AUTH,                  // source exchange
    RoutingKeys.auth.sessionRevoked  // routing key to match
  );

  // 4. Set prefetch — how many messages to process at once
  await channel.prefetch(RABBITMQ_DEFAULTS.PREFETCH_COUNT);

  // 5. Start consuming
  channel.consume(
    QueueNames.edm.sessionRevoked,
    async (msg) => {
      if (!msg) return;
      try {
        // Unwrap the envelope
        const { payload } = JSON.parse(msg.content.toString())
          as RabbitMessage<SessionRevokedEvent>;

        // Your business logic
        await prisma.serviceSession.updateMany({
          where: { authSessionId: payload.authSessionId, isActive: true },
          data:  { isActive: false },
        });

        channel.ack(msg); // success — remove from queue
      } catch (err) {
        channel.nack(msg, false, false); // failure — send to dead-letter
      }
    },
    { noAck: false }
  );
}

Register it

// src/integration/events/consumers/index.ts
import { startSessionRevokedConsumer } from "./session-revoked.consumer";

export async function startAllConsumers(): Promise<void> {
  await startSessionRevokedConsumer();
  // add more consumers here as you add them
}

Step 4c — Making an RPC call (caller side)

Use this pattern when your service needs to ask another service a question and wait for the answer.

// src/integration/rpc/clients/platform.rpc.client.ts
import { v4 as uuidv4 } from "uuid";
import {
  QueueNames, RABBITMQ_DEFAULTS,
  type GetCompanyModulesRequest,
  type GetCompanyModulesResponse,
} from "solarios";
import { getRabbitMQChannel } from "../../rabbitmq/rabbitmq.adapter";

export async function getCompanyModules(
  companyId: string
): Promise<GetCompanyModulesResponse> {
  const channel       = getRabbitMQChannel();
  const correlationId = uuidv4();

  // Create a temporary reply queue — exclusive to this call
  const { queue: replyQueue } = await channel.assertQueue("", {
    exclusive: true, autoDelete: true,
  });

  return new Promise((resolve) => {
    // Timeout — give up after RPC_TIMEOUT_MS
    const timeout = setTimeout(() => {
      resolve({ companyId, modules: [] }); // degrade gracefully
    }, RABBITMQ_DEFAULTS.RPC_TIMEOUT_MS);

    // Listen on reply queue BEFORE sending
    channel.consume(replyQueue, (msg) => {
      if (!msg) return;
      if (msg.properties.correlationId !== correlationId) return; // safety check

      clearTimeout(timeout);
      channel.ack(msg);
      resolve(JSON.parse(msg.content.toString()) as GetCompanyModulesResponse);
    }, { noAck: false });

    // Send the request
    channel.sendToQueue(
      QueueNames.rpc.companyModules.request,
      Buffer.from(JSON.stringify({ companyId } as GetCompanyModulesRequest)),
      {
        correlationId,
        replyTo:    replyQueue,
        persistent: false,
        expiration: String(RABBITMQ_DEFAULTS.RPC_TIMEOUT_MS),
      }
    );
  });
}

Usage in business logic:

// Await it — RPC is synchronous from the caller's perspective
const { modules } = await getCompanyModules(user.companyId);

Step 4d — Serving an RPC request (server side)

Use this pattern when your service needs to respond to requests from other services.

// src/integration/rpc/servers/company-modules.rpc.server.ts
import {
  QueueNames, RABBITMQ_DEFAULTS,
  type GetCompanyModulesRequest, type GetCompanyModulesResponse,
} from "solarios";
import { getRabbitMQChannel } from "../../rabbitmq/rabbitmq.adapter";
import { sendToQueue } from "../../rabbitmq/rabbitmq.service";
import prisma from "../../config/database";

export async function startCompanyModulesRpcServer(): Promise<void> {
  const channel = getRabbitMQChannel();

  await channel.assertQueue(QueueNames.rpc.companyModules.request, {
    ...RABBITMQ_DEFAULTS.QUEUE_OPTIONS,
  });

  // prefetch(1) — one request at a time for DB-heavy operations
  await channel.prefetch(1);

  channel.consume(
    QueueNames.rpc.companyModules.request,
    async (msg) => {
      if (!msg) return;

      // These two properties are REQUIRED for RPC — discard if missing
      const { replyTo, correlationId } = msg.properties;
      if (!replyTo || !correlationId) {
        channel.nack(msg, false, false);
        return;
      }

      let response: GetCompanyModulesResponse;

      try {
        const { companyId } = JSON.parse(msg.content.toString())
          as GetCompanyModulesRequest;

        const modules = await prisma.companyModule.findMany({
          where: { companyId, isActive: true },
        });

        response = { companyId, modules };
      } catch {
        response = { companyId: "", modules: [] }; // always reply, never leave caller hanging
      }

      // Reply to caller's temporary queue
      sendToQueue(replyTo, response, { correlationId });
      channel.ack(msg);
    },
    { noAck: false }
  );
}

Register it

// src/integration/rpc/servers/index.ts
import { startCompanyModulesRpcServer } from "./company-modules.rpc.server";

export async function startAllRpcServers(): Promise<void> {
  await startCompanyModulesRpcServer();
}

Step 5 — Add queues to definitions.json

Every new queue your service introduces must be added to hr-infrastructure/rabbitmq/definitions.json.

In the "queues" array:

{
  "name": "q.your-service.event-name",
  "vhost": "/",
  "durable": true,
  "auto_delete": false,
  "arguments": {}
}

In the "bindings" array:

{
  "source": "your-exchange",
  "vhost": "/",
  "destination": "q.your-service.event-name",
  "destination_type": "queue",
  "routing_key": "your-exchange.entity.action",
  "arguments": {}
}

Then restart RabbitMQ:

cd hr-infrastructure
docker compose restart rabbitmq

Step 6 — Add the queue name to solarios

Add the queue name to QueueNames in hr-shared-events/src/constants.ts:

yourService: {
  eventName: "q.your-service.event-name",
},

Rebuild and republish:

cd hr-shared-events
npm run build
npm publish

Update in your service:

npm install solarios@latest

Checklist — connecting a new service

• [ ] npm install solarios in the service
• [ ] Copy rabbitmq.adapter.ts from an existing service
• [ ] Copy rabbitmq.service.ts from an existing service
• [ ] Call initializeRabbitMQ() and declareExchanges() at bootstrap before HTTP
• [ ] For pub/sub publishing: create publisher file, call after DB write succeeds
• [ ] For pub/sub consuming: create consumer file, register in consumers/index.ts
• [ ] For RPC calling: create client file, await the result
• [ ] For RPC serving: create server file, always reply even on error, use prefetch(1)
• [ ] Add new queue names to solarios QueueNames
• [ ] Add new queues and bindings to definitions.json
• [ ] Restart RabbitMQ after updating definitions.json
• [ ] Update solarios version in all affected services

Common mistakes

Queue name mismatch — Publisher sends to "q.edm.user.created" but consumer listens on "q.edm.users.created". Messages arrive at RabbitMQ but no consumer picks them up. Always import from QueueNames.

Publishing before DB commit — If the transaction fails after publish, you have notified consumers about something that doesn't exist. Always publish after a successful transaction.

Not replying in RPC server — If your server throws without sending a reply, the caller waits until timeout (10 seconds) then gives up. Always send a reply even on error:

// Even on error — always reply
sendToQueue(replyTo, { error: "processing_failed" }, { correlationId });
channel.ack(msg);

Missing prefetch on RPC servers — Without prefetch(1), RabbitMQ pushes all queued requests at once. For DB-heavy operations this causes connection pool exhaustion.

Forgetting to ack — If you don't call channel.ack(msg), RabbitMQ keeps the message in the queue and re-delivers it to the next available consumer. This causes the same message to be processed multiple times.

Starting HTTP before RabbitMQ — If a request arrives and triggers an RPC call before the channel is ready, it throws. Always initialize RabbitMQ before starting the HTTP server.