🔌 Creating a WebSocket Service

This guide walks you through creating a complete WebSocket service using the Nodeblocks SDK. We'll build three distinct WebSocket patterns: INCOMING (client → server), OUTGOING (server → client), and BIDIRECTIONAL (client ↔ server).

📦 Required Packages: This example uses several packages. MongoDB must run as a replica set for change streams. Make sure to install them:

npm install express mongodb rxjs ws ramda dotenv cors @nodeblocks/backend-sdk
npm install @types/node @types/express @types/ramda @types/cors @types/ws --save-dev

⚙️ Module setup: The server bootstrap uses top-level await. Run it as an ESM module (for example with module: "NodeNext" in tsconfig.json and "type": "module" in package.json) or move the bootstrap code into an async function main().

---

🏗️ Service Architecture

In order to build the WebSocket service, we will implement the following core components:

1. WebSocket Features - Define WebSocket endpoints with different flow patterns
2. Service - Factory function that wires everything together
3. Server Setup - HTTP + WebSocket server configuration

When a client omits emitterId, the SDK adds a unique per-connection value to its inbound message and filters outbound traffic with notFromEmitter(socketId). This is not enforced: the current SDK merges the client payload after its generated value, so a client-supplied emitterId can override it. Do not make authorization or routing decisions based on this field until the SDK changes that merge order.

⚠️ Valid JSON only: The SDK currently calls JSON.parse without handling malformed WebSocket frames. Send JSON payloads only; malformed frames can throw from the connection handler rather than producing a controlled WebSocket error.

---

1️⃣ Create WebSocket Features

First, create a websocket.ts file inside the src/features directory.
Here we define four WebSocket features that demonstrate the main communication flows.

src/features/websocket.ts

import {Subject, interval, timer} from 'rxjs';
import {filter} from 'rxjs/operators';
import {primitives} from '@nodeblocks/backend-sdk';

const {compose, withRoute, withSchema, markAsFromEmitter, notFromEmitter} = primitives;

// 📥 INCOMING ONLY: Clients send messages to server
export const incomingMessagesFeature = compose(
  withSchema({
    type: 'object',
    properties: {
      name: {type: 'string'},
      message: {type: 'string'},
    },
    required: ['name'],
  }),
  withRoute({
    handler: (_: primitives.WsRouteHandlerPayload) => {
      const subject = new Subject();

      // INCOMING: Listen for messages from clients
      subject.subscribe(data => {
        console.log('Message received from client:', data);
        // Process the incoming message (save to DB, etc.)
        // Note: We DON'T send anything back via subject.next()
      });

      return subject;
    },
    path: '/api/messages',
    protocol: 'ws',
  }),
);

// 📤 OUTGOING ONLY: Server pushes data to the connected client
export const outgoingNotificationsFeature = compose(
  withRoute({
    handler: () => {
      const subject = new Subject();
      // OUTGOING: Send periodic notifications to the connected client
      const notificationsSubscription = interval(5000).subscribe(() => {
        subject.next({
          type: 'server_notification',
          message: 'Server heartbeat',
          timestamp: Date.now(),
        });
      });

      const cleanup = () => {
        notificationsSubscription.unsubscribe();
      };

      // Note: We DON'T subscribe to incoming messages
      subject.subscribe({
        complete: cleanup,
        error: cleanup,
      });

      return subject;
    },
    path: '/api/notifications',
    protocol: 'ws',
  }),
);

// 📤 OUTGOING ONLY: Database change streams
export const databaseChangesFeature = compose(
  withRoute({
    handler: (request: primitives.WsRouteHandlerPayload) => {
      const subject = new Subject();

      // Watch for changes in the profiles collection
      const changeStream = request.context.db.profiles.watch();

      changeStream.on('change', (data: unknown) => {
        console.log('Change stream data', data);
        // OUTGOING: Send data to the connected client
        subject.next({
          timestamp: Date.now(),
          type: 'profile_change',
          data,
        });
      });

      changeStream.on('error', (error: Error) => {
        console.error('Change stream error:', error);
        subject.error(error);
      });

      // Cleanup when WebSocket closes
      let subscription: {unsubscribe: () => void} | undefined;
      const cleanup = () => {
        void changeStream.close().catch(error => console.error('Change stream cleanup error:', error));
        if (subscription) subscription.unsubscribe();
      };

      subscription = subject.subscribe({
        complete: cleanup,
        error: cleanup,
      });

      return subject;
    },
    path: '/api/database-changes',
    protocol: 'ws',
  }),
);

// 🔄 BIDIRECTIONAL: Chat system (both incoming and outgoing)
type ChatMessagePayload = {
  username: string;
  message: string;
  roomId?: string;
  type?: string;
  timestamp?: number;
  emitterId?: string;
  [key: string]: unknown;
};

export const chatFeature = compose(
  withSchema({
    type: 'object',
    properties: {
      username: {type: 'string'},
      message: {type: 'string'},
      roomId: {type: 'string'},
    },
    required: ['username', 'message'],
  }),
  withRoute({
    handler: () => {
      // The SDK normally adds a connection-specific emitterId and filters
      // outbound traffic with notFromEmitter(socketId), but clients can
      // override it. Never trust it for authorization or routing.
      // The local emitter id below is only for app-level echo suppression.
      const localEmitterId = 'chat-handler';
      const subject = new Subject<ChatMessagePayload>();

      // INCOMING: Listen for messages from this client
      subject.pipe(filter(notFromEmitter(localEmitterId))).subscribe(data => {
        console.log('Chat message received:', data);

        // OUTGOING: Send the chat message back through this connection
        subject.next(
          markAsFromEmitter(localEmitterId, {
            type: 'chat_message',
            username: data.username,
            message: data.message,
            timestamp: Date.now(),
          }) as ChatMessagePayload,
        );
      });

      // OUTGOING: Send welcome message when client connects
      timer(1000).subscribe(() => {
        subject.next(
          markAsFromEmitter(localEmitterId, {
            type: 'welcome',
            message: 'Welcome to the chat!',
            timestamp: Date.now(),
          }) as ChatMessagePayload,
        );
      });

      return subject;
    },
    path: '/api/chat',
    protocol: 'ws',
  }),
);

Tip: For production chat streaming, prefer the built-in streamChatMessagesFeature from the Chat Service, which uses MongoDB change streams on chatMessages.

---

2️⃣ Create Service

Create a websocket.ts file inside the src/services directory.
The service composes the features and passes the WebSocket server into defService.

src/services/websocket.ts

import {partial} from 'ramda';
import {primitives} from '@nodeblocks/backend-sdk';
import {
  incomingMessagesFeature,
  outgoingNotificationsFeature,
  databaseChangesFeature,
  chatFeature,
} from '../features/websocket';

const {compose, defService} = primitives;

export const websocketService: primitives.Service = (dataStores, configuration, {webSocketServer} = {}) =>
  defService(
    partial(
      compose(
        incomingMessagesFeature, // 📥 INCOMING: Client messages
        outgoingNotificationsFeature, // 📤 OUTGOING: Server notifications
        databaseChangesFeature, // 📤 OUTGOING: Database changes
        chatFeature, // 🔄 BIDIRECTIONAL: Chat system
      ),
      [{dataStores, configuration}],
    ),
    webSocketServer,
  );

defService(adapter, webSocketServer) accepts the WebSocket server as its second argument. The public Service factory exposes it through the optional drivers object as { webSocketServer }, matching chatService.

---

3️⃣ Set Up Server

Create or update your index.ts file in the src directory.

src/index.ts

import {createServer} from 'http';
import 'dotenv/config';

import express from 'express';
import {MongoClient} from 'mongodb';
import {WebSocketServer} from 'ws';
import {middlewares, services} from '@nodeblocks/backend-sdk';
import cors from 'cors';
import {websocketService} from './services/websocket';

const {nodeBlocksErrorMiddleware} = middlewares;
const {authService, profileService} = services;

// Set up Express and WebSocket servers
const app = express();
const server = createServer(app);
const wss = new WebSocketServer({server});

// Connect once and reuse the collections across services.
const mongoClient = new MongoClient(process.env.MONGODB_URI || 'mongodb://localhost:27017/?authSource=admin', {
  auth: {
    password: process.env.MONGO_PASSWORD || 'password',
    username: process.env.MONGO_USER || 'user',
  },
});
await mongoClient.connect();
const db = mongoClient.db(process.env.MONGODB_DB_NAME || 'dev');
const dataStores = {
  identities: db.collection('identities'),
  onetimetokens: db.collection('onetimetokens'),
  organizations: db.collection('organizations'),
  products: db.collection('products'),
  profiles: db.collection('profiles'),
};

const authSecrets = {
  authEncSecret: 'your-encryption-secret',
  authSignSecret: 'your-signing-secret',
};

const identity = {
  typeIds: {
    admin: '100',
    guest: '000',
    regular: '001',
  },
};

// Configure CORS
app.use(
  cors({
    origin: '*',
    methods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'],
    allowedHeaders: ['*'],
  }),
);

// Add WebSocket service
app.use(
  websocketService(
    {
      profiles: dataStores.profiles,
    },
    {},
    {webSocketServer: wss},
  ),
);

// Add other services
app.use(
  authService(
    {
      identities: dataStores.identities,
      onetimetokens: dataStores.onetimetokens,
    },
    {
      authSecrets,
      maxFailedLoginAttempts: 5,
      accessTokenSignOptions: {expiresIn: '2h'},
      refreshTokenSignOptions: {expiresIn: '2d'},
      identity,
    },
    {
      mailService: {
        sendMail: mailData => {
          console.log('Auth email would be sent:', mailData);
          return Promise.resolve(true);
        },
      },
    },
  ),
);

app.use(
  profileService(
    {
      profiles: dataStores.profiles,
      identities: dataStores.identities,
      organizations: dataStores.organizations,
      products: dataStores.products,
    },
    {
      authSecrets,
      identity,
    },
  ),
);

// Error handling (must be last)
app.use(nodeBlocksErrorMiddleware());

// Start the server
const PORT = 8089;

server.listen(PORT, () => {
  console.log(`🚀 Server running on port ${PORT}`);
  console.log(`📡 WebSocket endpoints:`);
  console.log(`   📥 ws://localhost:${PORT}/api/messages (INCOMING: send messages)`);
  console.log(`   📤 ws://localhost:${PORT}/api/notifications (OUTGOING: receive notifications)`);
  console.log(`   📤 ws://localhost:${PORT}/api/database-changes (OUTGOING: receive DB changes)`);
  console.log(`   🔄 ws://localhost:${PORT}/api/chat (BIDIRECTIONAL: chat system)`);
});

const shutdown = () => {
  server.close(() => {
    void mongoClient.close().finally(() => process.exit(0));
  });
};

process.once('SIGINT', shutdown);
process.once('SIGTERM', shutdown);

---

4️⃣ Environment Setup

Create a .env file in your project root:

.env

MONGODB_URI=mongodb://localhost:27017/?authSource=admin
MONGODB_DB_NAME=your_app_database
MONGO_USER=user
MONGO_PASSWORD=password

---

🧪 Testing the Service

Test with Command Line

# Install wscat for command line testing
npm install -g wscat

# 📥 INCOMING: Send messages to server (you send, server receives)
wscat -c ws://localhost:8089/api/messages
> {"name": "John", "message": "Hello from client!"}

# 📤 OUTGOING: Receive notifications from server (server sends, you receive)
wscat -c ws://localhost:8089/api/notifications
# You'll see periodic server heartbeat messages

# 📤 OUTGOING: Receive database changes (server sends DB changes)
wscat -c ws://localhost:8089/api/database-changes
# You'll see database change notifications

# 🔄 BIDIRECTIONAL: Chat (both send and receive)
wscat -c ws://localhost:8089/api/chat
> {"username": "CLIUser", "message": "Hello chat!"}
# You'll see your message echoed back and a welcome message

Test with WebSocket Client

test-client.js

import WebSocket from 'ws';

// 📥 INCOMING: Send messages to server
const messageWs = new WebSocket('ws://localhost:8089/api/messages');
messageWs.on('open', () => {
  console.log('📥 Connected for sending messages');
  messageWs.send(
    JSON.stringify({
      name: 'NodeJS Client',
      message: 'Hello from Node.js!',
    }),
  );
});

// 📤 OUTGOING: Receive notifications from server
const notificationWs = new WebSocket('ws://localhost:8089/api/notifications');
notificationWs.on('open', () => {
  console.log('📤 Connected to receive notifications');
});
notificationWs.on('message', data => {
  console.log('📤 Notification received:', JSON.parse(data.toString()));
});

// 📤 OUTGOING: Receive database changes
const dbWs = new WebSocket('ws://localhost:8089/api/database-changes');
dbWs.on('open', () => {
  console.log('📤 Connected to receive database changes');
});
dbWs.on('message', data => {
  console.log('📤 Database change:', JSON.parse(data.toString()));
});

// 🔄 BIDIRECTIONAL: Chat system
const chatWs = new WebSocket('ws://localhost:8089/api/chat');
chatWs.on('open', () => {
  console.log('🔄 Connected to chat');
  chatWs.send(
    JSON.stringify({
      username: 'NodeJS User',
      message: 'Hello chat!',
    }),
  );
});
chatWs.on('message', data => {
  console.log('🔄 Chat message:', JSON.parse(data.toString()));
});

---

➡️ Next Steps

Now you can practice by adding more functionality to your WebSocket service:

• Authentication - Add authentication validators to secure WebSocket connections
• Rate Limiting - Prevent WebSocket abuse with connection limits
• Message Persistence - Store WebSocket messages in the database
• Broadcasting - Send messages to specific groups of clients
• Built-in chat streaming - Use chatService with { webSocketServer } for production messaging

🔗 Related Documentation

• Creating a Custom Service - Learn the basic service creation patterns
• Chat Service - Production WebSocket chat streaming
• Route Component - Understand route configuration options
• Authentication Service - Add identity authentication to your app
• Profile Service - Profile API used alongside this example