path: root/src/server/DashSession/Session/agents/monitor.ts

import { ExecOptions, exec } from 'child_process';
import * as _cluster from 'cluster';
import { Worker } from 'cluster';
import { blue, cyan, red, white, yellow } from 'colors';
import { readFileSync } from 'fs';
import { ValidationError, validate } from 'jsonschema';
import Repl, { ReplAction } from '../utilities/repl';
import { Configuration, Identifiers, colorMapping, configurationSchema, defaultConfig } from '../utilities/session_config';
import { Utilities } from '../utilities/utilities';
import { ExitHandler } from './applied_session_agent';
import IPCMessageReceiver from './process_message_router';
import { ErrorLike, MessageHandler, manage } from './promisified_ipc_manager';
import { ServerWorker } from './server_worker';

const cluster = _cluster as any;
const { isWorker, setupMaster, on, fork } = cluster;

/**
 * Validates and reads the configuration file, accordingly builds a child process factory
 * and spawns off an initial process that will respawn as predecessors die.
 */
export class Monitor extends IPCMessageReceiver {
    private static count = 0;
    private finalized = false;
    private exitHandlers: ExitHandler[] = [];
    private readonly config: Configuration;
    private activeWorker: Worker | undefined;
    private key: string | undefined;
    private repl: Repl;

    public static Create() {
        if (isWorker) {
            ServerWorker.IPCManager.emit('kill', {
                reason: 'cannot create a monitor on the worker process.',
                graceful: false,
                errorCode: 1,
            });
            process.exit(1);
        } else if (++Monitor.count > 1) {
            console.error(red('cannot create more than one monitor.'));
            process.exit(1);
        }
        return new Monitor();
    }

    private constructor() {
        super();
        console.log(this.timestamp(), cyan('initializing session...'));
        this.configureInternalHandlers();
        this.config = this.loadAndValidateConfiguration();
        this.initializeClusterFunctions();
        this.repl = this.initializeRepl();
    }

    protected configureInternalHandlers = () => {
        // handle exceptions in the master thread - there shouldn't be many of these
        // the IPC (inter process communication) channel closed exception can't seem
        // to be caught in a try catch, and is inconsequential, so it is ignored
        process.on('uncaughtException', ({ message, stack }): void => {
            if (message !== 'Channel closed') {
                this.mainLog(red(message));
                if (stack) {
                    this.mainLog(`uncaught exception\n${red(stack)}`);
                }
            }
        });

        this.on('kill', ({ reason, graceful, errorCode }) => this.killSession(reason, graceful, errorCode));
        this.on('lifecycle', ({ event }) => console.log(this.timestamp(), `${this.config.identifiers.worker.text} lifecycle phase (${event})`));
    };

    private initializeClusterFunctions = () => {
        // determines whether or not we see the compilation / initialization / runtime output of each child server process
        const output = this.config.showServerOutput ? 'inherit' : 'ignore';
        setupMaster({ stdio: ['ignore', output, output, 'ipc'] });

        // a helpful cluster event called on the master thread each time a child process exits
        on('exit', ({ process: { pid } }: { process: { pid: any } }, code: any, signal: any) => {
            const prompt = `server worker with process id ${pid} has exited with code ${code}${signal === null ? '' : `, having encountered signal ${signal}`}.`;
            this.mainLog(cyan(prompt));
            // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one
            this.spawn();
        });
    };

    public finalize = (sessionKey: string): void => {
        if (this.finalized) {
            throw new Error('Session monitor is already finalized');
        }
        this.finalized = true;
        this.key = sessionKey;
        this.spawn();
    };

    public readonly coreHooks = Object.freeze({
        onCrashDetected: (listener: MessageHandler<{ error: ErrorLike }>) => this.on(Monitor.IntrinsicEvents.CrashDetected, listener),
        onServerRunning: (listener: MessageHandler<{ isFirstTime: boolean }>) => this.on(Monitor.IntrinsicEvents.ServerRunning, listener),
    });

    /**
     * Kill this session and its active child
     * server process, either gracefully (may wait
     * indefinitely, but at least allows active networking
     * requests to complete) or immediately.
     */
    public killSession = async (reason: string, graceful = true, errorCode = 0) => {
        this.mainLog(cyan(`exiting session ${graceful ? 'clean' : 'immediate'}ly`));
        this.mainLog(`session exit reason: ${red(reason)}`);
        await this.executeExitHandlers(true);
        await this.killActiveWorker(graceful, true);
        process.exit(errorCode);
    };

    /**
     * Execute the list of functions registered to be called
     * whenever the process exits.
     */
    public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler);

    /**
     * Extend the default repl by adding in custom commands
     * that can invoke application logic external to this module
     */
    public addReplCommand = (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => {
        this.repl.registerCommand(basename, argPatterns, action);
    };

    public exec = (command: string, options?: ExecOptions) =>
        new Promise<void>(resolve => {
            exec(command, { ...options, encoding: 'utf8' }, (error, stdout, stderr) => {
                if (error) {
                    this.execLog(red(`unable to execute ${white(command)}`));
                    error.message.split('\n').forEach(line => line.length && this.execLog(red(`(error) ${line}`)));
                } else {
                    const outLines = stdout.split('\n').filter(line => line.length);
                    if (outLines.length) {
                        outLines.forEach(line => line.length && this.execLog(cyan(`(stdout) ${line}`)));
                    }
                    const errorLines = stderr.split('\n').filter(line => line.length);
                    if (errorLines.length) {
                        errorLines.forEach(line => line.length && this.execLog(yellow(`(stderr) ${line}`)));
                    }
                }
                resolve();
            });
        });

    /**
     * Generates a blue UTC string associated with the time
     * of invocation.
     */
    private timestamp = () => blue(`[${new Date().toUTCString()}]`);

    /**
     * A formatted, identified and timestamped log in color
     */
    public mainLog = (...optionalParams: any[]) => {
        console.log(this.timestamp(), this.config.identifiers.master.text, ...optionalParams);
    };

    /**
     * A formatted, identified and timestamped log in color for non-
     */
    private execLog = (...optionalParams: any[]) => {
        console.log(this.timestamp(), this.config.identifiers.exec.text, ...optionalParams);
    };

    /**
     * Reads in configuration .json file only once, in the master thread
     * and pass down any variables the pertinent to the child processes as environment variables.
     */
    private loadAndValidateConfiguration = (): Configuration => {
        let config: Configuration | undefined;
        try {
            console.log(this.timestamp(), cyan('validating configuration...'));
            config = JSON.parse(readFileSync('./session.config.json', 'utf8'));
            const options = {
                throwError: true,
                allowUnknownAttributes: false,
            };
            // ensure all necessary and no excess information is specified by the configuration file
            validate(config, configurationSchema, options);
            config = Utilities.preciseAssign({}, defaultConfig, config);
        } catch (error: any) {
            if (error instanceof ValidationError) {
                console.log(red('\nSession configuration failed.'));
                console.log('The given session.config.json configuration file is invalid.');
                console.log(`${error.instance}: ${error.stack}`);
                process.exit(0);
            } else if (error.code === 'ENOENT' && error.path === './session.config.json') {
                console.log(cyan('Loading default session parameters...'));
                console.log('Consider including a session.config.json configuration file in your project root for customization.');
                config = Utilities.preciseAssign({}, defaultConfig);
            } else {
                console.log(red('\nSession configuration failed.'));
                console.log('The following unknown error occurred during configuration.');
                console.log(error.stack);
                process.exit(0);
            }
        } finally {
            const { identifiers } = config!;
            Object.keys(identifiers).forEach(key => {
                const resolved = key as keyof Identifiers;
                const { text, color } = identifiers[resolved];
                identifiers[resolved].text = (colorMapping.get(color) || white)(`${text}:`);
            });
            return config!;
        }
    };

    /**
     * Builds the repl that allows the following commands to be typed into stdin of the master thread.
     */
    private initializeRepl = (): Repl => {
        const repl = new Repl({ identifier: () => `${this.timestamp()} ${this.config.identifiers.master.text}` });
        const boolean = /true|false/;
        const number = /\d+/;
        const letters = /[a-zA-Z]+/;
        repl.registerCommand('exit', [/clean|force/], args => this.killSession('manual exit requested by repl', args[0] === 'clean', 0));
        repl.registerCommand('restart', [/clean|force/], args => this.killActiveWorker(args[0] === 'clean'));
        repl.registerCommand('set', [letters, 'port', number, boolean], args => this.setPort(args[0], Number(args[2]), args[3] === 'true'));
        repl.registerCommand('set', [/polling/, number, boolean], args => {
            const newPollingIntervalSeconds = Math.floor(Number(args[1]));
            if (newPollingIntervalSeconds < 0) {
                this.mainLog(red('the polling interval must be a non-negative integer'));
            } else if (newPollingIntervalSeconds !== this.config.polling.intervalSeconds) {
                this.config.polling.intervalSeconds = newPollingIntervalSeconds;
                if (args[2] === 'true') {
                    Monitor.IPCManager.emit('updatePollingInterval', { newPollingIntervalSeconds });
                }
            }
        });
        return repl;
    };

    private executeExitHandlers = async (reason: Error | boolean) => Promise.all(this.exitHandlers.map(handler => handler(reason)));

    /**
     * Attempts to kill the active worker gracefully, unless otherwise specified.
     */
    private killActiveWorker = async (graceful = true, isSessionEnd = false): Promise<void> => {
        if (this.activeWorker && !this.activeWorker.isDead()) {
            if (graceful) {
                Monitor.IPCManager.emit('manualExit', { isSessionEnd });
            } else {
                await ServerWorker.IPCManager.destroy();
                this.activeWorker.process.kill();
            }
        }
    };

    /**
     * Allows the caller to set the port at which the target (be it the server,
     * the websocket, some other custom port) is listening. If an immediate restart
     * is specified, this monitor will kill the active child and re-launch the server
     * at the port. Otherwise, the updated port won't be used until / unless the child
     * dies on its own and triggers a restart.
     */
    private setPort = (port: 'server' | 'socket' | string, value: number, immediateRestart: boolean): void => {
        if (value > 1023 && value < 65536) {
            this.config.ports[port] = value;
            if (immediateRestart) {
                this.killActiveWorker();
            }
        } else {
            this.mainLog(red(`${port} is an invalid port number`));
        }
    };

    /**
     * Kills the current active worker and proceeds to spawn a new worker,
     * feeding in configuration information as environment variables.
     */
    private spawn = async (): Promise<void> => {
        await this.killActiveWorker();
        const {
            config: { polling, ports },
            key,
        } = this;
        this.activeWorker = fork({
            pollingRoute: polling.route,
            pollingFailureTolerance: polling.failureTolerance,
            serverPort: ports.server,
            socketPort: ports.socket,
            pollingIntervalSeconds: polling.intervalSeconds,
            session_key: key,
        });
        if (this.activeWorker) {
            Monitor.IPCManager = manage(this.activeWorker.process, this.handlers);
        }
        this.mainLog(cyan(`spawned new server worker with process id ${this.activeWorker?.process.pid}`));
    };
}

// eslint-disable-next-line no-redeclare
export namespace Monitor {
    export enum IntrinsicEvents {
        KeyGenerated = 'key_generated',
        CrashDetected = 'crash_detected',
        ServerRunning = 'server_running',
    }
}