Websocketバックエンド実装

docker-compose.local.yml

@@ -10,6 +10,7 @@ services:
       - .env.container
     ports:
       - '52600:52600' # backend
+      - '52601:52601' # backend websocket
       - '52800:52800' # frontend
       - '5555:5555' # Prisma Studio
     depends_on:

packages/backend/package.json

@@ -13,7 +13,8 @@
     "migrate:exe": "prisma migrate dev",
     "migrate:status": "prisma migrate status",
     "migrate:reset": "prisma migrate reset",
-    "db:studio": "prisma studio"
+    "db:studio": "prisma studio",
+    "prepare": "if [ \"$NODE_ENV\" != \"production\" ]; then ts-patch install; fi"
   },
   "author": "",
   "license": "ISC",
@@ -25,18 +26,23 @@
     "express": "^5.1.0",
     "prisma": "^6.9.0",
     "typescript": "^5.8.3",
-    "winston": "^3.17.0"
+    "typia": "^9.7.1",
+    "winston": "^3.17.0",
+    "ws": "^8.18.3"
   },
   "devDependencies": {
     "@mermaid-js/mermaid-cli": "^11.4.2",
     "@types/cors": "^2.8.19",
     "@types/express": "^5.0.3",
     "@types/node": "^22.15.30",
     "@types/winston": "^2.4.4",
+    "@types/ws": "^8.18.1",
     "prisma-erd-generator": "^2.0.4",
     "puppeteer": "^23.11.1",
+    "ts-patch": "^3.3.0",
     "tsc-alias": "^1.8.16",
     "tsc-watch": "^7.1.1",
+    "typescript": "^5.9.2",
     "vitest": "^1.6.1"
   }
 }

packages/backend/src/index.ts

@@ -4,6 +4,10 @@ import fs from 'fs';
 import path from 'path';
 import { errorHandler } from './middlewares/errorHandler';
 import { logger } from '@/utils/logger';
+import { WebSocketDonChannel } from './websocket/donChannel';
+import { WebSocketDon } from './websocket/messages';
+import { WebSocketServer } from 'ws';
+import { getActiveDons } from './usecases/getActiveDonsUsecase';
 
 const app = express();
 
@@ -66,6 +70,18 @@ const server = app.listen(port, host, () => {
   });
 });
 
+// WebSocketサーバー
+const wss = new WebSocketServer({ port: 52601 });
+const donChannel = new WebSocketDonChannel({ wss, getActiveDons });
+logger.info('WebSocket server started on ws://localhost:52601');
+
+// 5秒ごとに通知をテスト送信
+setInterval(async () => {
+  logger.info('Sending notification to active clients...');
+  const activeDons = await getActiveDons();
+  donChannel.notifyActiveDonState(activeDons);
+}, 5000);
+
 // Graceful shutdown
 process.on('SIGTERM', () => {
   logger.info('SIGTERM signal received: closing HTTP server');

packages/backend/src/utils/errorMessage.ts

@@ -0,0 +1,7 @@
+import typia from 'typia';
+
+export const typiaValidationErrorMessage = (errors: typia.IValidation.IError[]): string => {
+  return `Validation errors found:\n${errors
+    .map((e) => `On ${e.path} Expected ${e.expected} Received ${e.value}`)
+    .join('\n')}`;
+};

packages/backend/src/utils/logger.ts

@@ -1,6 +1,7 @@
 import winston from 'winston';
 
 const isDevelopment = process.env.NODE_ENV !== 'production';
+const isTest = process.env.NODE_ENV === 'test';
 
 const logFormat = winston.format.combine(
   winston.format.timestamp({
@@ -16,13 +17,14 @@ const devFormat = winston.format.combine(
   }),
   winston.format.errors({ stack: true }),
   winston.format.colorize(),
-  winston.format.printf(({ timestamp, level, message, stack }) => {
-    return `${timestamp} [${level}]: ${message}${stack ? `\n${stack}` : ''}`;
+  winston.format.printf(({ timestamp, level, message, stack, ...metadata }) => {
+    const metaString = Object.keys(metadata).length > 0 ? `: ${JSON.stringify(metadata)}` : '';
+    return `${timestamp} [${level}]: ${message}${metaString}${stack ? `\n${stack}` : ''}`;
   }),
 );
 
 const logger = winston.createLogger({
-  level: isDevelopment ? 'debug' : 'info',
+  level: isTest ? 'error' : isDevelopment ? 'debug' : 'info',
   format: isDevelopment ? devFormat : logFormat,
   transports: [
     new winston.transports.Console(),

packages/backend/src/websocket/__tests__/donChannel.test.ts

@@ -0,0 +1,219 @@
+import { vi, describe, it, expect, beforeEach } from 'vitest';
+import { WebSocketDonChannel } from '../donChannel';
+import { WebSocket, WebSocketServer } from 'ws';
+import { Don } from '../messages';
+
+vi.mock('ws');
+vi.mock('typia', () => ({
+  default: {
+    json: {
+      validateParse: vi.fn(() => ({
+        success: true,
+        data: { type: 'request', data: { include: true } },
+      })),
+    },
+  },
+}));
+
+describe('WebSocketDonChannel', () => {
+  let mockWss: WebSocketServer;
+  let donChannel: WebSocketDonChannel;
+
+  // ヘルパー関数：mockWebSocketを作成
+  const createMockWebSocket = () =>
+    ({
+      on: vi.fn(),
+      send: vi.fn(),
+      readyState: WebSocket.OPEN,
+    }) as any;
+
+  // ヘルパー関数：connectionハンドラーを取得
+  const getConnectionHandler = () => {
+    const calls = (mockWss.on as any).mock.calls;
+    const connectionCall = calls.find((call: any[]) => call[0] === 'connection');
+    return connectionCall[1];
+  };
+
+  beforeEach(() => {
+    mockWss = {
+      on: vi.fn(),
+      clients: new Set(),
+    } as any;
+
+    donChannel = new WebSocketDonChannel({ wss: mockWss });
+  });
+
+  it('WebSocketDonChannelが初期化される', () => {
+    expect(donChannel).toBeDefined();
+    expect(mockWss.on).toHaveBeenCalledWith('connection', expect.any(Function));
+    expect(mockWss.on).toHaveBeenCalledWith('error', expect.any(Function));
+  });
+
+  describe('メッセージ処理', () => {
+    it('正常なRequestNotificationMessageを受信して通知状態が有効になる', () => {
+      const mockWs = createMockWebSocket();
+      const connectionHandler = getConnectionHandler();
+
+      // WebSocket接続をシミュレート
+      connectionHandler(mockWs);
+
+      // messageハンドラーを取得
+      const messageHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+
+      // 正常なメッセージを送信
+      const validMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      messageHandler(validMessage);
+
+      // 通知が有効になっていることを確認（Don状態通知でテスト）
+      mockWss.clients = new Set([mockWs]);
+      donChannel.notifyActiveDonState([{ id: '1', state: 'ordered' }]);
+
+      expect(mockWs.send).toHaveBeenCalled();
+    });
+  });
+
+  describe('Don状態通知', () => {
+    it('通知購読中のクライアントにのみメッセージが送信される', () => {
+      const subscribedWs = createMockWebSocket();
+      const nonSubscribedWs = createMockWebSocket();
+
+      // 購読済みクライアントをmockWss.clientsに追加
+      mockWss.clients = new Set([subscribedWs, nonSubscribedWs]);
+
+      // 購読状態を設定（実際のコードでは通知リクエスト受信時に設定される）
+      const connectionHandler = getConnectionHandler();
+      connectionHandler(subscribedWs);
+
+      const messageHandler = subscribedWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+
+      // 購読リクエストを送信（これにより通知状態が有効になる）
+      const requestMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      messageHandler(requestMessage);
+
+      // Don状態を通知
+      const donState: Don[] = [
+        { id: '1', state: 'ordered' },
+        { id: '2', state: 'cooking' },
+      ];
+      donChannel.notifyActiveDonState(donState);
+
+      // 購読済みクライアントにのみ送信されることを確認
+      expect(subscribedWs.send).toHaveBeenCalledWith(
+        JSON.stringify({
+          type: 'state',
+          data: { dons: donState },
+        }),
+      );
+      expect(nonSubscribedWs.send).not.toHaveBeenCalled();
+    });
+
+    it('OPEN状態のWebSocketにのみメッセージが送信される', () => {
+      const openWs = createMockWebSocket();
+      const closedWs = { ...createMockWebSocket(), readyState: WebSocket.CLOSED };
+
+      mockWss.clients = new Set([openWs, closedWs]);
+
+      // 両方のクライアントを購読状態にする
+      const connectionHandler = getConnectionHandler();
+      connectionHandler(openWs);
+      connectionHandler(closedWs);
+
+      const openMessageHandler = openWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+      const closedMessageHandler = closedWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+
+      const requestMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      openMessageHandler(requestMessage);
+      closedMessageHandler(requestMessage);
+
+      const donState: Don[] = [{ id: '1', state: 'ordered' }];
+      donChannel.notifyActiveDonState(donState);
+
+      expect(openWs.send).toHaveBeenCalled();
+      expect(closedWs.send).not.toHaveBeenCalled();
+    });
+
+    it('送信されるメッセージのフォーマットが正しい', () => {
+      const mockWs = createMockWebSocket();
+      mockWss.clients = new Set([mockWs]);
+
+      const connectionHandler = getConnectionHandler();
+      connectionHandler(mockWs);
+
+      const messageHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+
+      const requestMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      messageHandler(requestMessage);
+
+      const donState: Don[] = [{ id: 'test-id', state: 'cooking' }];
+      donChannel.notifyActiveDonState(donState);
+
+      const expectedMessage = JSON.stringify({
+        type: 'state',
+        data: { dons: donState },
+      });
+
+      expect(mockWs.send).toHaveBeenCalledWith(expectedMessage);
+    });
+  });
+
+  describe('WebSocket切断・エラー処理', () => {
+    it('closeイベント時に通知状態がクリアされる', () => {
+      const mockWs = createMockWebSocket();
+      const connectionHandler = getConnectionHandler();
+
+      connectionHandler(mockWs);
+
+      // closeハンドラーを取得
+      const closeHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'close')[1];
+
+      // 通知状態を設定
+      const messageHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+      const requestMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      messageHandler(requestMessage);
+
+      // 通知が有効であることを確認
+      mockWs.send.mockClear();
+      mockWss.clients = new Set([mockWs]);
+      donChannel.notifyActiveDonState([{ id: '1', state: 'ordered' }]);
+      expect(mockWs.send).toHaveBeenCalled();
+
+      // closeイベントを発火
+      closeHandler();
+
+      // 通知状態がクリアされ、メッセージが送信されないことを確認
+      mockWs.send.mockClear();
+      donChannel.notifyActiveDonState([{ id: '1', state: 'ordered' }]);
+      expect(mockWs.send).not.toHaveBeenCalled();
+    });
+
+    it('errorイベント時に通知状態がクリアされる', () => {
+      const mockWs = createMockWebSocket();
+      const connectionHandler = getConnectionHandler();
+
+      connectionHandler(mockWs);
+
+      // errorハンドラーを取得
+      const errorHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'error')[1];
+
+      // 通知状態を設定
+      const messageHandler = mockWs.on.mock.calls.find((call: any[]) => call[0] === 'message')[1];
+      const requestMessage = JSON.stringify({ type: 'request', data: { include: true } });
+      messageHandler(requestMessage);
+
+      // 通知が有効であることを確認
+      mockWs.send.mockClear();
+      mockWss.clients = new Set([mockWs]);
+      donChannel.notifyActiveDonState([{ id: '1', state: 'ordered' }]);
+      expect(mockWs.send).toHaveBeenCalled();
+
+      // errorイベントを発火
+      const error = new Error('Connection error');
+      errorHandler(error);
+
+      // 通知状態がクリアされ、メッセージが送信されないことを確認
+      mockWs.send.mockClear();
+      donChannel.notifyActiveDonState([{ id: '1', state: 'ordered' }]);
+      expect(mockWs.send).not.toHaveBeenCalled();
+    });
+  });
+});