バージョン: 0.14.0 (最新)

🔌 WebSocket サービスの作成

このガイドでは Nodeblocks SDK を使った完全な WebSocket サービスを作成します。実装する通信パターンは、受信（クライアント → サーバー）、送信（サーバー → クライアント）、双方向（クライアント ↔ サーバー）です。

📦 必要なパッケージ: この例では複数のパッケージを使用します。MongoDB change stream のため、MongoDB はレプリカセットとして動作している必要があります。以下をインストールしてください。
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
⚙️ モジュール設定: サーバーブートストラップはトップレベル await を使用します。ESM モジュールとして実行するか（例: tsconfig.json の module: "NodeNext" と package.json の "type": "module"）、ブートストラップを async function main() に移してください。

🏗️ サービスアーキテクチャ

WebSocket 機能 — 異なるフローのエンドポイントを定義します。
サービス — 機能を結合するファクトリー関数です。
サーバーセットアップ — HTTP と WebSocket サーバーを構成します。

クライアントが emitterId を省略すると、SDK は受信メッセージへ接続ごとの値を追加し、notFromEmitter(socketId) で送信トラフィックをフィルターします。ただし現在の SDK は生成値の後にクライアントペイロードをマージするため、クライアント指定の emitterId が上書きできます。SDK のマージ順が変更されるまで、認可やルーティングの判断にこのフィールドを使用しないでください。

有効な JSON のみ: SDK は不正な WebSocket フレームを処理せずに JSON.parse を呼びます。JSON ペイロードだけを送信してください。不正なフレームは制御された WebSocket エラーではなく接続ハンドラーからスローされることがあります。

1. WebSocket 機能を作成する

まず、src/features ディレクトリ内に websocket.ts ファイルを作成します。ここでは、主な通信フローを示す 4 つの WebSocket 機能を定義します。

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;

// 📥 受信のみ: クライアントからサーバーへメッセージを送る
export const incomingMessagesFeature = compose(
  withSchema({
    type: 'object',
    properties: {
      name: {type: 'string'},
      message: {type: 'string'},
    },
    required: ['name'],
  }),
  withRoute({
    handler: (_: primitives.WsRouteHandlerPayload) => {
      const subject = new Subject();

      // 受信: クライアントからのメッセージを監視する
      subject.subscribe(data => {
        console.log('Message received from client:', data);
        // 受信メッセージを処理する（DB への保存など）
        // 注: subject.next() では何も返信しない
      });

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

// 📤 送信のみ: サーバーが接続済みクライアントへデータをプッシュする
export const outgoingNotificationsFeature = compose(
  withRoute({
    handler: () => {
      const subject = new Subject();
      // 送信: 接続済みクライアントへ定期通知を送る
      const notificationsSubscription = interval(5000).subscribe(() => {
        subject.next({
          type: 'server_notification',
          message: 'Server heartbeat',
          timestamp: Date.now(),
        });
      });

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

      // 注: 受信メッセージは購読しない
      subject.subscribe({
        complete: cleanup,
        error: cleanup,
      });

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

// 📤 送信のみ: データベースの変更ストリーム
export const databaseChangesFeature = compose(
  withRoute({
    handler: (request: primitives.WsRouteHandlerPayload) => {
      const subject = new Subject();

      // profiles コレクションの変更を監視する
      const changeStream = request.context.db.profiles.watch();

      changeStream.on('change', (data: unknown) => {
        console.log('Change stream data', data);
        // 送信: 接続済みクライアントへデータを送る
        subject.next({
          timestamp: Date.now(),
          type: 'profile_change',
          data,
        });
      });

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

      // WebSocket を閉じるときにクリーンアップする
      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',
  }),
);

// 🔄 双方向: チャットシステム（受信と送信の両方）
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: () => {
      // SDK は通常、接続固有の emitterId を追加し、
      // notFromEmitter(socketId) で送信トラフィックをフィルターします。
      // ただしクライアントが上書きできるため、認可やルーティングに信用してはいけません。
      // 以下のローカル emitter ID はアプリレベルのエコー抑止にのみ使用します。
      const localEmitterId = 'chat-handler';
      const subject = new Subject<ChatMessagePayload>();

      // 受信: このクライアントからのメッセージを監視する
      subject.pipe(filter(notFromEmitter(localEmitterId))).subscribe(data => {
        console.log('Chat message received:', data);

        // 送信: この接続を通じてチャットメッセージを返信する
        subject.next(
          markAsFromEmitter(localEmitterId, {
            type: 'chat_message',
            username: data.username,
            message: data.message,
            timestamp: Date.now(),
          }) as ChatMessagePayload,
        );
      });

      // 送信: クライアント接続時に歓迎メッセージを送る
      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',
  }),
);

本番のチャットストリーミングには、MongoDB の chatMessages change stream を使う Chat サービスの組み込み streamChatMessagesFeature を優先してください。

2. サービスを作成する

src/services/websocket.ts で機能を合成し、WebSocket サーバーを defService に渡します。

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, outgoingNotificationsFeature, databaseChangesFeature, chatFeature), [{dataStores, configuration}]),
    webSocketServer,
  );

defService(adapter, webSocketServer) は WebSocket サーバーを第 2 引数として受け取ります。公開 Service ファクトリーは、chatService と同様に任意のドライバーオブジェクト { webSocketServer } を通じて渡します。

3. サーバーをセットアップする

src ディレクトリにある index.ts ファイルを作成または更新します。

Express アプリケーションから HTTP サーバーを作成し、それに WebSocketServer を接続します。MongoDB へ 1 度だけ接続して各サービスでコレクションを再利用し、WebSocket サービスをマウントします。

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;

// Express と WebSocket サーバーをセットアップする
const app = express();
const server = createServer(app);
const wss = new WebSocketServer({server});

// 一度だけ接続し、各サービスでコレクションを再利用する。
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'),
  refreshtokens: db.collection('refreshtokens'),
  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',
  },
};

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

// WebSocket サービスを追加する
app.use(
  websocketService(
    {
      profiles: dataStores.profiles,
    },
    {},
    {webSocketServer: wss},
  ),
);

// 他のサービスを追加する
app.use(
  authService(
    {
      identities: dataStores.identities,
      refreshtokens: dataStores.refreshtokens,
      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,
    },
  ),
);

// エラー処理（必ず最後に登録する）
app.use(nodeBlocksErrorMiddleware());

// サーバーを起動する
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);

認証やプロフィールサービスも同じ Express アプリケーションへマウントできます。CORS は必要なオリジン、メソッド、ヘッダーに合わせて構成してください。

4. 環境設定

プロジェクトルートに .env を作成します。

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

サービスをテストする

コマンドラインでテストする

wscat を使う場合は npm install -g wscat を実行します。

# 受信: クライアントが送信し、サーバーが受信
wscat -c ws://localhost:8089/api/messages
> {"name":"John","message":"Hello from client!"}

# 送信: サーバー通知を受信
wscat -c ws://localhost:8089/api/notifications
# 定期的なサーバーハートビートメッセージが表示されます

# 送信: データベース変更を受信
wscat -c ws://localhost:8089/api/database-changes
# データベース変更の通知が表示されます

# 双方向: 送受信するチャット
wscat -c ws://localhost:8089/api/chat
> {"username":"CLIUser","message":"Hello chat!"}
# 送信したメッセージのエコーと歓迎メッセージが表示されます

WebSocket クライアントでテストする

Node.js クライアントでも各フローをテストできます。open で JSON.stringify したメッセージを送信し、message イベントで JSON.parse(data.toString()) を使用します。

test-client.js
import WebSocket from 'ws';

// 📥 受信: サーバーへメッセージを送信
const messageWs = new WebSocket('ws://localhost:8089/api/messages');
messageWs.on('open', () => {
  console.log('📥 メッセージ送信用に接続しました');
  messageWs.send(JSON.stringify({name: 'NodeJS Client', message: 'Hello from Node.js!'}));
});

// 📤 送信: 通知を受信
const notificationWs = new WebSocket('ws://localhost:8089/api/notifications');
notificationWs.on('open', () => console.log('📤 通知受信用に接続しました'));
notificationWs.on('message', data => console.log('📤 受信した通知:', JSON.parse(data.toString())));

// 📤 送信: データベース変更を受信
const dbWs = new WebSocket('ws://localhost:8089/api/database-changes');
dbWs.on('open', () => console.log('📤 データベース変更受信用に接続しました'));
dbWs.on('message', data => console.log('📤 データベース変更:', JSON.parse(data.toString())));

// 🔄 双方向: チャットシステム
const chatWs = new WebSocket('ws://localhost:8089/api/chat');
chatWs.on('open', () => {
  console.log('🔄 チャットに接続しました');
  chatWs.send(JSON.stringify({username: 'NodeJS User', message: 'Hello chat!'}));
});
chatWs.on('message', data => console.log('🔄 チャットメッセージ:', JSON.parse(data.toString())));

次のステップ

WebSocket 接続を保護する認証バリデーターを追加する。
接続制限でレート制限を実装する。
メッセージをデータベースへ永続化する。
特定のクライアントグループへブロードキャストする。
本番メッセージングでは { webSocketServer } を指定した chatService を使用する。

🏗️ サービスアーキテクチャ​

1. WebSocket 機能を作成する​

2. サービスを作成する​

3. サーバーをセットアップする​

4. 環境設定​

サービスをテストする​

コマンドラインでテストする​

WebSocket クライアントでテストする​

次のステップ​

関連ドキュメント​