← back to Designer Wallcoverings

DW-Agents/dw-agents/global-chat-system/chat-server.ts

452 lines

/**
 * DW-Agents Global Chat Server
 * Provides real-time inter-agent communication via WebSocket
 */

import express, { Request, Response } from 'express';
import { WebSocketServer, WebSocket } from 'ws';
import { v4 as uuidv4 } from 'uuid';
import * as fs from 'fs';
import * as path from 'path';
import cors from 'cors';
import {
  Agent,
  ChatMessage,
  ChatChannel,
  ChatConnection,
  SystemNotification
} from './types';

const app = express();
const PORT = parseInt(process.env.CHAT_PORT || '9999', 10);
const DATA_DIR = path.join(__dirname, 'data');

// Middleware
app.use(cors());
app.use(express.json());
app.use(express.static(path.join(__dirname, 'public')));

// Ensure data directory exists
if (!fs.existsSync(DATA_DIR)) {
  fs.mkdirSync(DATA_DIR, { recursive: true });
}

// In-memory stores
const connections = new Map<string, ChatConnection>();
const agents = new Map<string, Agent>();
const channels = new Map<string, ChatChannel>();
const messageHistory: ChatMessage[] = [];
const systemNotifications: SystemNotification[] = [];

// Load persisted data
function loadData() {
  try {
    const agentsFile = path.join(DATA_DIR, 'agents.json');
    if (fs.existsSync(agentsFile)) {
      const data = JSON.parse(fs.readFileSync(agentsFile, 'utf-8'));
      data.forEach((agent: Agent) => agents.set(agent.id, agent));
      console.log(`📋 Loaded ${agents.size} registered agents`);
    }

    const messagesFile = path.join(DATA_DIR, 'messages.json');
    if (fs.existsSync(messagesFile)) {
      const data = JSON.parse(fs.readFileSync(messagesFile, 'utf-8'));
      messageHistory.push(...data);
      console.log(`💬 Loaded ${messageHistory.length} message history`);
    }

    const channelsFile = path.join(DATA_DIR, 'channels.json');
    if (fs.existsSync(channelsFile)) {
      const data = JSON.parse(fs.readFileSync(channelsFile, 'utf-8'));
      data.forEach((channel: ChatChannel) => channels.set(channel.id, channel));
      console.log(`📢 Loaded ${channels.size} channels`);
    }
  } catch (error) {
    console.error('⚠️  Error loading data:', error);
  }
}

// Save data periodically
function saveData() {
  try {
    fs.writeFileSync(
      path.join(DATA_DIR, 'agents.json'),
      JSON.stringify(Array.from(agents.values()), null, 2)
    );

    // Keep only last 1000 messages
    const recentMessages = messageHistory.slice(-1000);
    fs.writeFileSync(
      path.join(DATA_DIR, 'messages.json'),
      JSON.stringify(recentMessages, null, 2)
    );

    fs.writeFileSync(
      path.join(DATA_DIR, 'channels.json'),
      JSON.stringify(Array.from(channels.values()), null, 2)
    );
  } catch (error) {
    console.error('⚠️  Error saving data:', error);
  }
}

// Create default channels
function initializeChannels() {
  if (channels.size === 0) {
    const defaultChannels: ChatChannel[] = [
      {
        id: 'general',
        name: 'General',
        description: 'General discussion for all agents',
        members: [],
        createdAt: new Date(),
        createdBy: 'system'
      },
      {
        id: 'alerts',
        name: 'Alerts',
        description: 'System alerts and critical notifications',
        members: [],
        createdAt: new Date(),
        createdBy: 'system'
      },
      {
        id: 'errors',
        name: 'Errors',
        description: 'Error reporting and debugging',
        members: [],
        createdAt: new Date(),
        createdBy: 'system'
      },
      {
        id: 'tasks',
        name: 'Tasks',
        description: 'Task coordination and updates',
        members: [],
        createdAt: new Date(),
        createdBy: 'system'
      }
    ];

    defaultChannels.forEach(channel => channels.set(channel.id, channel));
    console.log('📢 Created default channels');
  }
}

// WebSocket Server
const wss = new WebSocketServer({ noServer: true });

wss.on('connection', (ws: WebSocket, req: Request) => {
  const url = new URL(req.url || '', `http://${req.headers.host}`);
  const agentId = url.searchParams.get('agentId') || '';
  const agentName = url.searchParams.get('agentName') || 'Unknown';
  const userId = url.searchParams.get('userId') || undefined;
  const userName = url.searchParams.get('userName') || undefined;

  if (!agentId) {
    ws.close(1008, 'Agent ID required');
    return;
  }

  // Create connection
  const connection: ChatConnection = {
    agentId,
    agentName,
    ws,
    connectedAt: new Date(),
    userId,
    userName
  };

  connections.set(agentId, connection);

  // Update agent status
  if (!agents.has(agentId)) {
    agents.set(agentId, {
      id: agentId,
      name: agentName,
      port: 0, // Will be updated by agent
      category: 'unknown',
      online: true,
      lastSeen: new Date()
    });
  } else {
    const agent = agents.get(agentId)!;
    agent.online = true;
    agent.lastSeen = new Date();
  }

  // Broadcast agent online notification
  const notification: SystemNotification = {
    id: uuidv4(),
    timestamp: new Date(),
    type: 'agent_online',
    agentId,
    message: `${agentName} is now online`,
    severity: 'info'
  };
  systemNotifications.push(notification);
  broadcastSystemNotification(notification);

  console.log(`✅ ${agentName} (${agentId}) connected${userId ? ` - User: ${userName}` : ''}`);

  // Send connection confirmation
  ws.send(JSON.stringify({
    type: 'connected',
    agentId,
    timestamp: new Date(),
    onlineAgents: Array.from(connections.keys())
  }));

  // Handle incoming messages
  ws.on('message', (data: Buffer) => {
    try {
      const payload = JSON.parse(data.toString());
      handleMessage(agentId, agentName, payload, userId, userName);
    } catch (error) {
      console.error('❌ Error parsing message:', error);
    }
  });

  // Handle disconnect
  ws.on('close', () => {
    connections.delete(agentId);

    const agent = agents.get(agentId);
    if (agent) {
      agent.online = false;
      agent.lastSeen = new Date();
    }

    const offlineNotification: SystemNotification = {
      id: uuidv4(),
      timestamp: new Date(),
      type: 'agent_offline',
      agentId,
      message: `${agentName} went offline`,
      severity: 'info'
    };
    systemNotifications.push(offlineNotification);
    broadcastSystemNotification(offlineNotification);

    console.log(`❌ ${agentName} (${agentId}) disconnected`);
  });

  ws.on('error', (error) => {
    console.error(`❌ WebSocket error for ${agentName}:`, error);
  });
});

// Handle incoming messages
function handleMessage(
  fromAgentId: string,
  fromAgentName: string,
  payload: any,
  userId?: string,
  userName?: string
) {
  const { type, to, channel, content, priority, metadata } = payload;

  const message: ChatMessage = {
    id: uuidv4(),
    timestamp: new Date(),
    from: {
      agentId: fromAgentId,
      agentName: fromAgentName,
      userId,
      userName
    },
    content,
    type: type || 'broadcast',
    priority: priority || 'normal',
    metadata
  };

  // Direct message
  if (type === 'direct' && to) {
    message.to = {
      agentId: to.agentId,
      agentName: to.agentName
    };
    sendDirectMessage(message);
  }
  // Channel message
  else if (channel) {
    message.channel = channel;
    broadcastToChannel(channel, message);
  }
  // Broadcast to all
  else {
    broadcastMessage(message);
  }

  // Store message
  messageHistory.push(message);

  // Auto-save every 10 messages
  if (messageHistory.length % 10 === 0) {
    saveData();
  }
}

// Send direct message
function sendDirectMessage(message: ChatMessage) {
  const toAgentId = message.to?.agentId;
  if (!toAgentId) return;

  const connection = connections.get(toAgentId);
  if (connection && connection.ws.readyState === WebSocket.OPEN) {
    connection.ws.send(JSON.stringify({
      type: 'direct_message',
      message
    }));
  }

  // Also send to sender for confirmation
  const senderConnection = connections.get(message.from.agentId);
  if (senderConnection && senderConnection.ws.readyState === WebSocket.OPEN) {
    senderConnection.ws.send(JSON.stringify({
      type: 'message_sent',
      message
    }));
  }
}

// Broadcast to all connected agents
function broadcastMessage(message: ChatMessage) {
  connections.forEach((connection) => {
    if (connection.ws.readyState === WebSocket.OPEN) {
      connection.ws.send(JSON.stringify({
        type: 'broadcast_message',
        message
      }));
    }
  });
}

// Broadcast to channel members
function broadcastToChannel(channelId: string, message: ChatMessage) {
  const channel = channels.get(channelId);
  if (!channel) return;

  connections.forEach((connection) => {
    if (connection.ws.readyState === WebSocket.OPEN) {
      connection.ws.send(JSON.stringify({
        type: 'channel_message',
        channel: channelId,
        message
      }));
    }
  });
}

// Broadcast system notification
function broadcastSystemNotification(notification: SystemNotification) {
  connections.forEach((connection) => {
    if (connection.ws.readyState === WebSocket.OPEN) {
      connection.ws.send(JSON.stringify({
        type: 'system_notification',
        notification
      }));
    }
  });
}

// REST API Endpoints

// Get online agents
app.get('/api/agents/online', (req: Request, res: Response) => {
  const onlineAgents = Array.from(agents.values()).filter(a => a.online);
  res.json(onlineAgents);
});

// Get all agents
app.get('/api/agents', (req: Request, res: Response) => {
  res.json(Array.from(agents.values()));
});

// Get channels
app.get('/api/channels', (req: Request, res: Response) => {
  res.json(Array.from(channels.values()));
});

// Get message history
app.get('/api/messages', (req: Request, res: Response) => {
  const { agentId, channel, limit = 100 } = req.query;

  let filtered = messageHistory;

  if (agentId) {
    filtered = filtered.filter(
      m => m.from.agentId === agentId || m.to?.agentId === agentId
    );
  }

  if (channel) {
    filtered = filtered.filter(m => m.channel === channel);
  }

  const limited = filtered.slice(-(limit as number));
  res.json(limited);
});

// Get system notifications
app.get('/api/notifications', (req: Request, res: Response) => {
  const limit = parseInt(req.query.limit as string) || 50;
  res.json(systemNotifications.slice(-limit));
});

// Health check
app.get('/health', (req: Request, res: Response) => {
  res.json({
    status: 'ok',
    connections: connections.size,
    agents: agents.size,
    channels: channels.size,
    messages: messageHistory.length
  });
});

// Initialize
loadData();
initializeChannels();

// Start HTTP server
const server = app.listen(PORT, () => {
  console.log('\n💬 DW-Agents Global Chat Server');
  console.log('━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━');
  console.log(`🌍 External: http://45.61.58.125:${PORT}`);
  console.log(`🏠 Local: http://localhost:${PORT}`);
  console.log('\n📢 Available Channels:');
  channels.forEach(channel => {
    console.log(`   - ${channel.name}: ${channel.description}`);
  });
  console.log('\n✅ Chat server ready for connections...\n');
});

// Handle WebSocket upgrade
server.on('upgrade', (request, socket, head) => {
  wss.handleUpgrade(request, socket, head, (ws) => {
    wss.emit('connection', ws, request);
  });
});

// Auto-save every minute
setInterval(saveData, 60000);

// Cleanup old notifications every hour
setInterval(() => {
  const oneHourAgo = new Date(Date.now() - 3600000);
  const removed = systemNotifications.filter(n => n.timestamp > oneHourAgo);
  systemNotifications.length = 0;
  systemNotifications.push(...removed);
}, 3600000);

// Graceful shutdown
process.on('SIGINT', () => {
  console.log('\n🛑 Shutting down chat server...');
  saveData();
  server.close();
  process.exit(0);
});