node_server/index.ts

import config from "./config.json" with {type: "json"};
import { io } from "socket.io-client";
import {console_print} from "./utils/console_print.ts";
import {event_handler, init_event_handler_list} from "./event_handler/init.ts";
import {update_plugins} from "./service/server/update/plugins.ts";
import {update_version_list} from "./service/server/update/version_list.ts";
import {update_function_list} from './service/server/update/function_list.ts';
import {update_workflow_catalog} from './service/server/update/workflow_catalog.ts';
import {start_forward_server, type ControlTransport, type ForwardServerHandle} from "./forward_server/index.ts";
import {update_all_bot_status} from "./service/server/update/bot_status.ts";
import {bind_node_session, clear_node_session} from "./bot/forward/manager.ts";
import {Buffer} from 'node:buffer';
import {get_bot_context_status, stop_bot_context} from './bot/instance/index.ts';

async function main() {
    await init_event_handler_list();

    const {main_server, node} = config;
    if (!main_server || typeof main_server.host !== 'string' || !main_server.host.trim()
        || !Number.isInteger(main_server.port) || main_server.port < 1 || main_server.port > 65535
        || typeof main_server.use_https !== 'boolean'
        || !node || !Number.isSafeInteger(node.id) || node.id < 1
        || typeof node.key !== 'string' || !node.key.trim()) {
        console_print.log(console_print.error('配置文件错误，请检查 main_server 和 node 配置'));
        throw new Error('Invalid node configuration');
    }

    const reconnect_interval = retry_interval();

    const protocol = main_server.use_https ? 'https' : 'http';
    const main_server_url = `${protocol}://${main_server.host}:${main_server.port}`;
    const main_server_socket = io(main_server_url, {
        path: '/api/node',
        autoConnect: false,
        timeout: 10_000,
        reconnection: false,
    });
    let reconnect_timer: ReturnType<typeof setTimeout> | undefined;
    let reconnect_attempts = 0;
    let connection_sequence = 0;
    let stopped = false;
    let node_session_id: string | undefined;
    let relay_handle: ForwardServerHandle | undefined;
    let relay_starting: Promise<ForwardServerHandle> | undefined;
    let release_control: (() => void) | undefined;
    let shutdown: Promise<void> | undefined;
    let signal_exit_requested = false;

    function clear_session() {
        node_session_id = undefined;
        clear_node_session(main_server_socket);
        release_control?.();
        release_control = undefined;
    }

    function authenticated(sequence: number) {
        return !stopped && sequence === connection_sequence
            && main_server_socket.connected && node_session_id !== undefined;
    }

    function control_transport(sequence: number): ControlTransport {
        const listeners = new Map<(...args: any[]) => void, (...args: any[]) => void>();
        return {
            on(event, listener) {
                const guarded = (...args: any[]) => {
                    // Hosted relay commands may target any Node, not just this host.
                    const disconnected = event === 'disconnect' && !stopped
                        && !main_server_socket.connected && connection_sequence === sequence + 1;
                    if (authenticated(sequence) || disconnected) listener(...args);
                };
                listeners.set(listener, guarded);
                main_server_socket.on(event, guarded);
            },
            off(event, listener) {
                const guarded = listeners.get(listener);
                if (guarded) main_server_socket.off(event, guarded);
                listeners.delete(listener);
            },
            emit(event, ...args) {
                if (authenticated(sequence)) main_server_socket.emit(event, ...args);
            },
        };
    }

    function stop_connection() {
        if (!stopped) {
            stopped = true;
            connection_sequence++;
            clearTimeout(reconnect_timer);
            reconnect_timer = undefined;
            clear_session();
            main_server_socket.disconnect();
            process.exitCode = 0;
            shutdown = (async () => {
                let deadline: ReturnType<typeof setTimeout> | undefined;
                try {
                    await Promise.race([
                        Promise.allSettled([
                            (async () => {
                                // A failed startup has no handle to stop; it is already reported by initialization.
                                const handle = relay_handle ?? await relay_starting?.catch((): undefined => undefined);
                                await handle?.stop();
                            })(),
                            (async () => {
                                const contexts = await get_bot_context_status();
                                const results = await Promise.allSettled(contexts.map(({bot_id}) => stop_bot_context(bot_id)));
                                if (results.some(result => result.status === 'rejected')) throw new Error('Bot cleanup failed');
                            })(),
                        ]).then(results => {
                            if (results.some(result => result.status === 'rejected')) throw new Error('Shutdown failed');
                        }),
                        new Promise<never>((_resolve, reject) => {
                            deadline = setTimeout(() => reject(new Error('Shutdown timed out')), 10_000);
                        }),
                    ]);
                } catch {
                    console_print.log(console_print.error('节点关闭失败或超时'));
                    process.exitCode = 1;
                } finally {
                    clearTimeout(deadline);
                }
            })();
        }
        if (!signal_exit_requested) {
            signal_exit_requested = true;
            // Signal listeners override Node's default exit; do not leave bot timers alive.
            void shutdown!.then(() => process.exit(process.exitCode ?? 0));
        }
    }

    process.once('SIGINT', stop_connection);
    process.once('SIGTERM', stop_connection);

    function schedule_reconnect() {
        if (stopped || reconnect_timer !== undefined) {
            return;
        }
        console_print.log(console_print.info(`将在 ${reconnect_interval} 毫秒后重连主服务器，持续重试直至成功`));
        reconnect_timer = setTimeout(() => {
            reconnect_timer = undefined;
            reconnect_attempts++;
            console_print.log(console_print.info(`正在尝试重连主服务器，第 ${reconnect_attempts} 次`));
            try {
                main_server_socket.connect();
            } catch {
                console_print.log(console_print.error('连接主服务器失败，将继续重试'));
                schedule_reconnect();
            }
        }, reconnect_interval);
    }

    function retry_connection() {
        if (stopped) return;
        connection_sequence++;
        clear_session();
        main_server_socket.disconnect();
        schedule_reconnect();
    }

    main_server_socket.on('connect', async () => {
        if (stopped) return;
        const sequence = ++connection_sequence;
        clearTimeout(reconnect_timer);
        reconnect_timer = undefined;
        clear_session();
        console_print.log(console_print.info('已连接主服务器，正在注册节点'));
        try {
            const response = await main_server_socket
                .timeout(5000)
                .emitWithAck('register_node', {
                    id: node.id,
                    key: node.key,
                });
            // 旧连接的 ACK 不得影响重连后的注册状态。
            if (stopped || sequence !== connection_sequence || !main_server_socket.connected) {
                return;
            }
            if (!response || (response.status !== 'success' && response.status !== 'error')) {
                throw new Error('节点注册响应格式无效');
            }
            if (response.status === 'error') {
                console_print.log(console_print.error('节点注册被主服务器拒绝'));
                retry_connection();
                return;
            }
            const {data} = response;
            if (Object.keys(response).length !== 2 || !data || Array.isArray(data)
                || Object.keys(data).length !== 2 || typeof data.name !== 'string'
                || typeof data.node_session_id !== 'string' || !data.node_session_id.trim()
                || Buffer.byteLength(data.node_session_id, 'utf8') > 256) {
                throw new Error('节点注册响应格式无效');
            }
            node_session_id = data.node_session_id;
            bind_node_session(main_server_socket, node.id, node_session_id);
            console_print.log(console_print.success('节点注册成功'));
        } catch {
            if (stopped || sequence !== connection_sequence) {
                return;
            }
            console_print.log(console_print.error('节点注册失败，等待重新连接'));
            retry_connection();
            return;
        }
        let initialization_stage = '插件列表上报';
        try {
            // @ts-ignore
            await update_plugins(main_server_socket, node.id, () => authenticated(sequence));
            if (!authenticated(sequence)) return;
            initialization_stage = '函数列表上报';
            await update_function_list(main_server_socket, node.id, () => authenticated(sequence));
            if (!authenticated(sequence)) return;
            initialization_stage = '工作流目录上报';
            await update_workflow_catalog(main_server_socket, node.id, () => authenticated(sequence));
            if (!authenticated(sequence)) return;
            initialization_stage = '版本列表上报';
            // @ts-ignore
            await update_version_list(main_server_socket, node.id);
            if (!authenticated(sequence)) return;
            initialization_stage = '机器人状态上报';
            // @ts-ignore
            await update_all_bot_status(main_server_socket, node.id);
            if (!authenticated(sequence)) return;
            if (config.forward_server.status) {
                initialization_stage = '转发服务器启动';
                const control = control_transport(sequence);
                if (!relay_starting) {
                    relay_starting = start_forward_server({
                        forward_id: config.forward_server.id,
                        key: config.forward_server.key,
                        allowed_origins: config.forward_server.allowed_origins,
                        control,
                        path: config.forward_server.path,
                        namespace: config.forward_server.namespace,
                        port: config.forward_server.port,
                        logger: console_print,
                    }).then(handle => {
                        relay_handle = handle;
                        return handle;
                    }).catch(error => {
                        relay_starting = undefined;
                        throw error;
                    });
                }
                const handle = await relay_starting;
                if (!authenticated(sequence)) return;
                release_control = handle.bind_control(control);
                console_print.log(console_print.info('转发服务器已就绪'));
                try {
                    const response = await main_server_socket
                        .timeout(5000)
                        .emitWithAck('register_forward', {
                            id: config.forward_server.id,
                            key: config.forward_server.key,
                            node_id: node.id,
                        });
                    if (!authenticated(sequence)) {
                        return;
                    }
                    if (!response || (response.status !== 'success' && response.status !== 'error')) {
                        throw new Error('转发服务器注册响应格式无效');
                    }
                    if (response.status === 'error') {
                        const code = Number.isSafeInteger(response.data) ? String(response.data) : '未知';
                        console_print.log(console_print.error(`转发服务器注册被主服务器拒绝，错误码: ${code}`));
                        retry_connection();
                        return;
                    }
                    console_print.log(console_print.success('转发服务器注册成功'));
                } catch {
                    if (!authenticated(sequence)) {
                        return;
                    }
                    console_print.log(console_print.error('转发服务器注册请求异常'));
                    retry_connection();
                    return;
                }
            }
            reconnect_attempts = 0;
        } catch (error) {
            if (!authenticated(sequence)) return;
            let reason = '操作失败，错误详情已隐藏';
            if (initialization_stage === '转发服务器启动') {
                const messages: Record<string, string> = {
                    'A relay secret of at least 16 characters is required': 'forward_server.key 至少需要 16 个字符，且必须与 Main 登记值一致',
                    'forward_id must be a nonnegative safe integer': 'forward_server.id 必须是非负安全整数',
                    'allowed_origins must contain explicit HTTP(S) origins': 'forward_server.allowed_origins 必须是明确的 HTTP(S) Origin',
                    'path must be an absolute Socket.IO path without a trailing slash': 'forward_server.path 格式无效',
                    'namespace must be an absolute Socket.IO namespace': 'forward_server.namespace 格式无效',
                    'Invalid relay port': 'forward_server.port 格式无效',
                };
                if (error && typeof error === 'object' && 'message' in error && typeof error.message === 'string'
                    && Object.hasOwn(messages, error.message)) reason = messages[error.message];
                if (error && typeof error === 'object' && 'code' in error) {
                    if (error.code === 'EADDRINUSE') reason = '转发服务器监听端口已被占用';
                    if (error.code === 'EACCES') reason = '没有权限监听转发服务器端口';
                }
            }
            console_print.log(console_print.error(`节点注册后初始化失败，阶段: ${initialization_stage}，原因: ${reason}`));
            retry_connection();
        }
    });

    main_server_socket.on('connect_error', () => {
        console_print.log(console_print.error('连接主服务器失败'));
        schedule_reconnect();
    });

    main_server_socket.on('error', () => {
        console_print.log(console_print.error('主服务器返回错误'));
        retry_connection();
    });

    main_server_socket.on('disconnect', () => {
        connection_sequence++;
        clear_session();
        console_print.log(console_print.warning('与主服务器断开连接'));
        if (stopped) {
            return;
        }
        schedule_reconnect();
    });

    main_server_socket.onAny((event: string, ...args: any[]) => {
        if (!authenticated(connection_sequence)) return;
        void event_handler(main_server_socket, event, ...args).catch(() => {
            // event_handler 已记录错误，避免重复打印和未处理的 Promise 异常。
        });
    });

    console_print.log(console_print.info('正在连接主服务器'));
    try {
        main_server_socket.connect();
    } catch {
        console_print.log(console_print.error('连接主服务器失败，将继续重试'));
        schedule_reconnect();
    }
}

function retry_interval(): number {
    const interval = config.main_server?.auto_reconnect?.interval;
    return Number.isInteger(interval) && interval >= 1 && interval <= 2_147_483_647 ? interval : 10_000;
}

function bootstrap() {
    void main().catch(() => {
        console_print.log(console_print.error('节点启动失败，将继续重试'));
        setTimeout(bootstrap, retry_interval());
    });
}

bootstrap();