【2026年版】WebSocket＋AIリアルタイムアプリケーション開発ガイド：チャット・通知・コラボレーション

// server.js - WebSocket サーバー基本実装
import { WebSocketServer, WebSocket } from 'ws';
import { createServer } from 'http';

const server = createServer();
const wss = new WebSocketServer({ server });

// 接続管理
const clients = new Map();

wss.on('connection', (ws, req) => {
  const clientId = crypto.randomUUID();
  clients.set(clientId, { ws, connectedAt: new Date() });

  console.log(`Client connected: ${clientId}`);

  // メッセージ受信
  ws.on('message', (data) => {
    try {
      const message = JSON.parse(data.toString());
      handleMessage(clientId, message);
    } catch (error) {
      ws.send(JSON.stringify({ type: 'error', message: 'Invalid JSON' }));
    }
  });

  // 切断処理
  ws.on('close', (code, reason) => {
    clients.delete(clientId);
    console.log(`Client disconnected: ${clientId} (code: ${code})`);
  });

  // Ping/Pong でヘルスチェック
  ws.isAlive = true;
  ws.on('pong', () => { ws.isAlive = true; });

  // 接続確認メッセージ
  ws.send(JSON.stringify({
    type: 'connected',
    clientId,
    timestamp: new Date().toISOString()
  }));
});

// ヘルスチェック（30秒間隔）
const heartbeat = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (!ws.isAlive) return ws.terminate();
    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

function handleMessage(clientId, message) {
  switch (message.type) {
    case 'chat':
      broadcast(clientId, message);
      break;
    case 'typing':
      broadcastExcept(clientId, { type: 'typing', user: message.user });
      break;
    default:
      console.log('Unknown message type:', message.type);
  }
}

function broadcast(senderId, message) {
  const payload = JSON.stringify({
    ...message,
    senderId,
    timestamp: new Date().toISOString()
  });
  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(payload);
    }
  });
}

function broadcastExcept(excludeId, message) {
  const payload = JSON.stringify(message);
  clients.forEach((client, id) => {
    if (id !== excludeId && client.ws.readyState === WebSocket.OPEN) {
      client.ws.send(payload);
    }
  });
}

server.listen(8080, () => {
  console.log('WebSocket server running on port 8080');
});

// client.js - WebSocket クライアント（再接続機能付き）
class WebSocketClient {
  constructor(url, options = {}) {
    this.url = url;
    this.options = {
      reconnectInterval: 1000,
      maxReconnectInterval: 30000,
      reconnectDecay: 1.5,
      maxReconnectAttempts: 20,
      ...options
    };
    this.reconnectAttempts = 0;
    this.handlers = new Map();
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      console.log('WebSocket connected');
      this.reconnectAttempts = 0;
      this.emit('connected');
    };

    this.ws.onmessage = (event) => {
      try {
        const message = JSON.parse(event.data);
        this.emit(message.type, message);
      } catch (error) {
        console.error('Failed to parse message:', error);
      }
    };

    this.ws.onclose = (event) => {
      console.log(`WebSocket closed: ${event.code}`);
      this.emit('disconnected', event);
      if (!event.wasClean) {
        this.reconnect();
      }
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };
  }

  reconnect() {
    if (this.reconnectAttempts >= this.options.maxReconnectAttempts) {
      console.error('Max reconnect attempts reached');
      this.emit('reconnect_failed');
      return;
    }

    const delay = Math.min(
      this.options.reconnectInterval * Math.pow(
        this.options.reconnectDecay, this.reconnectAttempts
      ),
      this.options.maxReconnectInterval
    );

    this.reconnectAttempts++;
    console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
    setTimeout(() => this.connect(), delay);
  }

  send(type, data = {}) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({ type, ...data }));
    }
  }

  on(event, handler) {
    if (!this.handlers.has(event)) {
      this.handlers.set(event, []);
    }
    this.handlers.get(event).push(handler);
  }

  emit(event, data) {
    const handlers = this.handlers.get(event) || [];
    handlers.forEach(handler => handler(data));
  }
}

// socket-io-server.js
import { Server } from 'socket.io';
import { createServer } from 'http';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

const httpServer = createServer();
const io = new Server(httpServer, {
  cors: {
    origin: ['https://example.com'],
    methods: ['GET', 'POST']
  },
  connectionStateRecovery: {
    maxDisconnectionDuration: 2 * 60 * 1000,
    skipMiddlewares: true
  }
});

// Redis アダプター（スケーリング用）
const pubClient = createClient({ url: 'redis://localhost:6379' });
const subClient = pubClient.duplicate();
await Promise.all([pubClient.connect(), subClient.connect()]);
io.adapter(createAdapter(pubClient, subClient));

// 認証ミドルウェア
io.use(async (socket, next) => {
  const token = socket.handshake.auth.token;
  try {
    const user = await verifyToken(token);
    socket.user = user;
    next();
  } catch (error) {
    next(new Error('Authentication failed'));
  }
});

// 名前空間: チャット
const chatNamespace = io.of('/chat');

chatNamespace.on('connection', (socket) => {
  console.log(`User connected: ${socket.user.name}`);

  // ルーム参加
  socket.on('join_room', async (roomId) => {
    socket.join(roomId);
    socket.to(roomId).emit('user_joined', {
      user: socket.user.name,
      timestamp: new Date()
    });
  });

  // メッセージ送信
  socket.on('send_message', async (data) => {
    const message = {
      id: crypto.randomUUID(),
      user: socket.user.name,
      content: data.content,
      roomId: data.roomId,
      timestamp: new Date()
    };

    // DBに保存
    await saveMessage(message);

    // ルーム内にブロードキャスト
    chatNamespace.to(data.roomId).emit('new_message', message);
  });

  // タイピングインジケーター
  socket.on('typing_start', (roomId) => {
    socket.to(roomId).emit('user_typing', {
      user: socket.user.name
    });
  });

  socket.on('typing_stop', (roomId) => {
    socket.to(roomId).emit('user_stopped_typing', {
      user: socket.user.name
    });
  });

  socket.on('disconnect', () => {
    console.log(`User disconnected: ${socket.user.name}`);
  });
});

httpServer.listen(3000);

// ai-streaming-server.js - LLM応答のストリーミング
import express from 'express';
import OpenAI from 'openai';

const app = express();
const openai = new OpenAI();

app.get('/api/ai/stream', async (req, res) => {
  const { prompt } = req.query;

  // SSE ヘッダー設定
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no');

  try {
    const stream = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages: [{ role: 'user', content: prompt }],
      stream: true
    });

    for await (const chunk of stream) {
      const content = chunk.choices[0]?.delta?.content;
      if (content) {
        res.write(`data: ${JSON.stringify({ content })}\n\n`);
      }
    }

    res.write(`data: ${JSON.stringify({ done: true })}\n\n`);
    res.end();
  } catch (error) {
    res.write(`data: ${JSON.stringify({ error: error.message })}\n\n`);
    res.end();
  }
});

app.listen(3001);

// ai-chat-websocket.js - WebSocket経由のAIチャット
import { WebSocketServer } from 'ws';
import OpenAI from 'openai';

const wss = new WebSocketServer({ port: 8080 });
const openai = new OpenAI();

// セッション管理
const sessions = new Map();

wss.on('connection', (ws) => {
  const sessionId = crypto.randomUUID();
  sessions.set(sessionId, {
    messages: [
      { role: 'system', content: 'あなたは親切で丁寧な日本語AIアシスタントです。' }
    ]
  });

  ws.on('message', async (data) => {
    const { content } = JSON.parse(data.toString());
    const session = sessions.get(sessionId);

    // ユーザーメッセージを履歴に追加
    session.messages.push({ role: 'user', content });

    try {
      // ストリーミング応答
      const stream = await openai.chat.completions.create({
        model: 'gpt-4o',
        messages: session.messages,
        stream: true
      });

      let fullResponse = '';

      for await (const chunk of stream) {
        const token = chunk.choices[0]?.delta?.content;
        if (token) {
          fullResponse += token;
          ws.send(JSON.stringify({
            type: 'ai_stream',
            token,
            sessionId
          }));
        }
      }

      // 完了通知
      ws.send(JSON.stringify({ type: 'ai_complete', sessionId }));

      // AI応答を履歴に追加
      session.messages.push({ role: 'assistant', content: fullResponse });

    } catch (error) {
      ws.send(JSON.stringify({
        type: 'error',
        message: 'AI応答の生成に失敗しました'
      }));
    }
  });

  ws.on('close', () => {
    sessions.delete(sessionId);
  });
});

// collaboration-server.js - Yjs + WebSocket による共同編集
import { WebSocketServer } from 'ws';
import { setupWSConnection } from 'y-websocket/bin/utils';

const wss = new WebSocketServer({ port: 1234 });

wss.on('connection', (ws, req) => {
  // Yjs WebSocket接続のセットアップ
  setupWSConnection(ws, req, {
    docName: req.url.slice(1),  // URLパスからドキュメント名を取得
    gc: true                     // ガベージコレクション有効
  });
});

console.log('Collaboration server running on port 1234');

// collaboration-client.js - フロントエンド側
import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';
import { QuillBinding } from 'y-quill';
import Quill from 'quill';

// Yjs ドキュメント作成
const ydoc = new Y.Doc();

// WebSocket プロバイダー接続
const provider = new WebsocketProvider(
  'wss://collaboration.example.com',
  'document-room-123',
  ydoc
);

// 接続状態の監視
provider.on('status', (event) => {
  console.log('Connection status:', event.status);
});

// 共有テキストの取得
const ytext = ydoc.getText('quill');

// Quillエディタとバインド
const editor = new Quill('#editor', { theme: 'snow' });
new QuillBinding(ytext, editor, provider.awareness);

// アウェアネス（他のユーザーのカーソル位置等）
provider.awareness.setLocalStateField('user', {
  name: 'ユーザーA',
  color: '#ff6b6b'
});

# websocket-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: websocket-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: websocket-server
  template:
    metadata:
      labels:
        app: websocket-server
    spec:
      containers:
      - name: websocket
        image: my-websocket-server:latest
        ports:
        - containerPort: 8080
        env:
        - name: REDIS_URL
          value: "redis://redis-cluster:6379"
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "1000m"
            memory: "1Gi"
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: websocket-service
spec:
  type: ClusterIP
  selector:
    app: websocket-server
  ports:
  - port: 8080
    targetPort: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: websocket-ingress
  annotations:
    nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
    nginx.ingress.kubernetes.io/upstream-hash-by: "$remote_addr"
spec:
  rules:
  - host: ws.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: websocket-service
            port:
              number: 8080

// セキュリティミドルウェアの例
function securityMiddleware(ws, req) {
  // Origin検証
  const allowedOrigins = ['https://example.com', 'https://app.example.com'];
  const origin = req.headers.origin;
  if (!allowedOrigins.includes(origin)) {
    ws.close(4003, 'Forbidden origin');
    return false;
  }

  // レート制限
  const rateLimiter = new Map();
  const clientIP = req.socket.remoteAddress;
  const now = Date.now();
  const windowMs = 60000;  // 1分間
  const maxMessages = 100; // 最大100メッセージ/分

  if (!rateLimiter.has(clientIP)) {
    rateLimiter.set(clientIP, []);
  }

  const timestamps = rateLimiter.get(clientIP)
    .filter(t => t > now - windowMs);
  timestamps.push(now);
  rateLimiter.set(clientIP, timestamps);

  if (timestamps.length > maxMessages) {
    ws.close(4029, 'Rate limit exceeded');
    return false;
  }

  return true;
}