【Node.js】高级主题

个人主页：Guiat
归属专栏：node.js

正文

1. Node.js 高级主题概览

Node.js 高级主题涵盖了深入的技术概念和实践，包括事件循环机制、内存管理、性能优化、微服务架构、实时通信等核心领域。掌握这些高级主题对于构建高性能、可扩展的企业级应用至关重要。

1.1 高级主题架构图

2. 事件循环与异步编程深度解析

2.1 事件循环机制详解

事件循环阶段详解

// event-loop-demo.js
const fs = require('fs');

console.log('=== 事件循环演示 ===');

// 1. 同步代码
console.log('1. 同步代码执行');

// 2. process.nextTick (微任务，最高优先级)
process.nextTick(() => {
    console.log('2. process.nextTick 回调');
});

// 3. Promise (微任务)
Promise.resolve().then(() => {
    console.log('3. Promise.then 回调');
});

// 4. setImmediate (Check 阶段)
setImmediate(() => {
    console.log('4. setImmediate 回调');
});

// 5. setTimeout (Timer 阶段)
setTimeout(() => {
    console.log('5. setTimeout 回调');
}, 0);

// 6. I/O 操作 (Poll 阶段)
fs.readFile(__filename, () => {
    console.log('6. fs.readFile 回调');
    
    // 在 I/O 回调中的 setImmediate 会在下一次 setTimeout 之前执行
    setImmediate(() => {
        console.log('7. setImmediate 在 I/O 回调中');
    });
    
    setTimeout(() => {
        console.log('8. setTimeout 在 I/O 回调中');
    }, 0);
});

console.log('9. 同步代码结束');

/*
输出顺序：
1. 同步代码执行
9. 同步代码结束
2. process.nextTick 回调
3. Promise.then 回调
5. setTimeout 回调
4. setImmediate 回调
6. fs.readFile 回调
7. setImmediate 在 I/O 回调中
8. setTimeout 在 I/O 回调中
*/

2.2 异步编程模式演进

高级异步模式实现

// advanced-async-patterns.js

// 1. 异步迭代器
class AsyncDataProcessor {
    constructor(data) {
        this.data = data;
        this.index = 0;
    }

    async *[Symbol.asyncIterator]() {
        while (this.index < this.data.length) {
            // 模拟异步处理
            await new Promise(resolve => setTimeout(resolve, 100));
            yield this.data[this.index++];
        }
    }
}

// 使用异步迭代器
async function processDataAsync() {
    const processor = new AsyncDataProcessor([1, 2, 3, 4, 5]);
    
    for await (const item of processor) {
        console.log('处理项目:', item);
    }
}

// 2. 异步生成器与流控制
async function* asyncGenerator() {
    let i = 0;
    while (i < 10) {
        await new Promise(resolve => setTimeout(resolve, 500));
        yield i++;
    }
}

// 3. 高级 Promise 模式
class PromisePool {
    constructor(concurrency = 3) {
        this.concurrency = concurrency;
        this.running = [];
        this.queue = [];
    }

    async add(promiseFunction) {
        return new Promise((resolve, reject) => {
            this.queue.push({
                promiseFunction,
                resolve,
                reject
            });
            this.process();
        });
    }

    async process() {
        if (this.running.length >= this.concurrency || this.queue.length === 0) {
            return;
        }

        const { promiseFunction, resolve, reject } = this.queue.shift();
        const promise = promiseFunction()
            .then(resolve)
            .catch(reject)
            .finally(() => {
                this.running.splice(this.running.indexOf(promise), 1);
                this.process();
            });

        this.running.push(promise);
    }
}

// 4. 可取消的 Promise
class CancellablePromise {
    constructor(executor) {
        this.isCancelled = false;
        
        this.promise = new Promise((resolve, reject) => {
            this.cancel = () => {
                this.isCancelled = true;
                reject(new Error('Promise was cancelled'));
            };

            executor(
                (value) => {
                    if (!this.isCancelled) {
                        resolve(value);
                    }
                },
                (reason) => {
                    if (!this.isCancelled) {
                        reject(reason);
                    }
                }
            );
        });
    }

    then(onFulfilled, onRejected) {
        return this.promise.then(onFulfilled, onRejected);
    }

    catch(onRejected) {
        return this.promise.catch(onRejected);
    }
}

// 5. 异步重试机制
async function retryAsync(fn, maxRetries = 3, delay = 1000) {
    let lastError;
    
    for (let i = 0; i <= maxRetries; i++) {
        try {
            return await fn();
        } catch (error) {
            lastError = error;
            
            if (i === maxRetries) {
                throw lastError;
            }
            
            // 指数退避
            const waitTime = delay * Math.pow(2, i);
            await new Promise(resolve => setTimeout(resolve, waitTime));
        }
    }
}

module.exports = {
    AsyncDataProcessor,
    PromisePool,
    CancellablePromise,
    retryAsync,
    processDataAsync,
    asyncGenerator
};

3. 内存管理与性能优化

3.1 V8 内存管理机制

内存监控与分析工具

// memory-profiler.js
const v8 = require('v8');
const fs = require('fs');

class MemoryProfiler {
    constructor() {
        this.snapshots = [];
        this.startTime = Date.now();
    }

    // 获取内存使用情况
    getMemoryUsage() {
        const usage = process.memoryUsage();
        const heapStats = v8.getHeapStatistics();
        
        return {
            timestamp: Date.now(),
            rss: usage.rss,
            heapTotal: usage.heapTotal,
            heapUsed: usage.heapUsed,
            external: usage.external,
            arrayBuffers: usage.arrayBuffers,
            heapSizeLimit: heapStats.heap_size_limit,
            totalHeapSize: heapStats.total_heap_size,
            usedHeapSize: heapStats.used_heap_size,
            mallocedMemory: heapStats.malloced_memory,
            peakMallocedMemory: heapStats.peak_malloced_memory
        };
    }

    // 生成堆快照
    takeHeapSnapshot(filename) {
        const snapshotStream = v8.getHeapSnapshot();
        const fileStream = fs.createWriteStream(filename || `heap-${Date.now()}.heapsnapshot`);
        
        snapshotStream.pipe(fileStream);
        
        return new Promise((resolve, reject) => {
            fileStream.on('finish', () => {
                console.log(`堆快照已保存: ${filename}`);
                resolve(filename);
            });
            fileStream.on('error', reject);
        });
    }

    // 监控内存泄漏
    startMemoryLeak Detection() {
        const interval = setInterval(() => {
            const usage = this.getMemoryUsage();
            this.snapshots.push(usage);
            
            // 保留最近100个快照
            if (this.snapshots.length > 100) {
                this.snapshots.shift();
            }
            
            // 检测内存泄漏
            if (this.snapshots.length >= 10) {
                const recent = this.snapshots.slice(-10);
                const trend = this.calculateMemoryTrend(recent);
                
                if (trend.isIncreasing && trend.rate > 1024 * 1024) { // 1MB/snapshot
                    console.warn('检测到可能的内存泄漏:', {
                        trend: trend.rate,
                        currentUsage: usage.heapUsed
                    });
                }
            }
        }, 5000);

        return () => clearInterval(interval);
    }

    // 计算内存趋势
    calculateMemoryTrend(snapshots) {
        if (snapshots.length < 2) return { isIncreasing: false, rate: 0 };
        
        const first = snapshots[0].heapUsed;
        const last = snapshots[snapshots.length - 1].heapUsed;
        const rate = (last - first) / snapshots.length;
        
        return {
            isIncreasing: rate > 0,
            rate: rate
        };
    }

    // 强制垃圾回收 (需要 --expose-gc 标志)
    forceGC() {
        if (global.gc) {
            const before = this.getMemoryUsage();
            global.gc();
            const after = this.getMemoryUsage();
            
            console.log('垃圾回收效果:', {
                before: before.heapUsed,
                after: after.heapUsed,
                freed: before.heapUsed - after.heapUsed
            });
        } else {
            console.warn('垃圾回收不可用，请使用 --expose-gc 标志启动');
        }
    }

    // 生成内存报告
    generateReport() {
        const current = this.getMemoryUsage();
        const runtime = Date.now() - this.startTime;
        
        return {
            runtime: runtime,
            currentMemory: current,
            snapshots: this.snapshots.length,
            averageHeapUsed: this.snapshots.reduce((sum, s) => sum + s.heapUsed, 0) / this.snapshots.length
        };
    }
}

module.exports = MemoryProfiler;

3.2 性能优化策略

性能分析工具实现

// performance-analyzer.js
const { performance, PerformanceObserver } = require('perf_hooks');

class PerformanceAnalyzer {
    constructor() {
        this.metrics = new Map();
        this.observers = [];
        this.setupObservers();
    }

    // 设置性能观察器
    setupObservers() {
        // HTTP 请求性能观察
        const httpObserver = new PerformanceObserver((list) => {
            for (const entry of list.getEntries()) {
                this.recordMetric('http', {
                    name: entry.name,
                    duration: entry.duration,
                    startTime: entry.startTime
                });
            }
        });
        httpObserver.observe({ entryTypes: ['http'] });
        this.observers.push(httpObserver);

        // 函数性能观察
        const functionObserver = new PerformanceObserver((list) => {
            for (const entry of list.getEntries()) {
                this.recordMetric('function', {
                    name: entry.name,
                    duration: entry.duration,
                    startTime: entry.startTime
                });
            }
        });
        functionObserver.observe({ entryTypes: ['function', 'measure'] });
        this.observers.push(functionObserver);
    }

    // 记录指标
    recordMetric(type, data) {
        if (!this.metrics.has(type)) {
            this.metrics.set(type, []);
        }
        this.metrics.get(type).push({
            ...data,
            timestamp: Date.now()
        });
    }

    // 性能装饰器
    performanceDecorator(target, propertyKey, descriptor) {
        const originalMethod = descriptor.value;
        
        descriptor.value = async function(...args) {
            const start = performance.now();
            const markStart = `${propertyKey}-start`;
            const markEnd = `${propertyKey}-end`;
            const measureName = `${propertyKey}-duration`;
            
            performance.mark(markStart);
            
            try {
                const result = await originalMethod.apply(this, args);
                return result;
            } finally {
                performance.mark(markEnd);
                performance.measure(measureName, markStart, markEnd);
                
                const end = performance.now();
                console.log(`${propertyKey} 执行时间: ${(end - start).toFixed(2)}ms`);
            }
        };
        
        return descriptor;
    }

    // 函数性能测试
    async measureFunction(fn, iterations = 1000) {
        const results = [];
        
        for (let i = 0; i < iterations; i++) {
            const start = performance.now();
            await fn();
            const end = performance.now();
            results.push(end - start);
        }
        
        return {
            iterations,
            min: Math.min(...results),
            max: Math.max(...results),
            average: results.reduce((a, b) => a + b, 0) / results.length,
            median: results.sort((a, b) => a - b)[Math.floor(results.length / 2)],
            p95: results.sort((a, b) => a - b)[Math.floor(results.length * 0.95)],
            p99: results.sort((a, b) => a - b)[Math.floor(results.length * 0.99)]
        };
    }

    // CPU 使用率监控
    monitorCPU() {
        const startUsage = process.cpuUsage();
        const startTime = process.hrtime();
        
        return () => {
            const endUsage = process.cpuUsage(startUsage);
            const endTime = process.hrtime(startTime);
            
            const totalTime = endTime[0] * 1000000 + endTime[1] / 1000; // 微秒
            const cpuPercent = (endUsage.user + endUsage.system) / totalTime * 100;
            
            return {
                user: endUsage.user,
                system: endUsage.system,
                total: endUsage.user + endUsage.system,
                percentage: cpuPercent
            };
        };
    }

    // 生成性能报告
    generateReport() {
        const report = {
            timestamp: new Date().toISOString(),
            metrics: {}
        };

        for (const [type, data] of this.metrics) {
            const durations = data.map(d => d.duration).filter(d => d !== undefined);
            
            if (durations.length > 0) {
                report.metrics[type] = {
                    count: data.length,
                    averageDuration: durations.reduce((a, b) => a + b, 0) / durations.length,
                    minDuration: Math.min(...durations),
                    maxDuration: Math.max(...durations),
                    recentEntries: data.slice(-10)
                };
            }
        }

        return report;
    }

    // 清理资源
    cleanup() {
        this.observers.forEach(observer => observer.disconnect());
        this.metrics.clear();
    }
}

// 使用示例
const analyzer = new PerformanceAnalyzer();

// 装饰器使用示例
class DatabaseService {
    @analyzer.performanceDecorator
    async queryUsers() {
        // 模拟数据库查询
        await new Promise(resolve => setTimeout(resolve, 100));
        return [];
    }
}

module.exports = PerformanceAnalyzer;

4. 微服务架构与设计模式

4.1 微服务架构模式

微服务基础框架实现

// microservice-framework.js
const express = require('express');
const { EventEmitter } = require('events');
const axios = require('axios');

class MicroService extends EventEmitter {
    constructor(config) {
        super();
        this.config = {
            name: 'unnamed-service',
            port: 3000,
            version: '1.0.0',
            ...config
        };
        
        this.app = express();
        this.services = new Map();
        this.middlewares = [];
        this.routes = new Map();
        this.healthChecks = [];
        
        this.setupDefaultMiddleware();
        this.setupDefaultRoutes();
    }

    // 设置默认中间件
    setupDefaultMiddleware() {
        this.app.use(express.json());
        this.app.use(express.urlencoded({ extended: true }));
        
        // 请求追踪中间件
        this.app.use((req, res, next) => {
            req.traceId = this.generateTraceId();
            req.startTime = Date.now();
            
            res.on('finish', () => {
                const duration = Date.now() - req.startTime;
                this.emit('request', {
                    traceId: req.traceId,
                    method: req.method,
                    path: req.path,
                    statusCode: res.statusCode,
                    duration
                });
            });
            
            next();
        });
    }

    // 设置默认路由
    setupDefaultRoutes() {
        // 健康检查
        this.app.get('/health', async (req, res) => {
            const health = await this.performHealthCheck();
            const statusCode = health.status === 'healthy' ? 200 : 503;
            res.status(statusCode).json(health);
        });

        // 服务信息
        this.app.get('/info', (req, res) => {
            res.json({
                name: this.config.name,
                version: this.config.version,
                uptime: process.uptime(),
                memory: process.memoryUsage(),
                pid: process.pid
            });
        });

        // 指标端点
        this.app.get('/metrics', (req, res) => {
            res.json(this.getMetrics());
        });
    }

    // 注册服务
    registerService(name, config) {
        this.services.set(name, {
            name,
            url: config.url,
            timeout: config.timeout || 5000,
            retries: config.retries || 3,
            circuitBreaker: new CircuitBreaker(config.circuitBreaker)
        });
    }

    // 调用其他服务
    async callService(serviceName, path, options = {}) {
        const service = this.services.get(serviceName);
        if (!service) {
            throw new Error(`Service ${serviceName} not registered`);
        }

        const url = `${service.url}${path}`;
        const config = {
            timeout: service.timeout,
            headers: {
                'X-Trace-ID': options.traceId || this.generateTraceId(),
                'X-Service-Name': this.config.name,
                ...options.headers
            },
            ...options
        };

        try {
            return await service.circuitBreaker.execute(() => 
                axios(url, config)
            );
        } catch (error) {
            this.emit('service-call-error', {
                service: serviceName,
                url,
                error: error.message
            });
            throw error;
        }
    }

    // 添加中间件
    use(middleware) {
        this.middlewares.push(middleware);
        this.app.use(middleware);
    }

    // 添加路由
    route(method, path, handler) {
        const routeKey = `${method.toUpperCase()} ${path}`;
        this.routes.set(routeKey, handler);
        this.app[method.toLowerCase()](path, handler);
    }

    // 添加健康检查
    addHealthCheck(name, checkFunction) {
        this.healthChecks.push({ name, check: checkFunction });
    }

    // 执行健康检查
    async performHealthCheck() {
        const results = await Promise.allSettled(
            this.healthChecks.map(async ({ name, check }) => {
                try {
                    const result = await check();
                    return { name, status: 'healthy', ...result };
                } catch (error) {
                    return { name, status: 'unhealthy', error: error.message };
                }
            })
        );

        const checks = results.map(result => result.value || result.reason);
        const allHealthy = checks.every(check => check.status === 'healthy');

        return {
            status: allHealthy ? 'healthy' : 'unhealthy',
            timestamp: new Date().toISOString(),
            checks
        };
    }

    // 获取指标
    getMetrics() {
        return {
            service: this.config.name,
            version: this.config.version,
            uptime: process.uptime(),
            memory: process.memoryUsage(),
            cpu: process.cpuUsage(),
            routes: Array.from(this.routes.keys()),
            services: Array.from(this.services.keys())
        };
    }

    // 生成追踪ID
    generateTraceId() {
        return Math.random().toString(36).substr(2, 9);
    }

    // 启动服务
    async start() {
        return new Promise((resolve, reject) => {
            this.server = this.app.listen(this.config.port, (error) => {
                if (error) {
                    reject(error);
                } else {
                    console.log(`${this.config.name} started on port ${this.config.port}`);
                    this.emit('started');
                    resolve();
                }
            });
        });
    }

    // 停止服务
    async stop() {
        return new Promise((resolve) => {
            if (this.server) {
                this.server.close(() => {
                    console.log(`${this.config.name} stopped`);
                    this.emit('stopped');
                    resolve();
                });
            } else {
                resolve();
            }
        });
    }
}

// 熔断器实现
class CircuitBreaker {
    constructor(options = {}) {
        this.failureThreshold = options.failureThreshold || 5;
        this.timeout = options.timeout || 60000;
        this.monitoringPeriod = options.monitoringPeriod || 10000;
        
        this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
        this.failureCount = 0;
        this.lastFailureTime = null;
        this.successCount = 0;
    }

    async execute(operation) {
        if (this.state === 'OPEN') {
            if (Date.now() - this.lastFailureTime >= this.timeout) {
                this.state = 'HALF_OPEN';
                this.successCount = 0;
            } else {
                throw new Error('Circuit breaker is OPEN');
            }
        }

        try {
            const result = await operation();
            this.onSuccess();
            return result;
        } catch (error) {
            this.onFailure();
            throw error;
        }
    }

    onSuccess() {
        this.failureCount = 0;
        
        if (this.state === 'HALF_OPEN') {
            this.successCount++;
            if (this.successCount >= 3) {
                this.state = 'CLOSED';
            }
        }
    }

    onFailure() {
        this.failureCount++;
        this.lastFailureTime = Date.now();
        
        if (this.failureCount >= this.failureThreshold) {
            this.state = 'OPEN';
        }
    }
}

module.exports = { MicroService, CircuitBreaker };

4.2 设计模式在 Node.js 中的应用

高级设计模式实现

// design-patterns.js

// 1. 单例模式 - 数据库连接管理
class DatabaseManager {
    constructor() {
        if (DatabaseManager.instance) {
            return DatabaseManager.instance;
        }
        
        this.connections = new Map();
        DatabaseManager.instance = this;
    }

    async getConnection(config) {
        const key = `${config.host}:${config.port}/${config.database}`;
        
        if (!this.connections.has(key)) {
            const connection = await this.createConnection(config);
            this.connections.set(key, connection);
        }
        
        return this.connections.get(key);
    }

    async createConnection(config) {
        // 模拟数据库连接创建
        return {
            host: config.host,
            port: config.port,
            database: config.database,
            connected: true
        };
    }
}

// 2. 工厂模式 - 日志记录器工厂
class LoggerFactory {
    static createLogger(type, config) {
        switch (type) {
            case 'console':
                return new ConsoleLogger(config);
            case 'file':
                return new FileLogger(config);
            case 'database':
                return new DatabaseLogger(config);
            default:
                throw new Error(`Unknown logger type: ${type}`);
        }
    }
}

class ConsoleLogger {
    constructor(config) {
        this.level = config.level || 'info';
    }

    log(level, message) {
        if (this.shouldLog(level)) {
            console.log(`[${level.toUpperCase()}] ${message}`);
        }
    }

    shouldLog(level) {
        const levels = ['debug', 'info', 'warn', 'error'];
        return levels.indexOf(level) >= levels.indexOf(this.level);
    }
}

// 3. 装饰器模式 - 缓存装饰器
function cacheDecorator(ttl = 300000) {
    const cache = new Map();
    
    return function(target, propertyKey, descriptor) {
        const originalMethod = descriptor.value;
        
        descriptor.value = async function(...args) {
            const cacheKey = `${propertyKey}:${JSON.stringify(args)}`;
            const cached = cache.get(cacheKey);
            
            if (cached && Date.now() - cached.timestamp < ttl) {
                return cached.value;
            }
            
            const result = await originalMethod.apply(this, args);
            cache.set(cacheKey, {
                value: result,
                timestamp: Date.now()
            });
            
            return result;
        };
        
        return descriptor;
    };
}

// 4. 策略模式 - 支付处理策略
class PaymentProcessor {
    constructor() {
        this.strategies = new Map();
    }

    addStrategy(name, strategy) {
        this.strategies.set(name, strategy);
    }

    async processPayment(method, amount, details) {
        const strategy = this.strategies.get(method);
        if (!strategy) {
            throw new Error(`Payment method ${method} not supported`);
        }
        
        return await strategy.process(amount, details);
    }
}

class CreditCardStrategy {
    async process(amount, details) {
        // 信用卡支付逻辑
        return {
            success: true,
            transactionId: `cc_${Date.now()}`,
            amount,
            method: 'credit_card'
        };
    }
}

class PayPalStrategy {
    async process(amount, details) {
        // PayPal 支付逻辑
        return {
            success: true,
            transactionId: `pp_${Date.now()}`,
            amount,
            method: 'paypal'
        };
    }
}

// 5. 观察者模式 - 事件系统
class EventManager {
    constructor() {
        this.listeners = new Map();
    }

    subscribe(event, callback) {
        if (!this.listeners.has(event)) {
            this.listeners.set(event, []);
        }
        
        this.listeners.get(event).push(callback);
        
        // 返回取消订阅函数
        return () => {
            const callbacks = this.listeners.get(event);
            const index = callbacks.indexOf(callback);
            if (index > -1) {
                callbacks.splice(index, 1);
            }
        };
    }

    async publish(event, data) {
        const callbacks = this.listeners.get(event) || [];
        
        // 并行执行所有回调
        await Promise.allSettled(
            callbacks.map(callback => callback(data))
        );
    }
}

// 6. 命令模式 - 任务队列
class Command {
    constructor(execute, undo) {
        this.execute = execute;
        this.undo = undo;
    }
}

class TaskQueue {
    constructor() {
        this.commands = [];
        this.currentIndex = -1;
    }

    execute(command) {
        // 移除当前位置之后的命令
        this.commands = this.commands.slice(0, this.currentIndex + 1);
        
        // 添加新命令
        this.commands.push(command);
        this.currentIndex++;
        
        // 执行命令
        return command.execute();
    }

    undo() {
        if (this.currentIndex >= 0) {
            const command = this.commands[this.currentIndex];
            this.currentIndex--;
            return command.undo();
        }
    }

    redo() {
        if (this.currentIndex < this.commands.length - 1) {
            this.currentIndex++;
            const command = this.commands[this.currentIndex];
            return command.execute();
        }
    }
}

// 7. 中介者模式 - 聊天室
class ChatRoom {
    constructor() {
        this.users = new Map();
        this.rooms = new Map();
    }

    addUser(user) {
        this.users.set(user.id, user);
        user.setChatRoom(this);
    }

    removeUser(userId) {
        this.users.delete(userId);
    }

    sendMessage(fromUserId, toUserId, message) {
        const toUser = this.users.get(toUserId);
        if (toUser) {
            toUser.receiveMessage(fromUserId, message);
        }
    }

    broadcast(fromUserId, message, roomId) {
        const room = this.rooms.get(roomId);
        if (room) {
            room.members.forEach(userId => {
                if (userId !== fromUserId) {
                    this.sendMessage(fromUserId, userId, message);
                }
            });
        }
    }
}

class User {
    constructor(id, name) {
        this.id = id;
        this.name = name;
        this.chatRoom = null;
    }

    setChatRoom(chatRoom) {
        this.chatRoom = chatRoom;
    }

    sendMessage(toUserId, message) {
        if (this.chatRoom) {
            this.chatRoom.sendMessage(this.id, toUserId, message);
        }
    }

    receiveMessage(fromUserId, message) {
        console.log(`${this.name} received message from ${fromUserId}: ${message}`);
    }
}

module.exports = {
    DatabaseManager,
    LoggerFactory,
    cacheDecorator,
    PaymentProcessor,
    CreditCardStrategy,
    PayPalStrategy,
    EventManager,
    Command,
    TaskQueue,
    ChatRoom,
    User
};

5. 实时通信与 WebSocket

5.1 实时通信架构

WebSocket 服务器实现

// websocket-server.js
const WebSocket = require('ws');
const { EventEmitter } = require('events');
const jwt = require('jsonwebtoken');

class WebSocketServer extends EventEmitter {
    constructor(options = {}) {
        super();
        
        this.options = {
            port: 8080,
            verifyClient: null,
            ...options
        };
        
        this.clients = new Map();
        this.rooms = new Map();
        this.messageHandlers = new Map();
        
        this.setupServer();
        this.setupMessageHandlers();
    }

    setupServer() {
        this.wss = new WebSocket.Server({
            port: this.options.port,
            verifyClient: this.verifyClient.bind(this)
        });

        this.wss.on('connection', this.handleConnection.bind(this));
        
        console.log(`WebSocket server started on port ${this.options.port}`);
    }

    verifyClient(info) {
        if (this.options.verifyClient) {
            return this.options.verifyClient(info);
        }
        
        // 默认验证逻辑
        const token = this.extractToken(info.req);
        if (!token) {
            return false;
        }
        
        try {
            const decoded = jwt.verify(token, process.env.JWT_SECRET);
            info.req.user = decoded;
            return true;
        } catch (error) {
            return false;
        }
    }

    extractToken(req) {
        const url = new URL(req.url, `http://${req.headers.host}`);
        return url.searchParams.get('token');
    }

    handleConnection(ws, req) {
        const clientId = this.generateClientId();
        const user = req.user;
        
        const client = {
            id: clientId,
            ws,
            user,
            rooms: new Set(),
            lastPing: Date.now(),
            metadata: {}
        };
        
        this.clients.set(clientId, client);
        
        ws.on('message', (data) => this.handleMessage(client, data));
        ws.on('close', () => this.handleDisconnection(client));
        ws.on('error', (error) => this.handleError(client, error));
        ws.on('pong', () => this.handlePong(client));
        
        // 发送连接确认
        this.sendToClient(client, {
            type: 'connection',
            clientId,
            timestamp: Date.now()
        });
        
        this.emit('connection', client);
    }

    handleMessage(client, data) {
        try {
            const message = JSON.parse(data);
            const handler = this.messageHandlers.get(message.type);
            
            if (handler) {
                handler(client, message);
            } else {
                this.sendError(client, `Unknown message type: ${message.type}`);
            }
        } catch (error) {
            this.sendError(client, 'Invalid message format');
        }
    }

    setupMessageHandlers() {
        // 加入房间
        this.messageHandlers.set('join_room', (client, message) => {
            const { roomId } = message;
            this.joinRoom(client, roomId);
        });

        // 离开房间
        this.messageHandlers.set('leave_room', (client, message) => {
            const { roomId } = message;
            this.leaveRoom(client, roomId);
        });

        // 发送消息到房间
        this.messageHandlers.set('room_message', (client, message) => {
            const { roomId, content } = message;
            this.sendToRoom(roomId, {
                type: 'room_message',
                from: client.id,
                content,
                timestamp: Date.now()
            }, client.id);
        });

        // 私聊消息
        this.messageHandlers.set('private_message', (client, message) => {
            const { targetId, content } = message;
            const targetClient = this.clients.get(targetId);
            
            if (targetClient) {
                this.sendToClient(targetClient, {
                    type: 'private_message',
                    from: client.id,
                    content,
                    timestamp: Date.now()
                });
            } else {
                this.sendError(client, 'Target client not found');
            }
        });

        // 心跳
        this.messageHandlers.set('ping', (client, message) => {
            client.lastPing = Date.now();
            this.sendToClient(client, { type: 'pong' });
        });
    }

    joinRoom(client, roomId) {
        if (!this.rooms.has(roomId)) {
            this.rooms.set(roomId, {
                id: roomId,
                clients: new Set(),
                metadata: {}
            });
        }
        
        const room = this.rooms.get(roomId);
        room.clients.add(client.id);
        client.rooms.add(roomId);
        
        // 通知房间内其他用户
        this.sendToRoom(roomId, {
            type: 'user_joined',
            userId: client.id,
            roomId,
            timestamp: Date.now()
        }, client.id);
        
        // 发送房间信息给新用户
        this.sendToClient(client, {
            type: 'room_joined',
            roomId,
            users: Array.from(room.clients),
            timestamp: Date.now()
        });
    }

    leaveRoom(client, roomId) {
        const room = this.rooms.get(roomId);
        if (room) {
            room.clients.delete(client.id);
            client.rooms.delete(roomId);
            
            // 如果房间为空，删除房间
            if (room.clients.size === 0) {
                this.rooms.delete(roomId);
            } else {
                // 通知房间内其他用户
                this.sendToRoom(roomId, {
                    type: 'user_left',
                    userId: client.id,
                    roomId,
                    timestamp: Date.now()
                });
            }
        }
    }

    sendToClient(client, message) {
        if (client.ws.readyState === WebSocket.OPEN) {
            client.ws.send(JSON.stringify(message));
        }
    }

    sendToRoom(roomId, message, excludeClientId = null) {
        const room = this.rooms.get(roomId);
        if (room) {
            room.clients.forEach(clientId => {
                if (clientId !== excludeClientId) {
                    const client = this.clients.get(clientId);
                    if (client) {
                        this.sendToClient(client, message);
                    }
                }
            });
        }
    }

    broadcast(message, excludeClientId = null) {
        this.clients.forEach((client, clientId) => {
            if (clientId !== excludeClientId) {
                this.sendToClient(client, message);
            }
        });
    }

    sendError(client, error) {
        this.sendToClient(client, {
            type: 'error',
            message: error,
            timestamp: Date.now()
        });
    }

    handleDisconnection(client) {
        // 从所有房间中移除客户端
        client.rooms.forEach(roomId => {
            this.leaveRoom(client, roomId);
        });
        
        this.clients.delete(client.id);
        this.emit('disconnection', client);
    }

    handleError(client, error) {
        console.error(`WebSocket error for client ${client.id}:`, error);
        this.emit('error', { client, error });
    }

    handlePong(client) {
        client.lastPing = Date.now();
    }

    generateClientId() {
        return Math.random().toString(36).substr(2, 9);
    }

    // 健康检查 - 清理断开的连接
    startHealthCheck() {
        setInterval(() => {
            const now = Date.now();
            const timeout = 30000; // 30秒超时
            
            this.clients.forEach((client, clientId) => {
                if (now - client.lastPing > timeout) {
                    console.log(`Client ${clientId} timed out`);
                    client.ws.terminate();
                    this.handleDisconnection(client);
                }
            });
        }, 10000); // 每10秒检查一次
    }

    // 获取统计信息
    getStats() {
        return {
            totalClients: this.clients.size,
            totalRooms: this.rooms.size,
            roomDetails: Array.from(this.rooms.values()).map(room => ({
                id: room.id,
                clientCount: room.clients.size
            }))
        };
    }
}

module.exports = WebSocketServer;