factorio-clusterio/packages/controller/src/Controller.ts

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<lib.PluginNodeEnvInfo> = 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<ControllerDebugEvents>();
	private _events = new events.EventEmitter<ControllerEvents>();

	/** 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.RequestClass<unknown, unknown>, lib.RequestHandler<unknown, unknown>> = new Map();
	_registeredRequests: Map<lib.RequestClass<unknown, unknown>, lib.RequestHandler<unknown, unknown>> = new Map();
	_registeredEvents = new Map<lib.EventClass<unknown>, lib.EventHandler<unknown>>();
	_snoopedEvents = new Map();

	devServer: import("vite").ViteDevServer | null = null;

	startedAtMs = Date.now();
	autosaveInterval?: ReturnType<typeof setInterval>;
	systemMetricsInterval?: ReturnType<typeof setInterval>;

	logDirectory: string = "";
	clusterLogIndex: lib.LogIndex | null = null;
	clusterLogBuildInterval: ReturnType<typeof setInterval> | null = null;

	// Cache for mod portal requests
	modPortalCache = new Map<string, { timestamp: number, data: any[] }>();
	// 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.safeOutputFile(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);

		// 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<lib.SystemInfo>(),
		/** Mapping of host id to host record */
		public hosts = new lib.SubscribableDatastore<HostRecord>(),
		/** Mapping of instance id to instance record */
		instances = new lib.SubscribableDatastore<InstanceRecord>(),
		/** Mapping of save id to save details */
		public saves = new lib.SubscribableDatastore<lib.SaveDetails>(),
		/** Mapping of mod pack id to mod pack */
		public modPacks = new lib.SubscribableDatastore<lib.ModPack>(),
		/** 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<lib.Role>(),
		/** Mapping of user id to user record */
		users = new lib.SubscribableDatastore<UserRecord>(),
		/** Mapping of note id to note attached to a resource */
		public notes = new lib.SubscribableDatastore<lib.Note>(),
	) {
		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<number>} 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<string, string>();
		const pluginDirs = new Map<string, string>();
		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<lib.SystemInfo>[] = [];
			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<typeof lib.SystemInfo.jsonSchema>[] {
		const serialized = rawJson as Static<typeof lib.SystemInfo.jsonSchema>[];
		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<typeof HostRecord.jsonSchema>[] {
		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<typeof InstanceRecord.jsonSchema>[] {
		const serialized = rawJson as Static<typeof InstanceRecord.jsonSchema>[];
		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<typeof lib.ModPack.jsonSchema>[] {
		const serialized = rawJson as Static<typeof lib.ModPack.jsonSchema>[];
		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;
		});
	}

	static migrateRoles(rawJson: unknown[]): Static<typeof lib.Role.jsonSchema>[] {
		const serialized = rawJson as Static<typeof lib.Role.jsonSchema>[];
		// 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["id"], UserRecord>): 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<any> {
		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<object[]> {
		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<object[]> {
		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

		// 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<User>) {
		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<void> {
		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(/</g, "\\u003c");

			fs.readFile(
				path.join(import.meta.dirname, "..", "..", "..", "web", "index.html"),
				"utf8",
			).then(async (content) => {
				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<any> | 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<T>(event: lib.Event<T>, 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<Req extends InstanceId, Res>(request: Req & lib.Request<Req, Res>): Promise<Res>;
	sendToHostByInstanceId<T extends InstanceId>(event: T & lib.Event<T>): void;
	async sendToHostByInstanceId(requestOrEvent: lib.Request<unknown, unknown> | lib.Event<unknown>) {
		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<Req extends InstanceId, Res>(request: Req & lib.Request<Req, Res>) {
		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<T extends { instanceId: number }>(event: T & lib.Event<T>) {
		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);
	}
}