import type winston from "winston"; import { BlockList, type AddressInfo, isIPv4, isIPv6 } from "net"; import type { ControllerArgs } from "../controller.js"; import type ControlConnection from "./ControlConnection.js"; import express, { type Request, type Response, type NextFunction, type Application } from "express"; import type { Static } from "@sinclair/typebox"; import finalhandler from "finalhandler"; import compression from "compression"; import events from "events"; import fs from "node:fs/promises"; import http from "http"; import https from "https"; import jwt from "jsonwebtoken"; import path from "path"; import semver from "semver"; import stream from "stream"; import { fileURLToPath, pathToFileURL } from "node:url"; import * as lib from "@clusterio/lib"; import { logger, Summary, Gauge } from "@clusterio/lib"; import HttpCloser from "./HttpCloser.js"; import InstanceRecord from "./InstanceRecord.js"; import * as metrics from "./metrics.js"; import * as routes from "./routes.js"; import User from "./User.js"; import UserRecord from "./UserRecord.js"; import UserManager from "./UserManager.js"; import WsServer from "./WsServer.js"; import HostRecord from "./HostRecord.js"; import HostPairingManager from "./HostPairingManager.js"; import { BaseControllerPlugin, ControllerHooks } from "./BaseControllerPlugin.js"; import ControllerRouter from "./ControllerRouter.js"; import InstanceManager from "./InstanceManager.js"; const endpointDurationSummary = new Summary( "clusterio_controller_http_endpoint_duration_seconds", "Time it took to respond to a an HTTP request", { labels: ["route"] } ); const logSizeGauge = new Gauge( "clusterio_controller_log_bytes", "Size of all log files currently stored on the controller." ); type InstanceId = { instanceId: number }; type ControllerDebugEvents = { "message": [ { direction: "in" | "out", content: string } ], }; type ControllerEvents = { "stop": [], }; /** * Manages all controller related operations * @alias module:controller/src/Controller */ export default class Controller { clusterLogger: winston.Logger; /** Array of plugin info objects for known plugins */ pluginInfos: lib.PluginNodeEnvInfo[]; /** Controller config. */ config: lib.ControllerConfig; app: Application; httpServer: http.Server | null = null; httpServerCloser: HttpCloser | null = null; httpsServer: https.Server | null = null; httpsServerCloser: HttpCloser | null = null; /** Hooks which plugins can attach to */ hooks = new ControllerHooks(logger); /** Plugins which are currently loaded */ loadedPlugins: Set = new Set(); /** Latest versions of plugins on npm, as shown on the plugins page */ pluginLatestVersionCache: lib.LatestVersionCache = new Map(); /** WebSocket server */ wsServer: WsServer; router = new ControllerRouter(this); trustedProxies: BlockList; debugEvents = new events.EventEmitter(); private _events = new events.EventEmitter(); /** Event subscription controller */ subscriptions: lib.SubscriptionController; /** User manager for the cluster, also contains access to the user records */ users: UserManager; /** Instance manager for the cluster, also contains access to the user records */ instances: InstanceManager; /** Pending requests from hosts to join the cluster */ hostPairing: HostPairingManager; // Possible states are new, starting, running, stopping, stopped private _state: string = "new"; private _shouldStop: boolean = false; /** * Set to true before calling stop to have the controller restart * instead of stopping. This only works if {@link Controller.canRestart} * is true. */ public shouldRestart: boolean = false; _fallbackedRequests: Map, lib.RequestHandler> = new Map(); _registeredRequests: Map, lib.RequestHandler> = new Map(); _registeredEvents = new Map, lib.EventHandler>(); _snoopedEvents = new Map(); devServer: import("vite").ViteDevServer | null = null; startedAtMs = Date.now(); autosaveInterval?: ReturnType; systemMetricsInterval?: ReturnType; logDirectory: string = ""; clusterLogIndex: lib.LogIndex | null = null; clusterLogBuildInterval: ReturnType | null = null; // Cache for mod portal requests modPortalCache = new Map(); // Cache for factorio versions factorioVersions = new lib.ValueCache(async () => { try { return await lib.fetchFactorioVersions(); } catch (err: any) { logger.warn(`Failed to retrieve Factorio versions ${err.message}`); return []; } }); // Cache for the latest stable/experimental factorio releases. Persisted to // disk so that release channel targets can still be resolved when // factorio.com is unavailable. latestReleases = new lib.ValueCache(async () => { const cachePath = path.join(this.config.get("controller.database_directory"), "latest-releases.json"); try { const releases = await lib.fetchLatestReleases(); await lib.atomicOverwriteFile(cachePath, JSON.stringify(releases, null, "\t")); return releases; } catch (err: any) { logger.warn(`Failed to retrieve Factorio releases ${err.message}`); try { return JSON.parse(await fs.readFile(cachePath, "utf8")) as lib.LatestReleases; } catch (readErr: any) { if (readErr.code !== "ENOENT") { logger.warn(`Failed to read cached Factorio releases ${readErr.message}`); } return {}; } } }); /** Create a controller with its datastores and mod store loaded from disk */ static async fromConfig( config: lib.ControllerConfig, pluginInfos: lib.PluginNodeEnvInfo[], clusterLogger: winston.Logger, canRestart = false, recoveryMode = false, ) { let databaseDirectory = config.get("controller.database_directory"); await fs.mkdir(databaseDirectory, { recursive: true }); // Migrate combined user and role file from 2.0.0-alpha.22 await UserManager.attemptMigrateLegacyUsersFile( path.join(databaseDirectory, "users.json"), path.join(databaseDirectory, "roles.json"), ); const systems = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "systems.json"), lib.SystemInfo.fromJSON.bind(lib.SystemInfo), this.migrateSystems, )); const hosts = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "hosts.json"), HostRecord.fromJSON.bind(HostRecord), this.migrateHosts, this.finaliseHosts, )); const instances = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "instances.json"), InstanceRecord.fromJSON.bind(InstanceRecord), this.migrateInstances, this.finaliseInstances, )); // Save plugin versions written while loading for (const instance of instances.values()) { if (instance.config.dirty) { instances.set(instance); } } const saves = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "saves.json"), lib.SaveDetails.fromJSON.bind(lib.SaveDetails) )); const modPacks = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "mod-packs.json"), lib.ModPack.fromJSON.bind(lib.ModPack), this.migrateModPacks, )); const roles = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "roles.json"), lib.Role.fromJSON.bind(lib.Role), this.migrateRoles, )); const users = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "users.json"), UserRecord.fromJSON.bind(UserRecord), undefined, this.finaliseUsers, )); const notes = await lib.SubscribableDatastore.fromProvider(new lib.JsonIdDatastoreProvider( path.join(databaseDirectory, "notes.json"), lib.Note.fromJSON.bind(lib.Note), )); let modsDirectory = config.get("controller.mods_directory"); await fs.mkdir(modsDirectory, { recursive: true }); const modStore = await lib.ModStore.fromDirectory(modsDirectory); await this.validateExports(modPacks, "static"); // Add default mod packs if (modPacks.size === 0) { modPacks.setMany(lib.ModPack.defaultModPacks); } // Add default roles lib.Role.ensureDefaultAdminRole(roles); lib.Role.ensureDefaultPlayerRole(roles); return new this( clusterLogger, pluginInfos, config, canRestart, recoveryMode, systems, hosts, instances, saves, modPacks, modStore, roles, users, notes, ); } constructor( clusterLogger: winston.Logger, pluginInfos: lib.PluginNodeEnvInfo[], config: lib.ControllerConfig, /** * If true indicates that there is a process monitor present that * will restart the controller on non-zero exit codes. */ public canRestart: boolean = false, /** * If true indicates the controller is in recovery mode and should * disable certain actions such as loading plugins or connecting to hosts */ public recoveryMode: boolean = false, public systems = new lib.SubscribableDatastore(), /** Mapping of host id to host record */ public hosts = new lib.SubscribableDatastore(), /** Mapping of instance id to instance record */ instances = new lib.SubscribableDatastore(), /** Mapping of save id to save details */ public saves = new lib.SubscribableDatastore(), /** Mapping of mod pack id to mod pack */ public modPacks = new lib.SubscribableDatastore(), /** Mods stored on the controller */ public modStore = new lib.ModStore(config.get("controller.mods_directory"), new Map()), /** Mapping of mod pack id to mod pack */ public roles = new lib.SubscribableDatastore(), /** Mapping of user id to user record */ users = new lib.SubscribableDatastore(), /** Mapping of note id to note attached to a resource */ public notes = new lib.SubscribableDatastore(), ) { this.clusterLogger = clusterLogger; this.pluginInfos = pluginInfos; this.config = config; this.app = express(); this.app.locals.controller = this; this.app.locals.streams = new Map(); this.trustedProxies = this.parseTrustedProxies(); this.wsServer = new WsServer(this); this.users = new UserManager(users, roles, config); this.instances = new InstanceManager(instances, this); this.hostPairing = new HostPairingManager(this); this.modStore.on("change", mod => { this.modsUpdated([mod]); }); // Handle subscriptions for all internal properties this.subscriptions = new lib.SubscriptionController(config.get("controller.max_subscription_filters")); this.subscriptions.handle(lib.SystemInfoUpdateEvent, this.handleSystemInfoSubscription.bind(this)); this.subscriptions.handle(lib.HostUpdatesEvent, this.handleHostSubscription.bind(this)); this.subscriptions.handle(lib.HostPairingUpdatesEvent, this.handleHostPairingSubscription.bind(this)); this.subscriptions.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsSubscription.bind(this)); this.subscriptions.handle( lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsSubscription.bind(this) ); this.subscriptions.handle(lib.ModPackUpdatesEvent, this.handleModPackSubscription.bind(this)); this.subscriptions.handle(lib.ModUpdatesEvent, this.handleModSubscription.bind(this)); this.subscriptions.handle(lib.UserUpdatesEvent, this.handleUserSubscription.bind(this)); this.subscriptions.handle(lib.RoleUpdatesEvent, this.handleRoleSubscription.bind(this)); this.subscriptions.handle(lib.NoteUpdatesEvent, this.handleNoteSubscription.bind(this)); // Handle updates for datastores this.systems.on("update", this.systemsUpdated.bind(this)); this.hosts.on("update", this.hostsUpdated.bind(this)); this.hostPairing.records.on("update", this.hostPairingRequestsUpdated.bind(this)); this.instances.records.on("update", this.instanceUpdated.bind(this)); this.saves.on("update", this.savesUpdated.bind(this)); this.modPacks.on("update", this.modPacksUpdated.bind(this)); this.roles.on("update", this.rolesUpdated.bind(this)); this.users.records.on("update", this.usersUpdated.bind(this)); this.notes.on("update", this.notesUpdated.bind(this)); } get authSecret() { // Auth secret is never null after startup return Buffer.from(this.config.get("controller.auth_secret")!, "base64"); } async start(args: ControllerArgs) { if (this._state !== "new") { throw new Error(`Cannot start in state ${this._state}`); } this._state = "starting"; try { await this._startInternal(args); } catch (err) { await this._stopInternal(); this._state = "stopped"; this._events.emit("stop"); throw err; } if (this._shouldStop) { this.stop(); } } /** * Get the total size of the logs stored * * @returns {Promise} size in bytes of stored log files. */ async logSize() { return (await Promise.all([ lib.directorySize(path.join(this.logDirectory, "cluster")), lib.directorySize(path.join(this.logDirectory, "controller")), lib.directorySize(path.join(this.logDirectory, "host")), ])).reduce((a, v) => a + v, 0); } async _startInternal(args: ControllerArgs) { this.logDirectory = args.logDirectory; this.clusterLogIndex = await lib.LogIndex.load(path.join(this.logDirectory, "cluster")); this.clusterLogBuildInterval = setInterval(() => { if (this.clusterLogIndex) { this.clusterLogIndex.buildIndex().catch( err => logger.error(`Error building cluster log index:\n${err.stack}`) ); } }, 600e3); logSizeGauge.callback = async () => { logSizeGauge.set(await this.logSize()); }; // Start Vite development server if enabled if (args.dev || args.devPlugin) { await this._startDevServer(args); } this.config.on("fieldChanged", (field, curr, prev) => { if (field === "controller.autosave_interval") { this.onAutosaveIntervalChanged(); } else if (field === "controller.system_metrics_interval") { this.onSystemMetricsIntervalChanged(); } else if (field === "controller.trusted_proxies") { this.trustedProxies = this.parseTrustedProxies(); } else if (field === "controller.static_url") { this.app.locals.staticRoot = this.config.get("controller.static_url"); } this.hooks.controllerConfigFieldChanged.invoke(field, curr, prev); }); // Make sure we're actually going to listen on a port let httpPort = this.config.get("controller.http_port"); let httpsPort = this.config.get("controller.https_port"); let bindAddress = this.config.get("controller.bind_address") || ""; if (!httpPort && !httpsPort) { logger.fatal("Error: at least one of http_port and https_port must be configured"); process.exitCode = 1; return; } let tls_cert = this.config.get("controller.tls_certificate"); let tls_key = this.config.get("controller.tls_private_key"); if (httpsPort && (!tls_cert || !tls_key)) { throw new lib.StartupError( "tls_certificate and tls_private_key must be configure in order to use https_port" ); } Controller.addAppRoutes(this.app, this.pluginInfos); if (this.devServer) { this.app.use(this.devServer.middlewares); } if (!this.devServer) { let manifestPath = path.join(import.meta.dirname, "..", "..", "web", "manifest.json"); let manifest = await Controller.loadJsonObject(manifestPath); if (!manifest.main) { logger.error("Missing main entry in dist/web/manifest.json"); } this.app.locals.webBundle = { main: manifest.main || "no_web_build", imports: manifest.imports || {}, }; } this.app.locals.staticRoot = this.config.get("controller.static_url"); // Load plugins await this.loadPlugins(); // Log all express errors from middleware or routes (app.use is missing the type overload) this.app.use((err: Error, req: express.Request, res: express.Response, next: express.NextFunction): void => { const url = lib.redactUrl(req.originalUrl); logger.log({ level: "error", message: "Error while handling HTTP" + `${req.httpVersion} ${req.method} ${url}:\n${err.stack}`, meta: { method: req.method, url, httpVersion: req.httpVersion, headers: Controller.loggableHeaders(req.headers), query: lib.redactQuery(req.query), }, }); // Create and call the final handler, this is the same method used by express // This is used instead of next(err) to stop express from logging the error to the console finalhandler(req, res, { env: this.app.get("env") })(err); }); this.wsServer = new WsServer(this); // Only start listening for connections after all plugins have loaded if (httpPort) { this.httpServer = http.createServer(this.app); this.httpServerCloser = new HttpCloser(this.httpServer); await this.listen(this.httpServer, httpPort, bindAddress); logger.info(`Listening for HTTP on port ${(this.httpServer.address() as AddressInfo).port}`); } if (httpsPort && tls_cert && tls_key) { let certificate, privateKey; try { certificate = await fs.readFile(tls_cert); privateKey = await fs.readFile(tls_key); } catch (err: any) { throw new lib.StartupError( `Error loading ssl certificate: ${err.message}` ); } this.httpsServer = https.createServer({ key: privateKey, cert: certificate, }, this.app); this.httpsServerCloser = new HttpCloser(this.httpsServer); await this.listen(this.httpsServer, httpsPort, bindAddress); logger.info(`Listening for HTTPS on port ${(this.httpsServer.address() as AddressInfo).port}`); } this.hostPairing.start(); this.onAutosaveIntervalChanged(); this.onSystemMetricsIntervalChanged(); logger.info("Started controller"); this.startedAtMs = Date.now(); this._state = "running"; } async _startDevServer(args: ControllerArgs) { logger.warn("Vite development mode enabled"); const vite = await import("vite"); const webUiVite = await import("@clusterio/web_ui/vite.mjs"); const devPlugins = new Map(); const pluginDirs = new Map(); for (let name of (args.devPlugin ?? []).map(String)) { let info = this.pluginInfos.find(i => i.name === name); if (!info) { throw new lib.StartupError(`No plugin named ${name}`); } pluginDirs.set(name, path.dirname(info.packagePath)); devPlugins.set(name, webUiVite.devPluginUrl(name)); } this.app.locals.devPlugins = devPlugins; const controllerDir = path.join(import.meta.dirname, "..", "..", ".."); this.devServer = await vite.createServer({ configFile: path.join(controllerDir, "vite.config.js"), root: controllerDir, base: "/static/", appType: "custom", server: { middlewareMode: true, preTransformRequests: false, }, plugins: [webUiVite.clusterioDevPlugins(pluginDirs)], }); // Plugins served from source import the shared modules from the development server, the web // interface has to use the same copies, so it is served from source as well with --dev-plugin. this.app.locals.webBundle = { main: this.devServer.config.plugins.find(p => p.name === "clusterio:app")!.api.entry, imports: Object.fromEntries(webUiVite.sharedModules.map( name => [name, webUiVite.devSharedUrl(name)] )), }; this.app.locals.devServer = this.devServer; } /** * Stops the controller. * * Save data and bring down the active connections. This is the reverse * of start and is valid at any point in time. */ async stop() { if (this._state === "starting") { this._shouldStop = true; await events.once(this._events, "stop"); return; } else if (this._state === "running") { await this._stopInternal(); } this._state = "stopped"; this._events.emit("stop"); } async _stopInternal() { // This function should never throw. this._state = "stopping"; if (this.shouldRestart) { logger.info("Restarting controller"); process.exitCode = 1; } else { logger.info("Stopping controller"); } if (this.clusterLogBuildInterval) { clearInterval(this.clusterLogBuildInterval); } if (this.systemMetricsInterval) { clearInterval(this.systemMetricsInterval); this.systemMetricsInterval = undefined; } if (this.autosaveInterval) { clearInterval(this.autosaveInterval); this.autosaveInterval = undefined; } this.hostPairing.stop(); if (this.clusterLogIndex) { await this.clusterLogIndex.save(); } if (this.devServer) { await this.devServer.close(); } await this.hooks.shutdown.invoke(); await this.wsServer.stop(); let stopTasks = []; logger.info("Stopping HTTP(S) server"); if (this.httpServer && this.httpServer.listening && this.httpServerCloser) { stopTasks.push(this.httpServerCloser.close()); } if (this.httpsServer && this.httpsServer.listening && this.httpsServerCloser) { stopTasks.push(this.httpsServerCloser.close()); } await Promise.all(stopTasks); logger.info("Saving data"); await this.saveData(); logger.info("Goodbye"); } parseTrustedProxies() { const trustedProxies = new BlockList(); const proxiesString = this.config.get("controller.trusted_proxies"); if (proxiesString) { const proxies = proxiesString.split(",").map(s => s.trim()); for (const proxy of proxies) { const [ip, prefix] = proxy.split("/"); // eslint-disable-next-line no-nested-ternary const type = isIPv4(ip) ? "ipv4" : isIPv6(ip) ? "ipv6" : undefined; if (!type) { logger.error(`Invalid proxy '${proxy}': not an IP address`); continue; } try { if (prefix) { trustedProxies.addSubnet(ip, Number.parseInt(prefix, 10), type); } else { trustedProxies.addAddress(ip, type); } } catch (err: any) { logger.error(`Invalid proxy '${proxy}': ${err.message}`); } } } return trustedProxies; } onSystemMetricsIntervalChanged() { if (this.systemMetricsInterval) { clearInterval(this.systemMetricsInterval); this.systemMetricsInterval = undefined; } const systemMetricsIntervalSeconds = this.config.get("controller.system_metrics_interval"); if (systemMetricsIntervalSeconds > 0) { this.systemMetricsInterval = setInterval( this.updateSystems.bind(this), systemMetricsIntervalSeconds * 1000 ); } } async updateSystems() { try { const requests: Promise[] = []; for (let hostConnection of this.wsServer.hostConnections.values()) { if (!hostConnection.connected) { continue; } requests.push(hostConnection.send(new lib.SystemInfoRequest())); } const restartRequired = await this.checkRestartRequired(); requests.push(lib.gatherSystemInfo( "controller", this.canRestart, restartRequired, this.startedAtMs, await this.listPlugins(), )); const newMetrics = await Promise.all(requests); for (const metric of newMetrics) { this.systems.set(metric); } } catch (err: any) { logger.error(`Unexpected error updating system infos:\n${err.stack ?? err.message}`); } } /** Plugins on the controller, including those installed or updated since it started */ async listPlugins() { return await lib.listPluginDetails( this.pluginInfos, pluginInfo => this.loadedPlugins.has(pluginInfo), pluginInfo => this.config.get(`${pluginInfo.name}.load_plugin`), ); } async checkRestartRequired() { if (this.config.restartRequired) { return true; } let packageName = "@clusterio/controller"; try { // First check the clusterio version const runningVersion = this.config.get("controller.version"); const packageJsonPath = fileURLToPath( import.meta.resolve("@clusterio/controller/package.json"), ); const packageJson = JSON.parse(await fs.readFile(packageJsonPath, "utf8")); if (runningVersion !== packageJson.version) { this.config.restartRequired = true; return true; } // Second check plugin versions for (const pluginInfo of this.pluginInfos) { if ( !pluginInfo.controllerEntrypoint && !pluginInfo.webEntrypoint || !this.config.get(`${pluginInfo.name}.load_plugin`) ) { continue; } packageName = pluginInfo.npmPackage ?? pluginInfo.name; const pluginPackageJson = JSON.parse(await fs.readFile(pluginInfo.packagePath, "utf8")); if (pluginInfo.version !== pluginPackageJson.version) { this.config.restartRequired = true; return true; } } } catch (err: any) { logger.warn(`Failed to read package.json for ${packageName}:\n${err.stack ?? err.message}`); } return false; } async checkRestartDowngrade() { try { const runningVersion = this.config.get("controller.version"); const packageJsonPath = fileURLToPath(import.meta.resolve("@clusterio/controller/package.json")); const packageJson = JSON.parse(await fs.readFile(packageJsonPath, "utf8")); const installedVersion = packageJson.version; if (!semver.valid(runningVersion) || !semver.valid(installedVersion)) { logger.warn( `Unable to compare running controller version ${runningVersion} ` + `with installed version ${installedVersion}.` ); return null; } if (semver.lt(installedVersion, runningVersion)) { return { installedVersion, runningVersion }; } } catch (err: any) { logger.warn(`Failed to check controller version before restart:\n${err.stack ?? err.message}`); } return null; } onAutosaveIntervalChanged() { if (this.autosaveInterval) { clearInterval(this.autosaveInterval); this.autosaveInterval = undefined; } if (this.recoveryMode) { logger.warn("Recovery | autosaving disabled"); return; } const autosaveIntervalSeconds = this.config.get("controller.autosave_interval"); if (autosaveIntervalSeconds > 0) { this.autosaveInterval = setInterval(this.saveData.bind(this), autosaveIntervalSeconds * 1000); } } /** * Save all data currently in memory to disk * * Note: If a save is currently in progress this will wait until that * completes and start another save. */ async saveData() { await this._saveDataAsyncSerial.invoke(); } private _saveDataAsyncSerial = new lib.AsyncSerialMergingCallback(async () => { try { await this._saveDataInternal(); } catch (err: any) { logger.error(`Unexpected error during saveData:\n${err.stack}`); } }); private async _saveDataInternal() { await Promise.all([ this.config.save(), this.systems.save(), this.hosts.save(), this.instances.records.save(), this.saves.save(), this.modPacks.save(), this.roles.save(), this.users.records.save(), this.hooks.save.invoke(), this.notes.save(), ]); } static migrateSystems(rawJson: unknown[]): Static[] { const serialized = rawJson as Static[]; return serialized.map(json => { if (!json.canRestart) { // Added in 2.0.0.alpha.17 json.canRestart = false; } if (!json.restartRequired) { // Added in 2.0.0.alpha.21 json.restartRequired = false; } if (json.systemStartedAtMs === undefined) { // Added in 2.0.0.alpha.23 json.systemStartedAtMs = Date.now(); } if (json.processStartedAtMs === undefined) { // Added in 2.0.0.alpha.23 json.processStartedAtMs = Date.now(); } return json; }); } static migrateHosts(rawJson: unknown[]): Static[] { let serialized = rawJson as any; // New format 2.0.0.alpha.19 if (serialized.length && serialized[0] instanceof Array) { serialized = serialized.map((e: any) => e[1]); } return serialized; } static finaliseHosts(host: HostRecord): HostRecord { if (host.connected) { host.connected = false; host.updatedAtMs = Date.now(); } return host; } static migrateInstances(rawJson: unknown[]): Static[] { const serialized = rawJson as Static[]; return serialized.map(json => { if (!json.config) { // New format 2.0.0.alpha.14 return { config: json as any, status: "running" }; // Use running to force updatedAtMs } return json; }); } static finaliseInstances(instance: InstanceRecord): InstanceRecord { const status = instance.config.get("instance.assigned_host") === null ? "unassigned" : "unknown"; if (instance.status !== status) { instance.status = status; instance.updatedAtMs = Date.now(); } return instance; } static migrateModPacks(rawJson: unknown[]): Static[] { const serialized = rawJson as Static[]; return serialized.map(json => { for (const mod of json.mods) { // migrate: 2.0.0-alpha.22 - old buildins used X.Y rather than X.Y.Z // All other versions are kept verbatim so they still match the mod's file name. if (lib.isPartialVersion(mod.version)) { mod.version = lib.normaliseFullVersion(mod.version) as lib.SourceVersion; } } return json; }); } /** Remove export manifests that reference files missing from staticDirectory */ static async validateExports(modPacks: lib.SubscribableDatastore, staticDirectory: string) { const invalid: lib.ModPack[] = []; for (const modPack of modPacks.valuesMutable()) { if (!modPack.exportManifest) { continue; } const missing: string[] = []; for (const file of Object.values(modPack.exportManifest.assets)) { const valid = path.basename(file) === file && await fs.stat(path.join(staticDirectory, file)).then(s => s.isFile(), () => false); if (!valid) { missing.push(file); } } if (missing.length) { Controller.dropExport(modPack, `missing ${missing.join(", ")}`); invalid.push(modPack); } } if (invalid.length) { modPacks.setMany(invalid); } } /** Clear the export manifest of a mutable mod pack, caller must save it */ static dropExport(modPack: lib.ModPack, reason: string) { logger.warn(`Removing export data of mod pack ${modPack.name} (id ${modPack.id}): ${reason}`); modPack.exportManifest = undefined; } static migrateRoles(rawJson: unknown[]): Static[] { const serialized = rawJson as Static[]; // migrate: host token permissions were removed with host pairing in 2.1 const removedPermissions = new Set([ "core.host.generate_token", "core.host.revoke_token", "core.host.create_config", ]); return serialized.map(json => { json.permissions = json.permissions.map(permission => { // migrate: core.instance.save.list.subscribe was renamed in alpha 17 if (permission === "core.instance.save.list.subscribe") { return "core.instance.save.subscribe"; } return permission; }).filter(permission => !removedPermissions.has(permission)); return lib.Role.pluginMigrations.reduce(json); }); } static finaliseUsers(user: UserRecord, users: Map): UserRecord { const existingUser = users.get(user.id); if (existingUser) { // All lowercase ids in 2.0.0.alpha.19 if (user.playerStats.onlineTimeMs <= existingUser.playerStats.onlineTimeMs) { // We assume the user with the lower online time is the duplicate existingUser.merge(user); return existingUser; } user.merge(existingUser); } return user; } static async loadJsonObject(filePath: string, throwOnMissing: boolean = false): Promise { let manifest = {}; try { manifest = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" })); } catch (err: any) { if (throwOnMissing || err.code !== "ENOENT") { throw err; } } return manifest; } /** * Query controller log * * @param filter - * Filter to limit entries with. Note that only the controller log can * be queried from this function. * @returns log entries matching the filter */ async queryControllerLog(filter: lib.QueryLogFilter): Promise { return lib.queryLog( path.join(this.logDirectory, "controller"), filter, null, ); } /** * Query cluster log * * @param filter - * Filter to limit entries with. * @returns log entries matching the filter */ async queryClusterLog(filter: lib.QueryLogFilter): Promise { return await lib.queryLog( path.join(this.logDirectory, "cluster"), filter, this.clusterLogIndex, ); } static addAppRoutes(app: Application, pluginInfos: lib.PluginNodeEnvInfo[]) { app.use((req: Request, res: Response, next) => { const startNs = process.hrtime.bigint(); stream.finished(res, () => { let routePath = "static"; if (req.route && req.route.path) { routePath = req.route.path; } const endNs = process.hrtime.bigint(); const durationMs = (Number(endNs - startNs) / 1e6); endpointDurationSummary.labels(routePath).observe(durationMs / 1e3); const url = lib.redactUrl(req.originalUrl); logger.log({ level: "http", message: `HTTP${req.httpVersion} ${req.method} ${url}` + ` (Status ${res.statusCode} in ${durationMs}ms)`, meta: { method: req.method, url, statusCode: res.statusCode, responseTime: durationMs, httpVersion: req.httpVersion, headers: Controller.loggableHeaders(req.headers), query: lib.redactQuery(req.query), }, }); }); next(); }); app.use(compression()); // Set folder to serve static content from (the website) const staticOptions = { immutable: true, maxAge: 1000 * 86400 * 365 }; app.use("/static", express.static(path.join(import.meta.dirname, "..", "..", "web", "static"), staticOptions) ); app.use("/static", express.static("static", staticOptions)); // Used for data export files app.use("/static", routes.missingExportAsset); // Add API routes routes.addRouteHandlers(app); // Add routes for the web interface for (let route of routes.webRoutes) { app.get(route, Controller.serveWeb(route)); } for (let pluginInfo of pluginInfos) { for (let route of pluginInfo.routes || []) { app.get(route, Controller.serveWeb(route)); } app.use("/static", express.static(pluginInfo.webStaticPath, staticOptions)); } } systemsUpdated(systems: lib.SystemInfo[]) { this.subscriptions.broadcast(new lib.SystemInfoUpdateEvent(systems)); } async handleSystemInfoSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const systems = [...this.systems.values()].filter( metric => metric.updatedAtMs > request.lastRequestTimeMs, ); return systems.length ? new lib.SystemInfoUpdateEvent(systems) : null; } /** * Generate access token for host * * @returns access token for host * @param hostId - ID of host to generate a token for. */ generateHostToken(hostId: number): string { return jwt.sign({ aud: "host", host: hostId }, this.authSecret); } /** * Check a host token * * @param token - Host token to check. * @returns ID of the host the token was issued to. * @throws if the token is invalid, its host is not registered or it was issued before tokenValidAfter. */ verifyHostToken(token: string): number { // migrate: allow pre-rename tokens issued to hosts before alpha-14 const payload = jwt.verify(token, this.authSecret, { audience: ["host", "slave"] }); if (typeof payload === "string") { throw new Error("unexpected JsonWebToken type"); } const hostId = payload.host !== undefined ? payload.host : payload.slave; const host = this.hosts.get(hostId); if (!host) { throw new Error("host is not registered, start the host again to request to join the cluster"); } if (!payload.iat || payload.iat < host.tokenValidAfter) { throw new Error("invalid token"); } return hostId; } hostsUpdated(hosts: HostRecord[]) { let updates = hosts.map(host => host.toHostDetails()); this.subscriptions.broadcast(new lib.HostUpdatesEvent(updates)); } hostPairingRequestsUpdated(requests: lib.HostPairingDetails[]) { this.subscriptions.broadcast(new lib.HostPairingUpdatesEvent(requests)); } async handleHostPairingSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const requests = [...this.hostPairing.records.values()].filter( pairing => pairing.updatedAtMs > request.lastRequestTimeMs, ); return requests.length ? new lib.HostPairingUpdatesEvent(requests) : null; } async handleHostSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const hosts = [...this.hosts.values()].filter( host => host.updatedAtMs > request.lastRequestTimeMs, ).map(host => host.toHostDetails()); return hosts.length ? new lib.HostUpdatesEvent(hosts) : null; } /** * Removes all saves currently stored for the given instance, if any * @param instanceId - Id of Instance to clear saves for. * @internal */ clearSavesOfInstance(instanceId: number) { this.saves.deleteMany([...this.saves.values()].filter(save => save.instanceId === instanceId)); } /** * Notify the config of an instance updated * * Used to push config changes to the assigned host when the config of * an instance has changed. * * @param instance - * Instance to nodify the config updated. */ async instanceConfigUpdated(instance: InstanceRecord) { let hostId = instance.config.get("instance.assigned_host"); if (hostId !== null) { let connection = this.wsServer.hostConnections.get(hostId); if (connection) { await connection.send( new lib.InstanceAssignInternalRequest(instance.id, instance.config.toRemote("host")) ); } } } instanceUpdated(instances: InstanceRecord[]) { for (const instance of instances) { if (instance.status === "stopped") { this.subscriptions.unsubscribeAddress(lib.Address.fromShorthand({instanceId: instance.id})); } } const updates = instances.map(instance => instance.toInstanceDetails()); this.subscriptions.broadcast(new lib.InstanceDetailsUpdatesEvent(updates)); } async handleInstanceDetailsSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const instances = [...this.instances.values()].filter( instance => instance.updatedAtMs > request.lastRequestTimeMs, ).map(instance => instance.toInstanceDetails()); return instances.length ? new lib.InstanceDetailsUpdatesEvent(instances) : null; } savesUpdated(saves: lib.SaveDetails[]) { this.subscriptions.broadcast(new lib.InstanceSaveDetailsUpdatesEvent(saves)); } async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const saves = [...this.saves.values()].filter( save => save.updatedAtMs > request.lastRequestTimeMs ); return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null; } modPacksUpdated(modPacks: lib.ModPack[]) { this.subscriptions.broadcast(new lib.ModPackUpdatesEvent(modPacks)); this.hooks.modPacksUpdated.invoke(modPacks); } async handleModPackSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const modPacks = [...this.modPacks.values()].filter( modPack => modPack.updatedAtMs > request.lastRequestTimeMs, ); return modPacks.length ? new lib.ModPackUpdatesEvent(modPacks) : null; } modsUpdated(mods: lib.ModInfo[]) { // ModStore sets updatedAtMs for mods this.subscriptions.broadcast(new lib.ModUpdatesEvent(mods)); this.hooks.modsUpdated.invoke(mods); } async handleModSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const mods = [...this.modStore.files.values()].filter( mod => mod.updatedAtMs > request.lastRequestTimeMs, ); return mods.length ? new lib.ModUpdatesEvent(mods) : null; } usersUpdated(users: UserRecord[]) { this.subscriptions.broadcast(new lib.UserUpdatesEvent(users)); } async handleUserSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const users = [...this.users.records.values()].filter( user => user.updatedAtMs > request.lastRequestTimeMs, ); return users.length ? new lib.UserUpdatesEvent(users) : null; } /** * Notify connected control clients under the given user that the * permissions for this user may have changed. * @param user - User permisions updated for. */ userPermissionsUpdated(user: Readonly) { for (let controlConnection of this.wsServer.controlConnections.values()) { if (controlConnection.user === user) { this._controlPermissionsUpdated(controlConnection); } } } _controlPermissionsUpdated(controlConnection: ControlConnection) { const user = controlConnection.user; this.subscriptions.checkLinkPermissions(controlConnection, user); controlConnection.checkLogPermission(); controlConnection.send( new lib.AccountUpdateEvent([...user.roles].map(r => ({ name: r.name, id: r.id, permissions: [...r.permissions], }))) ); } rolesUpdated(roles: lib.Role[]) { this.subscriptions.broadcast(new lib.RoleUpdatesEvent(roles)); this.hooks.rolesUpdated.invoke(roles); // Notify connected control clients with the given role that the permissions may have changed. for (const role of roles) { for (let controlConnection of this.wsServer.controlConnections.values()) { if (controlConnection.user.roles.has(role)) { this._controlPermissionsUpdated(controlConnection); } } } } async handleRoleSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } const roles = [...this.roles.values()].filter( role => role.updatedAtMs > request.lastRequestTimeMs, ); return roles.length ? new lib.RoleUpdatesEvent(roles) : null; } notesUpdated(notes: lib.Note[]) { this.subscriptions.broadcast(new lib.NoteUpdatesEvent(notes)); } async handleNoteSubscription(request: lib.SubscriptionRequest) { if (request.action === "unsubscribe") { return null; } // Deleted notes are kept as tombstones so resubscribing clients learn of the deletion const notes = [...this.notes.values()].filter( note => note.updatedAtMs > request.lastRequestTimeMs && (request.lastRequestTimeMs > 0 || !note.isDeleted), ); return notes.length ? new lib.NoteUpdatesEvent(notes) : null; } /** * Set or remove the note attached to a resource * @param resourceType - Type of resource the note belongs to. * @param resourceId - Id of the resource within its type. * @param content - New text of the note, empty removes the note. * @param updatedBy - Name of the user making the change. */ setNote(resourceType: string, resourceId: string, content: string, updatedBy = "") { const existing = this.notes.getMutable(lib.Note.idFor(resourceType, resourceId)); if (content === "") { this.deleteNote(resourceType, resourceId, updatedBy); return; } if (existing) { existing.content = content; existing.updatedBy = updatedBy; existing.isDeleted = false; this.notes.set(existing); } else { this.notes.set(new lib.Note(resourceType, resourceId, content, updatedBy)); } } /** * Remove the note attached to a resource, if any * @param resourceType - Type of resource the note belongs to. * @param resourceId - Id of the resource within its type. * @param updatedBy - Name of the user making the change. */ deleteNote(resourceType: string, resourceId: string | number, updatedBy = "") { const note = this.notes.getMutable(lib.Note.idFor(resourceType, resourceId)); if (note && !note.isDeleted) { note.content = ""; note.updatedBy = updatedBy; note.isDeleted = true; this.notes.set(note); } } async loadPlugins() { const context = { metrics, logger, controller: this }; for (let pluginInfo of this.pluginInfos) { try { const manifestPath = path.posix.join(pluginInfo.requirePath, "dist", "web", "manifest.json"); const resolvedUrl = import.meta.resolve(manifestPath); const resolvedPath = resolvedUrl.startsWith("file://") ? fileURLToPath(resolvedUrl) : resolvedUrl; pluginInfo.manifest = await Controller.loadJsonObject(resolvedPath, true); } catch (err) { if (lib.pluginNeedsWebBuild(pluginInfo)) { const stack = (err as Error).stack ?? (err as Error).message; logger.warn(`Unable to load dist/web/manifest.json for plugin ${pluginInfo.name}\n${stack}`); } } if (!this.config.get(`${pluginInfo.name}.load_plugin`)) { continue; } if (this.recoveryMode) { logger.warn(`Recovery | force disabled plugin ${pluginInfo.name}`); continue; } try { await lib.loadPlugin( pluginInfo, "controller", context, "ControllerPlugin", BaseControllerPlugin, ); this.loadedPlugins.add(pluginInfo); } catch (err: any) { if (err.code === "InstallationError") { throw err; } throw new lib.PluginError(pluginInfo.name, err); } logger.info(`Loaded plugin ${pluginInfo.name}`); } } /** * Calls listen on server capturing any errors that occurs * binding to the port. Also adds handler for WebSocket * upgrade event. * * @param {module:net.Server} server - Server to start the listening on. * @param {*} args - Arguments to the .listen() call on the server. * @returns {Promise} promise that resolves the server is listening. */ listen(server: http.Server|https.Server, ...args: any[]): Promise { return new Promise((resolve, reject) => { server.on("upgrade", (req, socket, head) => { logger.verbose("handling WebSocket upgrade"); this.wsServer.handleUpgrade(req, socket, head); }); function wrapError(err: any) { reject(new lib.StartupError( `Server listening failed: ${err.message}` )); } server.once("error", wrapError); server.listen(...args, () => { server.off("error", wrapError); resolve(); }); }); } /** * Returns the URL needed to connect to the controller. * * @returns controller URL. */ getControllerUrl(): string { return Controller.calculateControllerUrl(this.config); } static calculateControllerUrl(config: lib.ControllerConfig) { let url = config.get("controller.public_url"); if (!url) { if (config.get("controller.https_port")) { url = `https://localhost:${config.get("controller.https_port")}/`; } else { url = `http://localhost:${config.get("controller.http_port")}/`; } } return url; } /** * Copy of request headers with credentials masked for logging * * @param headers - headers of the incoming request. * @returns headers safe to write to the log. */ static loggableHeaders(headers: http.IncomingHttpHeaders) { const result = { ...headers }; for (const name of ["authorization", "cookie", "proxy-authorization", "x-access-token"]) { if (result[name] !== undefined) { result[name] = "[redacted]"; } } return result; } /** * Serves the web interface with the root path set appropriately * * @param route - route the interface is served under. * @returns Experess.js route handler. */ static serveWeb(route: string) { // The depth is is the number of slashes in the route minus one, but due // to lenient matching on trailing slashes in the route we need to // compensate if the request path contains a slash but not the route, // and vice versa. let routeDepth = (route.match(/\//g) || []).length - 1 - Number(route.slice(-1) === "/"); return function(req: Request, res: Response, next: NextFunction) { let depth = routeDepth + Number(req.path.slice(-1) === "/"); let webRoot = "../".repeat(depth) || "./"; let staticRoot = res.app.locals.staticRoot ?? `${webRoot}static/`; const { main, imports } = res.app.locals.webBundle; const importMap = JSON.stringify({ imports: Object.fromEntries(Object.entries(imports).map( ([name, file]) => [name, `${staticRoot}${routes.stripStaticPrefix(file)}`] )), }).replace(/ { content = content .replace(/__CLUSTER_NAME__/g, res.app.locals.controller.config.get("controller.name")) .replace(/__WEB_ROOT__/g, webRoot) .replace(/__STATIC_ROOT__/g, staticRoot) .replace(/__IMPORT_MAP__/g, () => importMap) .replace(/__MAIN_BUNDLE__/g, () => routes.stripStaticPrefix(main)) ; if (res.app.locals.devServer) { content = await res.app.locals.devServer.transformIndexHtml(req.originalUrl, content); } res.type("text/html"); res.send(content); }).catch(err => { next(err); }); }; } handle(Class: any, handler: any) { if (Class.type === "request") { this.handleRequest(Class, handler); } else if (Class.type === "event") { this.handleEvent(Class, handler); } else { throw new Error(`Class ${Class.name} has unrecognized type ${Class.type}`); } } handleRequest(Request: any, handler: any) { if (!lib.Link._requestsByClass.has(Request)) { throw new Error(`Unregistered Request class ${Request.name}`); } if (this._registeredRequests.has(Request)) { throw new Error(`Request ${Request.name} is already registered`); } this._registeredRequests.set(Request, handler); } fallbackRequest(Request: any, handler: any) { if (!lib.Link._requestsByClass.has(Request)) { throw new Error(`Unregistered Request class ${Request.name}`); } if (this._fallbackedRequests.has(Request)) { throw new Error(`Request ${Request.name} is already fallbacked`); } this._fallbackedRequests.set(Request, handler); } handleEvent(Event: any, handler: any) { if (!lib.Link._eventsByClass.has(Event)) { throw new Error(`Unregistered Event class ${Event.name}`); } if (this._registeredEvents.has(Event)) { throw new Error(`Event ${Event.name} is already registered`); } this._registeredEvents.set(Event, handler); } snoopEvent(Event: any, handler: any) { if (!lib.Link._eventsByClass.has(Event)) { throw new Error(`Unregistered Event class ${Event.name}`); } if (this._snoopedEvents.has(Event)) { throw new Error(`Event ${Event.name} is already snooped`); } this._snoopedEvents.set(Event, handler); } /** * Send request or event * * Routes the given request or event to the given destination. The * destination argument supports address shorthands, see {@link * module:lib.Address.fromShorthand} * * @param address - Where to send it. * @param requestOrEvent - The request or event to send. * @returns * Promise that resolves to the response if a request was sent or * undefined if it was an event. */ sendTo(address: lib.AddressShorthand, requestOrEvent: any): Promise | void { let dst = lib.Address.fromShorthand(address); if (requestOrEvent.constructor.type === "request") { return this.sendRequest(requestOrEvent, dst); } if (requestOrEvent.constructor.type === "event") { return this.sendEvent(requestOrEvent, dst); } throw Error(`Unknown type ${requestOrEvent.constructor.type}.`); } sendRequest(request: any, dst: lib.Address) { let connection; if (dst.type === lib.Address.controller) { throw new Error(`Message would return back to sender ${dst}.`); } else if (dst.type === lib.Address.control) { connection = this.wsServer.controlConnections.get(dst.id); if (!connection) { throw new lib.NotFoundError("Target control connection does not exist."); } } else if (dst.type === lib.Address.instance) { const instance = this.instances.getForRequest(dst.id); let hostId = instance.config.get("instance.assigned_host"); if (hostId === null) { throw new lib.ConflictError("Instance is not assigned to a host"); } connection = this.wsServer.hostConnections.get(hostId); if (!connection) { throw new lib.UnavailableError("Host containing instance is not connected"); } } else if (dst.type === lib.Address.host) { connection = this.wsServer.hostConnections.get(dst.id); if (!connection) { throw new lib.UnavailableError("Host is not connected"); } } else { throw new Error(`Unknown address type ${dst.type}`); } return connection.sendRequest(request, dst); } sendEvent(event: lib.Event, dst: lib.Address) { let connection; if (dst.type === lib.Address.controller) { throw new Error(`Message would return back to sender ${dst}.`); } else if (dst.type === lib.Address.control) { connection = this.wsServer.controlConnections.get(dst.id); if (!connection) { throw new Error("Target control connection does not exist."); } } else if (dst.type === lib.Address.instance) { const instance = this.instances.get(dst.id); if (!instance) { throw new Error(`Instance with ID ${dst.id} does not exist`); } let hostId = instance.config.get("instance.assigned_host"); if (hostId === null) { throw new Error("Instance is not assigned to a host"); } connection = this.wsServer.hostConnections.get(hostId); if (!connection) { throw new Error("Host containing instance is not connected"); } } else if (dst.type === lib.Address.host) { connection = this.wsServer.hostConnections.get(dst.id); if (!connection) { throw new Error("Host is not connected"); } } else if (dst.type === lib.Address.broadcast) { if (dst.id === lib.Address.control) { for (let controlConnection of this.wsServer.controlConnections.values()) { controlConnection.sendEvent(event, dst); } } else if (dst.id === lib.Address.instance || dst.id === lib.Address.host) { const plugin = event.constructor.plugin; for (let hostConnection of this.wsServer.hostConnections.values()) { if (plugin && !hostConnection.plugins.has(plugin)) { continue; } hostConnection.sendEvent(event, dst); } } else { throw new Error(`Unexpected broadcast target ${dst.id}`); } return; } else { throw new Error(`Unknown address type ${dst.type}`); } connection.sendEvent(event, dst); } async sendToHostByInstanceId(request: Req & lib.Request): Promise; sendToHostByInstanceId(event: T & lib.Event): void; async sendToHostByInstanceId(requestOrEvent: lib.Request | lib.Event) { if (requestOrEvent.constructor.type === "request") { return this.sendRequestToHostByInstanceId(requestOrEvent as any); } if (requestOrEvent.constructor.type === "event") { return this.sendEventToHostByInstanceId(requestOrEvent as any); } throw Error(`Unknown type ${(requestOrEvent.constructor as any).type}.`); } async sendRequestToHostByInstanceId(request: Req & lib.Request) { let instance = this.instances.getForRequest(request.instanceId); let hostId = instance.config.get("instance.assigned_host"); if (hostId === null) { throw new lib.ConflictError("Instance is not assigned to a host"); } let connection = this.wsServer.hostConnections.get(hostId); if (!connection) { throw new lib.UnavailableError("Host containing instance is not connected"); } return await connection.send(request); } sendEventToHostByInstanceId(event: T & lib.Event) { let instance = this.instances.getForRequest(event.instanceId); let hostId = instance.config.get("instance.assigned_host"); if (hostId === null) { return; } let connection = this.wsServer.hostConnections.get(hostId); if (!connection) { return; } connection.send(event); } }