factorio-clusterio/packages/host/src/Host.ts

import dgram from "node:dgram";
import fs from "node:fs/promises";
import path from "path";
import pidusage from "pidusage";
import semver from "semver";
import setBlocking from "set-blocking";
import stream from "stream";
import { fileURLToPath } from "node:url";
import util from "util";

// internal libraries
import * as lib from "@clusterio/lib";
import { logger } from "@clusterio/lib";

import type { HostConnector } from "../host.js";
import Instance from "./Instance.js";
import InstanceConnection from "./InstanceConnection.js";
import { BaseHostPlugin, HostHooks } from "./BaseHostPlugin.js";


function checkRequestSaveName(name: string) {
	try {
		lib.checkFilename(name);
	} catch (err: any) {
		throw new lib.ValidationError(`Save name ${err.message}`);
	}
}

async function copySaveExclusive(sourcePath: string, targetPath: string) {
	await fs.copyFile(
		sourcePath,
		targetPath,
		fs.constants.COPYFILE_EXCL | fs.constants.COPYFILE_FICLONE
	);
}

/**
 * Move a file without overwriting an existing target, fails with EEXIST
 * if the target exists.
 */
async function moveSaveExclusive(sourcePath: string, targetPath: string) {
	try {
		await fs.link(sourcePath, targetPath);
	} catch (err: any) {
		// Hard links fail across filesystems and on some filesystems with
		// varying codes, the copy is still exclusive.
		if (err.code === "EEXIST") {
			throw err;
		}
		await copySaveExclusive(sourcePath, targetPath);
	}
	try {
		await fs.unlink(sourcePath);
	} catch (err: any) {
		await fs.unlink(targetPath).catch(() => {});
		throw err;
	}
}

/**
 * Checks that no socket is bound to the given UDP port, on IPv4 or IPv6
 *
 * Factorio binds the game port on both 0.0.0.0 and [::].
 *
 * @param port - UDP port to check.
 * @returns true if the port can be bound.
 * @internal
 */
async function udpPortFree(port: number) {
	for (const type of ["udp4", "udp6"] as const) {
		const socket = dgram.createSocket({ type, ipv6Only: type === "udp6" });
		try {
			await new Promise<void>((resolve, reject) => {
				socket.once("error", reject);
				socket.bind(port, resolve);
			});
		} catch (err: any) {
			// Other errors, such as no IPv6 support on the host, don't stop Factorio from binding the port.
			if (err.code === "EADDRINUSE") {
				return false;
			}
		} finally {
			socket.close();
		}
	}
	return true;
}

/**
 * Searches for instances in the provided directory
 *
 * Looks through all sub-dirs of the provided directory for valid
 * instance definitions and return a mapping of instance id to
 * instance configs and the root instance paths.
 *
 * @param instancesDir - Directory containing instances
 * @returns
 *     mapping between instance id and information about this instance.
 * @internal
 */
async function discoverInstances(instancesDir: string) {
	const instances = new Map<number, UnloadedInstance>();
	for (let entry of await fs.readdir(instancesDir, { withFileTypes: true })) {
		if (entry.isDirectory()) {
			let instanceConfig = new lib.InstanceConfig("host", Host.instanceConfigWarning);
			const instancePath = path.join(instancesDir, entry.name);
			const configPath = path.join(instancePath, "instance.json");

			try {
				instanceConfig = await lib.InstanceConfig.fromFile("host", configPath);
				instanceConfig.update(Host.instanceConfigWarning, false); // File might be missing the warning

			} catch (err: any) {
				if (err.code === "ENOENT") {
					continue; // Ignore folders without config.json
				}

				logger.error(`Error occurred while parsing ${configPath}: ${err.message}`);
				continue;
			}

			if (instances.has(instanceConfig.get("instance.id"))) {
				logger.warn(`Ignoring instance with duplicate id in folder ${entry.name}`);
				continue;
			}

			logger.verbose(`found instance ${instanceConfig.get("instance.name")} in ${instancePath}`);
			instances.set(instanceConfig.get("instance.id"), {
				path: instancePath,
				config: instanceConfig,
			});
		}
	}
	return instances;
}

// Requests that wake up instances are responsible for making sure the
// instance status is correctly updated even if they fail with an error. This
// means they should typically use try catch on anything that can throw
// and call notifyExit upon any errors.
const instanceStartingMessages = new Set([
	lib.InstanceStartRequest.name,
	lib.InstanceLoadScenarioRequest.name,
	lib.InstanceCreateSaveRequest.name,
	lib.InstanceExportDataRequest.name,
]);

export class HostRouter {
	constructor(
		public host: Host
	) { }

	/**
	 * Forward a message to the next hop towards its destination.
	 *
	 * @param origin -
	 *    Link the message originated from.
	 * @param message - Message to process.
	 * @param entry - message entry if message is a request or an event.
	 * @param hasFallback - is fallback available
	 * @returns true if the message was handled, false if fallback
	 *     is requested
	 */
	forwardMessage(
		origin: lib.Link,
		message: lib.MessageRoutable,
		entry: lib.RequestEntry | lib.EventEntry | undefined,
		hasFallback: boolean,
	) {
		let dst = message.dst;
		let nextHop: lib.Link | undefined;
		let msg: string | undefined;
		if (dst.type === lib.Address.broadcast) {
			this.broadcastMessage(origin, message, entry as lib.EventEntry);
			return true;
		} else if (
			dst.type === lib.Address.controller
			|| dst.type === lib.Address.control
			|| dst.type === lib.Address.host && dst.id !== this.host.config.get("host.id")
			|| dst.type === lib.Address.instance && !this.host.assignedInstances.has(dst.id)
		) {
			nextHop = this.host;
		} else if (dst.type === lib.Address.instance && this.host.assignedInstances.has(dst.id)) {
			nextHop = this.host.instanceConnections.get(dst.id)!;
		}

		if (nextHop === origin) {
			msg = `Message would return back to sender ${origin.connector.dst}.`;
			nextHop = undefined;
		}

		if (message.type === "request") {
			if (
				dst.type === lib.Address.instance
				&& instanceStartingMessages.has((message as lib.MessageRequest).name)
			) {
				this.wakeInstance(origin, message, nextHop);
				return true;

			} else if (!nextHop) {
				if (dst.type === lib.Address.instance) {
					if (hasFallback) {
						return false;
					}
					origin.connector.sendResponseError(
						new lib.ResponseError(msg || "Instance is not running.", "UnavailableError"), message.src
					);
				}
				return true;
			}

			// XXX What if the session is invalidated and there is no
			// response?  Need to track pending requests here.
		}

		if (nextHop) {
			this.sendMessage(nextHop, message, origin);
		} else {
			this.warnUnrouted(message, msg);
		}

		return true;
	}

	broadcastMessage(
		origin: lib.Link,
		message: lib.MessageRoutable,
		entry: lib.EventEntry,
	) {
		let dst = message.dst;
		if (message.type !== "event") {
			this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`);
			return;
		}
		const plugin = entry.Event.plugin;
		if (dst.id === lib.Address.instance) {
			for (let instanceConnection of this.host.instanceConnections.values()) {
				if (instanceConnection !== origin && (!plugin || instanceConnection.plugins.has(plugin))) {
					instanceConnection.connector.forward(message);
				}
			}
			if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) {
				this.host.connector.forward(message);
			}
		} else if (dst.id === lib.Address.host) {
			// Sent up to the controller, which passes it on to the other
			// hosts. Nothing left to do when it came from there.
			if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) {
				this.host.connector.forward(message);
			}
		} else if (dst.id === lib.Address.control) {
			if (this.host !== origin) {
				if (!plugin || this.host.serverPlugins.has(plugin)) {
					this.host.connector.forward(message);
				}
			} else {
				logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from controller.`);
			}
		} else {
			this.warnUnrouted(message, `Unexpected broadcast target ${dst.id}`);
		}
	}

	wakeInstance(origin: lib.Link, message: lib.MessageRoutable, nextHop?: lib.Link) {
		let dst = message.dst;
		if (nextHop) {
			origin.connector.sendResponseError(
				new lib.ResponseError("Instance is already running.", "ConflictError"), message.src
			);
			return;
		}
		if (this.host._shuttingDown) {
			origin.connector.sendResponseError(
				new lib.ResponseError("Host is shutting down.", "UnavailableError"), message.src
			);
			return;
		}

		this.host._connectInstance(dst.id).then(instanceConnection => {
			instanceConnection.connector.forward(message);
		}).catch(err => {
			logger.error(`Error starting instance:\n${err.stack ?? err.message}`, this.host.instanceLogMeta(dst.id));
			origin.connector.sendResponseError(lib.Link.errorToResponse(err), message.src);
		});
	}

	sendMessage(nextHop: lib.Link, message: lib.MessageRoutable, origin: lib.Link) {
		try {
			if (message.type === "request") {
				nextHop.forwardRequest(message as lib.MessageRequest, origin);
			} else {
				nextHop.connector.forward(message);
			}
		} catch (err: any) {
			if (message.type === "request") {
				origin.connector.sendResponseError(lib.Link.errorToResponse(err), message.src);
			}
			logger.warn(`Failed to deliver ${(message as any).name || "message"} (${message.type}): ${err.message}`);
			if (err instanceof lib.InvalidMessage && err.errors) {
				logger.warn(JSON.stringify(err.errors, null, "\t"));
			}
		}
	}

	warnUnrouted(message: lib.MessageRoutable, msg?: string) {
		let dst = message.dst;
		let baseMsg = `No destination for ${message.constructor.name} routed from ${message.src} to ${dst}`;
		logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`);
	}
}

// Instance information required to location and load an instance
type UnloadedInstance = {
	path: string,
	config: lib.InstanceConfig,
};

/**
 * Handles running the host
 *
 * Connects to the controller over the WebSocket and manages intsances.
 */
export default class Host extends lib.Link {
	declare "connector": HostConnector;

	router = new HostRouter(this);
	pluginInfos: lib.PluginNodeEnvInfo[];
	config: lib.HostConfig;

	/** Hooks which plugins can attach to */
	hooks = new HostHooks(logger);
	/** Plugins which are currently loaded */
	loadedPlugins: Set<lib.PluginNodeEnvInfo> = new Set();

	/** A map from instance id to instance connection. Only present when instance is running. */
	instanceConnections = new Map<number, InstanceConnection>();
	/** A map of instances that exist in the instance directory of the host */
	discoveredInstances = new Map<number, UnloadedInstance>();
	/** A map of instances that are assigned to this host */
	assignedInstances = new Map<number, UnloadedInstance>();

	adminlist = new Set<string>();
	banlist = new Map<string, string>();
	whitelist = new Set<string>();

	serverVersion = "unknown";
	serverPlugins = new Map<string, string>();

	_startup = true;
	_disconnecting = false;
	_shuttingDown = false;
	/** Instances waiting on {@link Instance.fromConfig} */
	_startingInstances = new Set<number>();

	startedAtMs = Date.now();

	static instanceConfigWarning = {
		"_warning": "Changes to this file will be overwritten by the controller's copy.",
	};

	/** Create a host with its mod store loaded from disk */
	static async fromConfig(
		hostConfig: lib.HostConfig,
		pluginInfos: lib.PluginNodeEnvInfo[],
		connector: HostConnector,
		canRestart = false,
		recoveryMode = false,
	) {
		const modsDirectory = hostConfig.get("host.mods_directory");
		await fs.mkdir(modsDirectory, { recursive: true });
		const modStore = await lib.ModStore.fromDirectory(modsDirectory);
		return new this(connector, hostConfig, pluginInfos, canRestart, recoveryMode, modStore);
	}

	constructor(
		connector: HostConnector,
		hostConfig: lib.HostConfig,
		pluginInfos: lib.PluginNodeEnvInfo[],
		/**
		 * If true indicates that there is a process monitor present that
		 * will restart this host on a non-zero exit codes.
		 */
		public canRestart = false,
		/**
		 * If true indicates the host is in recovery mode and should
		 * disable certain actions such as loading plugins or instance autostart
		 */
		public recoveryMode = false,
		public modStore = new lib.ModStore(hostConfig.get("host.mods_directory"), new Map()),
	) {
		super(connector);

		this.pluginInfos = pluginInfos;
		this.config = hostConfig;

		this.config.on("fieldChanged", (name, curr, prev) => {
			if (name === "host.name" || name === "host.public_address") {
				this.sendHostUpdate();
			}
			this.hooks.hostConfigFieldChanged.invoke(name, curr, prev);
		});

		this.connector.on("hello", data => {
			this.serverVersion = data.version;
			this.serverPlugins = new Map(Object.entries(data.plugins));
		});

		this.connector.on("connect", () => {
			if (this._shuttingDown) {
				return;
			}

			this.sendHostUpdate();
			this.updateInstances().catch((err) => {
				if (err instanceof lib.SessionLost) {
					return undefined;
				}

				logger.fatal(`Unexpected error updating instances:\n${err.stack}`);
				return this.shutdown();
			});
		});

		this.connector.on("close", () => {
			if (this._shuttingDown) {
				return;
			}

			if (this._disconnecting) {
				this._disconnecting = false;
				this.connector.connect().catch((err) => {
					logger.fatal(`Unexpected error reconnecting to controller:\n${err.stack}`);
					return this.shutdown();
				});
			} else {
				logger.fatal("Controller connection was unexpectedly closed");
				this.shutdown();
			}
		});

		this.connector.on("error", err => {
			logger.fatal(err.message);
			this.shutdown();
		});

		for (let event of ["connect", "drop", "resume", "close"] as const) {
			this.connector.on(event, () => {
				let message = new lib.ControllerConnectionEvent(event);
				for (let instanceConnection of this.instanceConnections.values()) {
					instanceConnection.send(message);
				}
				this.hooks.controllerConnectionEvent.invoke(event);
			});
		}

		this.handle(lib.HostStopRequest, this.handleHostStopRequest.bind(this));
		this.handle(lib.HostRestartRequest, this.handleHostRestartRequest.bind(this));
		this.handle(lib.HostMarkRestartRequiredRequest, this.handleHostMarkRestartRequiredRequest.bind(this));
		this.handle(lib.HostUpdateRequest, this.handleHostUpdateRequest.bind(this));
		this.handle(lib.HostConfigGetRequest, this.handleHostConfigGetRequest.bind(this));
		this.handle(lib.HostConfigSetRequest, this.handleHostConfigSetRequest.bind(this));
		this.handle(lib.HostConfigSetFieldRequest, this.handleHostConfigSetFieldRequest.bind(this));
		this.handle(lib.HostConfigSetPropRequest, this.handleHostConfigSetPropRequest.bind(this));
		this.handle(lib.HostMetricsRequest, this.handleHostMetricsRequest.bind(this));
		this.handle(lib.SyncUserListsEvent, this.handleSyncUserListsEvent.bind(this));
		this.handle(lib.SystemInfoRequest, this.handleSystemInfoRequest.bind(this));
		this.handle(lib.PluginListRequest, this.handlePluginListRequest.bind(this));
		this.handle(lib.PluginUpdateRequest, this.handlePluginUpdateRequest.bind(this));
		this.handle(lib.PluginInstallRequest, this.handlePluginInstallRequest.bind(this));
		this.handle(lib.UpdateAllRequest, this.handleUpdateAllRequest.bind(this));

		this.snoopEvent(lib.InstanceAdminlistUpdateEvent, this.handleAdminlistUpdateEvent.bind(this));
		this.snoopEvent(lib.InstanceBanlistUpdateEvent, this.handleBanlistUpdateEvent.bind(this));
		this.snoopEvent(lib.InstanceWhitelistUpdateEvent, this.handleWhitelistUpdateEvent.bind(this));

		this.handle(lib.InstanceAssignInternalRequest, this.handleInstanceAssignInternalRequest.bind(this));
		this.handle(lib.InstanceUnassignInternalRequest, this.handleInstanceUnassignInternalRequest.bind(this));
		this.fallbackRequest(
			lib.InstanceSaveDetailsListRequest, this.fallbackInstanceSaveDetailsListRequest.bind(this),
		);
		this.handle(lib.InstanceRenameSaveRequest, this.handleInstanceRenameSaveRequest.bind(this));
		this.handle(lib.InstanceCopySaveRequest, this.handleInstanceCopySaveRequest.bind(this));
		this.handle(lib.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this));
		this.handle(lib.InstanceDeleteSaveRequest, this.handleInstanceDeleteSaveRequest.bind(this));
		this.handle(lib.InstancePullSaveRequest, this.handleInstancePullSaveRequest.bind(this));
		this.handle(lib.InstancePushSaveRequest, this.handleInstancePushSaveRequest.bind(this));
		this.handle(lib.InstanceDeleteInternalRequest, this.handleInstanceDeleteInternalRequest.bind(this));
	}

	async loadPlugins() {
		const context = { logger, host: this };
		for (let pluginInfo of this.pluginInfos) {
			if (
				!pluginInfo.hostEntrypoint && !pluginInfo.instanceEntrypoint
				|| !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,
					"host",
					context,
					"HostPlugin",
					BaseHostPlugin,
				);

				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}`);
		}
	}

	async _createNewInstanceDir(name: string) {
		name = lib.cleanFilename(name);
		try {
			lib.checkFilename(name);
		} catch (err: any) {
			throw new Error(`Instance folder was unepectedly invalid: name ${err.message}`);
		}

		let instancesDir = this.config.get("host.instances_directory");
		for (let i = 1; ; i++) {
			let candidateDir = path.join(instancesDir, i === 1 ? name : `${name}-${i}`);
			try {
				await fs.mkdir(candidateDir);
			} catch (err: any) {
				if (err.code === "EEXIST") {
					continue;
				}
				throw err;
			}
			return candidateDir;
		}
	}

	async broadcastEventToInstance<T>(event: lib.Event<T>) {
		for (let instanceConnection of this.instanceConnections.values()) {
			if (event.constructor.plugin && !instanceConnection.plugins.has(event.constructor.plugin)) {
				continue;
			}
			instanceConnection.send(event);
		}
	}

	async handleHostStopRequest() {
		this.shutdown();
	}

	async handleHostRestartRequest() {
		if (!this.canRestart) {
			throw new lib.ConflictError("Cannot restart, host does not have a process monitor to restart it.");
		}
		if (await lib.packageOperationRunning()) {
			throw new lib.ConflictError("Cannot restart while a plugin install or update is running on this machine.");
		}
		const downgrade = await this.checkRestartDowngrade();
		if (downgrade) {
			const { installedVersion, runningVersion } = downgrade;
			throw new lib.ConflictError(
				`Cannot restart host with installed Clusterio version ${installedVersion} because it is older than ` +
				`running version ${runningVersion}. Stop the host before starting the older version manually.`
			);
		}
		process.exitCode = 1;
		this.shutdown();
	}

	async handleHostMarkRestartRequiredRequest() {
		this.config.restartRequired = true;
	}

	async handleHostConfigGetRequest() {
		return this.config.toRemote("control");
	}

	private static validateHostConfigSetField(fieldName: string) {
		if (fieldName === "host.id") {
			// Changing host.id at runtime is not worth the effort to
			// support and will break a lot of things if it is allowed.
			throw new lib.ConflictError("Setting 'host.id' while host is running is not supported");
		}
	}

	async handleHostConfigSetRequest(request: lib.HostConfigSetRequest) {
		try {
			for (const [fieldName, fieldValue] of Object.entries(request.fields)) {
				Host.validateHostConfigSetField(fieldName);
				if (typeof fieldValue === "object") {
					for (const [propName, propValue] of Object.entries(fieldValue)) {
						this.config.stageProp(fieldName as keyof lib.HostConfigFields, propName, propValue, "control");
					}
				} else {
					this.config.stage(fieldName as keyof lib.HostConfigFields, fieldValue, "control");
				}
			}

			this.config.commitStaging();
		} finally {
			this.config.revertStaging();
		}
	}

	async handleHostConfigSetFieldRequest(request: lib.HostConfigSetFieldRequest) {
		Host.validateHostConfigSetField(request.field);
		this.config.set(request.field as keyof lib.HostConfigFields, request.value, "control");
	}

	async handleHostConfigSetPropRequest(request: lib.HostConfigSetPropRequest) {
		let { field, prop, value } = request;
		Host.validateHostConfigSetField(field);
		this.config.setProp(field as keyof lib.HostConfigFields, prop, value, "control");
	}

	async handleSyncUserListsEvent(event: lib.SyncUserListsEvent) {
		let updateList = (list: Set<string>, updatedList: Set<string>, Event: lib.EventClass<unknown>) => {
			let added = new Set(updatedList);
			let removed = new Set(list);
			list.forEach(el => added.delete(el));
			updatedList.forEach(el => removed.delete(el));

			for (let name of added) {
				list.add(name);
				this.broadcastEventToInstance(new Event(name, true));
			}

			for (let name of removed) {
				list.delete(name);
				this.broadcastEventToInstance(new Event(name, false));
			}
		};

		updateList(this.adminlist, event.adminlist, lib.InstanceAdminlistUpdateEvent);
		updateList(this.whitelist, event.whitelist, lib.InstanceWhitelistUpdateEvent);

		let addedOrChanged = new Map(event.banlist);
		let removed = new Set(this.banlist.keys());
		addedOrChanged.forEach((_, name) => removed.delete(name));
		this.banlist.forEach((reason, name) => {
			if (addedOrChanged.get(name) === reason) {
				addedOrChanged.delete(name);
			}
		});

		for (let [name, reason] of addedOrChanged) {
			this.banlist.set(name, reason);
			this.broadcastEventToInstance(new lib.InstanceBanlistUpdateEvent(name, true, reason));
		}

		for (let name of removed) {
			this.banlist.delete(name);
			this.broadcastEventToInstance(new lib.InstanceBanlistUpdateEvent(name, false, ""));
		}
	}

	async handleAdminlistUpdateEvent(event: lib.InstanceAdminlistUpdateEvent) {
		let { name, admin } = event;
		if (admin) {
			this.adminlist.add(name);
		} else {
			this.adminlist.delete(name);
		}
	}

	async handleBanlistUpdateEvent(event: lib.InstanceBanlistUpdateEvent) {
		let { name, banned, reason } = event;
		if (banned) {
			this.banlist.set(name, reason);
		} else {
			this.banlist.delete(name);
		}
	}

	async handleWhitelistUpdateEvent(event: lib.InstanceWhitelistUpdateEvent) {
		let { name, whitelisted } = event;
		if (whitelisted) {
			this.whitelist.add(name);
		} else {
			this.whitelist.delete(name);
		}
	}

	async downloadMod(mod: lib.ModRecord) {
		let streamId = await this.send(new lib.ModDownloadRequest(mod.name, mod.version, mod.sha1));

		let url = new URL(this.config.get("host.controller_url"));
		url.pathname += `api/stream/${streamId}`;

		const file = lib.ModInfo.filename(mod.name, mod.version);
		const filePath = path.join(this.modStore.modsDirectory, file);
		await lib.downloadFile(url, filePath, "overwrite");

		const modInfo = await this.modStore.loadFile(file);
		if (mod.sha1 && mod.sha1 !== modInfo.sha1) {
			throw new Error(`Checksum mismatch downloading ${file} from controller`);
		}
		logger.info(`Downloaded ${file} from controller`);
		return modInfo;
	}

	private _fetchModQueue = new lib.AsyncSerialCallback(this._fetchModInternal.bind(this));

	async fetchMod(mod: lib.ModRecord) {
		return await this._fetchModQueue.invoke(mod);
	}

	private async _fetchModInternal(mod: lib.ModRecord) {
		// The mod may have already been downloaded in a previous call
		let modInfo = this.modStore.getMod(mod.name, mod.version);
		if (modInfo && (!mod.sha1 || mod.sha1 === modInfo.sha1)) {
			return modInfo;
		}

		// The mods folder may be shared with the controller and thus
		// already have this mod, try loading it.
		try {
			modInfo = await this.modStore.loadMod(mod.name, mod.version);
		} catch (err: any) {
			// A broken zip file may have been left in the mods folder
			if (err.code !== "ENOENT") {
				logger.error(`Error loading mod ${mod.name} ${mod.version}: ${err.message}`);
			}
		}
		if (modInfo && (!mod.sha1 || mod.sha1 === modInfo.sha1)) {
			return modInfo;
		}

		// Fetch the mod from the controller.
		return await this.downloadMod(mod);
	}

	async fetchMods(mods: Iterable<lib.ModRecord>, factorioVersion: lib.PartialVersion) {
		const builtinModNames = lib.ModPack.getBuiltinModNames(factorioVersion);
		const modInfos: Promise<lib.ModInfo>[] = [];
		for (const mod of mods) {
			if (builtinModNames.includes(mod.name)) {
				continue;
			}
			modInfos.push(this.fetchMod(mod));
		}
		return await Promise.all(modInfos);
	}

	async handleInstanceAssignInternalRequest(request: lib.InstanceAssignInternalRequest) {
		let { instanceId, config } = request;
		let instance = this.assignedInstances.get(instanceId);
		if (instance) {
			instance.config.update(config, true, "controller");
			logger.verbose(`Updated config for ${instance.path}`, this.instanceLogMeta(instanceId, instance));

		} else {
			instance = this.discoveredInstances.get(instanceId);
			if (instance) {
				await Instance.populate_folders(instance.path);
				instance.config.update(config, true, "controller");

			} else {
				let instanceConfig = new lib.InstanceConfig("host", Host.instanceConfigWarning);
				instanceConfig.update(config, false, "controller");

				let instanceDir = await this._createNewInstanceDir(instanceConfig.get("instance.name"));
				instanceConfig.filepath = path.join(instanceDir, "instance.json");

				logger.info(`Creating ${instanceDir}`);
				await Instance.populate_folders(instanceDir);
				instance = {
					path: instanceDir,
					config: instanceConfig,
				};

				this.discoveredInstances.set(instanceId, instance);
			}

			this.assignedInstances.set(instanceId, instance);
			logger.verbose(`assigned instance ${instance.config.get("instance.name")}`);
		}

		// Somewhat hacky, but in the event of a lost session the status is
		// resent on assigment since the controller sends an assigment
		// request for all the instances it knows should be on this host.
		let instanceConnection = this.instanceConnections.get(instanceId);
		this.send(
			new lib.InstanceStatusChangedEvent(
				instanceId,
				instanceConnection ? instanceConnection.instance.status : this._unconnectedStatus(instanceId),
				instanceConnection ? instanceConnection.instance.server.gamePort : this.gamePort(instanceId),
				instanceConnection
					? instanceConnection.instance.server.version
					// TODO: factorio.version is not validated
					: instance.config.get("factorio.version") as lib.TargetVersion,
			)
		);

		// Send the new list of saves for this assigned instance to the controller.
		await this.sendSaveListUpdate(instanceId, path.join(instance.path, "saves"));

		// save a copy of the instance config
		await instance.config.save();
	}

	async handleInstanceUnassignInternalRequest(request: lib.InstanceUnassignInternalRequest) {
		let instanceId = request.instanceId;
		let instance = this.assignedInstances.get(instanceId);
		if (instance) {
			let instanceConnection = this.instanceConnections.get(instanceId);
			if (instanceConnection && ["starting", "running"].includes(instanceConnection.instance.status)) {
				await instanceConnection.send(new lib.InstanceStopRequest());
			}

			this.assignedInstances.delete(instanceId);
			logger.verbose(`unassigned instance ${instance.config.get("instance.name")}`);
		}
	}

	instanceLogMeta(instanceId: number, instance?: { config: lib.InstanceConfig }) {
		instance = instance || this.assignedInstances.get(instanceId);
		if (!instance) {
			return { instance_id: instanceId, instance_name: String(instanceId) };
		}
		return { instance_id: instanceId, instance_name: instance.config.get("instance.name") };
	}

	getRequestInstance(instanceId: number) {
		let instance = this.assignedInstances.get(instanceId);
		if (!instance) {
			throw new lib.NotFoundError(`Instance with ID ${instanceId} does not exist`);
		}
		return instance;
	}

	/**
	 * Retrieved assigned port of the given instance
	 * @param instanceId - ID of the instance to get the assigned game port of:
	 * @returns Assigned game port or undefined if it does not exist.
	 */
	gamePort(instanceId: number) {
		const instance = this.discoveredInstances.get(instanceId);
		if (!instance) {
			return undefined;
		}
		return (
			instance.config.get("factorio.game_port")
			?? instance.config.get("factorio.host_assigned_game_port")
			?? undefined
		);
	}

	assignGamePort(instanceId: number, inUse = new Set<number>()) {
		const availablePorts = lib.parseRanges(this.config.get("host.factorio_port_range"), 1, 2 ** 16 - 1);
		for (const port of inUse) {
			availablePorts.delete(port);
		}
		for (const [id, instance] of this.assignedInstances) {
			if (id === instanceId) {
				continue;
			}
			const port = (
				instance.config.get("factorio.game_port")
				?? instance.config.get("factorio.host_assigned_game_port")
			);
			if (port !== null) {
				availablePorts.delete(port);
			}
		}
		const instanceToAssign = this.assignedInstances.get(instanceId)!;
		const assignedPort = instanceToAssign.config.get("factorio.host_assigned_game_port");
		if (assignedPort !== null && availablePorts.has(assignedPort)) {
			return assignedPort;
		}
		if (!availablePorts.size) {
			const others = inUse.size ? ` (${inUse.size} in use by other processes)` : "";
			throw new lib.ConflictError(`No available port left to assign from host.factorio_port_range${others}`);
		}
		const [newPort] = availablePorts;
		instanceToAssign.config.set("factorio.host_assigned_game_port", newPort);
		return newPort;
	}

	/**
	 * Assign a game port that no other process is bound to
	 *
	 * Replaces the stored port if it is taken, so a port another process
	 * holds is not reused on every start.
	 *
	 * @param instanceId - ID of the instance to assign a game port to.
	 * @returns the assigned port.
	 */
	async assignFreeGamePort(instanceId: number) {
		const inUse = new Set<number>();
		for (;;) {
			// Picking and storing the port before checking it keeps concurrent assignments from picking the same one.
			const port = this.assignGamePort(instanceId, inUse);
			if (await udpPortFree(port)) {
				return port;
			}
			logger.warn(`Game port ${port} is in use by another process`, this.instanceLogMeta(instanceId));
			inUse.add(port);
		}
	}

	/** Status of an instance without a connection */
	_unconnectedStatus(instanceId: number) {
		return this._startingInstances.has(instanceId) ? "starting" as const : "stopped" as const;
	}

	/**
	 * Initialize and connect an unloaded instance
	 *
	 * @param instanceId - ID of instance to initialize.
	 * @returns connection to instance.
	 */
	async _connectInstance(instanceId: number) {
		let unloadedInstance = this.getRequestInstance(instanceId);
		if (this.instanceConnections.has(instanceId) || this._startingInstances.has(instanceId)) {
			throw new lib.ConflictError(`Instance with ID ${instanceId} is running`);
		}

		let hostAddress = new lib.Address(lib.Address.host, this.config.get("host.id"));
		let instanceAddress = new lib.Address(lib.Address.instance, instanceId);
		let [connectionClient, connectionServer] = lib.VirtualConnector.makePair(instanceAddress, hostAddress);
		let instance: Instance;
		this._startingInstances.add(instanceId);
		try {
			instance = await Instance.fromConfig(
				unloadedInstance.config, this, connectionClient,
				unloadedInstance.path, this.config.get("host.factorio_directory"),
			);
		} finally {
			this._startingInstances.delete(instanceId);
		}
		let instanceConnection = new InstanceConnection(connectionServer, this, instance);

		this.instanceConnections.set(instanceId, instanceConnection);
		await instance.loadPlugins(this.pluginInfos);

		return instanceConnection;
	}

	async handleSystemInfoRequest() {
		const restartRequired = await this.checkRestartRequired();
		return lib.gatherSystemInfo(
			this.config.get("host.id"), this.canRestart, restartRequired, this.startedAtMs, await this.listPlugins(),
		);
	}

	async checkRestartRequired() {
		if (this.config.restartRequired) {
			return true;
		}

		let packageName = "@clusterio/host";
		try {
			// First check the clusterio version
			const runningVersion = this.config.get("host.version");
			const packageJsonPath = fileURLToPath(import.meta.resolve("@clusterio/host/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.hostEntrypoint && !pluginInfo.instanceEntrypoint
					|| !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("host.version");
			const packageJsonPath = fileURLToPath(import.meta.resolve("@clusterio/host/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 host version ${runningVersion} ` +
					`with installed version ${installedVersion}.`
				);
				return null;
			}

			if (semver.lt(installedVersion, runningVersion)) {
				return { installedVersion, runningVersion };
			}
		} catch (err: any) {
			logger.warn(`Failed to check host version before restart:\n${err.stack ?? err.message}`);
		}

		return null;
	}

	async handleHostMetricsRequest() {
		let requests: Promise<InstanceType<typeof lib.InstanceMetricsRequest["Response"]>>[] = [];
		for (let instanceConnection of this.instanceConnections.values()) {
			requests.push(instanceConnection.send(new lib.InstanceMetricsRequest()));
		}

		let results = [];
		let pluginResults = await this.hooks.metrics.collect();
		for (let metricIterator of pluginResults) {
			for await (let metric of metricIterator) {
				results.push(metric);
			}
		}

		for (let response of await Promise.all(requests)) {
			results.push(...response.results);
		}

		for await (let result of lib.defaultRegistry.collect()) {
			if (result.metric.name.startsWith("process_")) {
				results.push(lib.serializeResult(result, {
					addLabels: { "host_id": String(this.config.get("host.id")) },
					metricName: result.metric.name.replace("process_", "clusterio_host_"),
				}));

			} else if (result.metric.name.startsWith("system_")) {
				results.push(lib.serializeResult(result, {
					addLabels: { "host_id": String(this.config.get("host.id")) },
					metricName: result.metric.name.replace("system_", "clusterio_host_system_"),
				}));

			} else {
				results.push(lib.serializeResult(result));
			}
		}

		return { results };
	}

	async fallbackInstanceSaveDetailsListRequest(
		request: lib.InstanceSaveDetailsListRequest,
		src: lib.Address,
		dst: lib.Address
	) {
		let instance = this.getRequestInstance(dst.id);
		return await Instance.listSaves(dst.id, path.join(instance.path, "saves"), null);
	}

	async handleInstanceRenameSaveRequest(request: lib.InstanceRenameSaveRequest) {
		let { instanceId, oldName, newName } = request;
		checkRequestSaveName(oldName);
		checkRequestSaveName(newName);
		let instance = this.getRequestInstance(instanceId);
		try {
			await moveSaveExclusive(
				path.join(instance.path, "saves", oldName),
				path.join(instance.path, "saves", newName),
			);
		} catch (err: any) {
			if (err.code === "EEXIST") {
				throw new lib.ConflictError(`${newName} already exists`);
			}
			if (err.code === "ENOENT") {
				throw new lib.NotFoundError(`${oldName} does not exist`);
			}
			throw err;
		}
		await this.sendSaveListUpdate(instanceId, path.join(instance.path, "saves"));
	}

	async handleInstanceCopySaveRequest(request: lib.InstanceCopySaveRequest) {
		let { instanceId, source, destination } = request;
		checkRequestSaveName(source);
		checkRequestSaveName(destination);
		let instance = this.getRequestInstance(instanceId);
		try {
			await fs.copyFile(
				path.join(instance.path, "saves", source),
				path.join(instance.path, "saves", destination),
				fs.constants.COPYFILE_EXCL | fs.constants.COPYFILE_FICLONE
			);
		} catch (err: any) {
			if (err.code === "ENOENT") {
				throw new lib.NotFoundError(`${source} does not exist`);
			}
			throw err;
		}
		await this.sendSaveListUpdate(instanceId, path.join(instance.path, "saves"));
	}

	async _moveOrCopySave(sourcePath: string, targetPath: string, copy: boolean) {
		const { dir, name, ext } = path.parse(targetPath);
		for (let i = 1; ; i++) {
			targetPath = path.format({ dir, name: i === 1 ? name : `${name}-${i}`, ext });
			try {
				if (copy) {
					await copySaveExclusive(sourcePath, targetPath);
				} else {
					await moveSaveExclusive(sourcePath, targetPath);
				}
				return path.basename(targetPath);
			} catch (err: any) {
				if (err.code !== "EEXIST") {
					throw err;
				}
			}
		}
	}

	async handleInstanceTransferSaveRequest(request: lib.InstanceTransferSaveRequest) {
		let { sourceName, targetName, copy, sourceInstanceId, targetInstanceId } = request;
		checkRequestSaveName(sourceName);
		checkRequestSaveName(targetName);
		let sourceInstance = this.getRequestInstance(sourceInstanceId);
		let targetInstance = this.getRequestInstance(targetInstanceId);

		if (!targetName.endsWith(".zip")) {
			targetName += ".zip";
		}

		// For consistency with remote transfer initiated through pullSave the
		// target is renamed if it already exists.
		const sourcePath = path.join(sourceInstance.path, "saves", sourceName);
		try {
			targetName = await this._moveOrCopySave(
				sourcePath,
				path.join(targetInstance.path, "saves", targetName),
				copy,
			);
		} catch (err: any) {
			// ENOENT may also come from a missing target saves folder.
			if (err.code === "ENOENT" && !await fs.access(sourcePath).then(() => true, () => false)) {
				throw new lib.NotFoundError(`${sourceName} does not exist`);
			}
			throw err;
		}

		await this.sendSaveListUpdate(sourceInstanceId, path.join(sourceInstance.path, "saves"));
		await this.sendSaveListUpdate(targetInstanceId, path.join(targetInstance.path, "saves"));

		return targetName;
	}

	async sendSaveListUpdate(instanceId: number, savesDir: string) {
		let instanceConnection = this.instanceConnections.get(instanceId);
		let saveList: lib.SaveDetails[];
		if (instanceConnection) {
			saveList = await instanceConnection.send(new lib.InstanceSaveDetailsListRequest());
		} else {
			saveList = await Instance.listSaves(instanceId, savesDir, null);
		}

		this.send(new lib.InstanceSaveDetailsUpdatesEvent(saveList, instanceId));
	}

	async handleInstanceDeleteSaveRequest(request: lib.InstanceDeleteSaveRequest) {
		let { instanceId, name } = request;
		checkRequestSaveName(name);
		let instance = this.getRequestInstance(instanceId);

		try {
			await fs.unlink(path.join(instance.path, "saves", name));
		} catch (err: any) {
			if (err.code === "ENOENT") {
				throw new lib.NotFoundError(`${name} does not exist`);
			}
			throw err;
		}
		await this.sendSaveListUpdate(instanceId, path.join(instance.path, "saves"));
	}

	async handleInstancePullSaveRequest(request: lib.InstancePullSaveRequest) {
		let { instanceId, streamId, name } = request;
		checkRequestSaveName(name);
		let instance = this.getRequestInstance(instanceId);

		let url = new URL(this.config.get("host.controller_url"));
		url.pathname += `api/stream/${streamId}`;
		let savesDir = path.join(instance.path, "saves");
		const savedPath = await lib.downloadFile(url, path.join(savesDir, name), "rename");
		name = path.basename(savedPath);
		await this.sendSaveListUpdate(instanceId, savesDir);
		return name;
	}

	async handleInstancePushSaveRequest(request: lib.InstancePushSaveRequest) {
		let { instanceId, streamId, name } = request;
		checkRequestSaveName(name);
		let instance = this.getRequestInstance(instanceId);

		let content;
		try {
			content = (await fs.open(path.join(instance.path, "saves", name))).createReadStream();
		} catch (err: any) {
			if (err.code === "ENOENT") {
				throw new lib.NotFoundError(`${name} does not exist`);
			}
			throw err;
		}

		let url = new URL(this.config.get("host.controller_url"));
		url.pathname += `api/stream/${streamId}`;
		fetch(url, {
			method: "PUT",
			body: content,
			duplex: "half",
		}).catch(err => {
			logger.error(`Error pushing save to controller:\n${err.stack}`, this.instanceLogMeta(instanceId));
		});
	}

	async handleInstanceDeleteInternalRequest(request: lib.InstanceDeleteInternalRequest) {
		let instanceId = request.instanceId;
		if (this.instanceConnections.has(instanceId)) {
			throw new lib.ConflictError(`Instance with ID ${instanceId} is running`);
		}

		let instance = this.discoveredInstances.get(instanceId);
		if (!instance) {
			throw new lib.NotFoundError(`Instance with ID ${instanceId} does not exist`);
		}

		this.discoveredInstances.delete(instanceId);
		this.assignedInstances.delete(instanceId);
		await fs.rm(instance.path, { recursive: true, maxRetries: 10 });
	}

	async handleHostUpdateRequest(request: lib.HostUpdateRequest) {
		if (!this.config.get("host.allow_remote_updates")) {
			throw new lib.ForbiddenError("Remote updates are disabled on this machine");
		}
		return await lib.updatePackage("@clusterio/host");
	}

	async handlePluginUpdateRequest(request: lib.PluginUpdateRequest) {
		if (!this.config.get("host.allow_plugin_updates")) {
			throw new lib.ForbiddenError("Plugin updates are disabled on this machine");
		}
		return await lib.handlePluginUpdate(request.pluginPackage, this.pluginInfos);
	}

	async handlePluginInstallRequest(request: lib.PluginInstallRequest) {
		if (!this.config.get("host.allow_plugin_install")) {
			throw new lib.ForbiddenError("Plugin installs are disabled on this machine");
		}
		const npmPackage = await lib.handlePluginInstall(request.pluginPackage);
		// A new plugin is not in pluginInfos, so checkRestartRequired cannot notice it. Version
		// changes to loaded plugins are noticed there.
		if (!this.pluginInfos.some(pluginInfo => pluginInfo.npmPackage === npmPackage)) {
			this.config.restartRequired = true;
		}
	}

	async handleUpdateAllRequest(request: lib.UpdateAllRequest) {
		if (!this.config.get("host.allow_remote_updates")) {
			throw new lib.ForbiddenError("Remote updates are disabled on this machine");
		}
		if (!this.config.get("host.allow_plugin_updates")) {
			throw new lib.ForbiddenError("Plugin updates are disabled on this machine");
		}
		return await lib.handleUpdateAll("@clusterio/host", this.pluginInfos);
	}

	async handlePluginListRequest(request: lib.PluginListRequest) {
		return await this.listPlugins();
	}

	/** Plugins on this host, 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`),
			pluginInfo => Boolean(pluginInfo.hostEntrypoint || pluginInfo.instanceEntrypoint),
		);
	}

	sendHostUpdate() {
		this.send(
			new lib.HostInfoUpdateEvent(
				new lib.HostInfoUpdate(
					this.config.get("host.name"),
					this.config.get("host.public_address"),
				),
			),
		);
	}

	/**
	 * Discover available instances
	 *
	 * Looks through the instances directory for instances and updates
	 * the host and controller with the new list of instances.
	 */
	async updateInstances() {
		this.discoveredInstances = await discoverInstances(this.config.get("host.instances_directory"));
		let list = [];
		for (let [instanceId, instance] of this.discoveredInstances) {
			let instanceConnection = this.instanceConnections.get(instanceId);
			list.push(new lib.HostInstanceUpdate(
				instance.config.toRemote("controller"),
				instanceConnection ? instanceConnection.instance.status : this._unconnectedStatus(instanceId),
				instanceConnection ? instanceConnection.instance.server.gamePort : this.gamePort(instanceId),
				instanceConnection ? instanceConnection.instance.server.version : undefined,
			));
		}
		await this.send(new lib.InstancesUpdateRequest(list));

		// Handle configured auto startup instances
		if (this._startup) {
			this._startup = false;

			for (let [instanceId, instance] of this.assignedInstances) {
				if (instance.config.get("instance.auto_start")) {
					if (this.recoveryMode) {
						logger.warn(`Recovery | skipping auto startup for ${instance.config.get("instance.name")}`);
						continue;
					}
					try {
						let instanceConnection = await this._connectInstance(instanceId);
						await instanceConnection.instance.handleInstanceStartRequest(new lib.InstanceStartRequest());
					} catch (err: any) {
						logger.error(
							`Error during auto startup for ${instance.config.get("instance.name")}:\n${err.stack}`,
							this.instanceLogMeta(instanceId, instance)
						);
					}
				}
			}
		}
	}

	async prepareDisconnect() {
		await this.hooks.prepareControllerDisconnect.invoke(this);
		for (let instanceConnection of this.instanceConnections.values()) {
			await instanceConnection.send(new lib.PrepareControllerDisconnectRequest());
		}
		this._disconnecting = true;
		return await super.prepareDisconnect();
	}

	/**
	 * Stops all instances and closes the connection
	 */
	async shutdown() {
		if (this._shuttingDown) {
			return;
		}
		this._shuttingDown = true;

		await this.hooks.shutdown.invoke();

		for (let instanceConnection of this.instanceConnections.values()) {
			try {
				await instanceConnection.instance.stop();
			} catch (err: any) {
				logger.error(`Unexpected error stopping instance:\n${err.stack}`);
			}
		}

		try {
			await this.connector.disconnect();
		} catch (err: any) {
			if (!(err instanceof lib.SessionLost)) {
				logger.error(`Unexpected error preparing disconnect:\n${err.stack}`);
			}
		}

		logger.info("Saving config");
		await this.config.save();
		// Save host side in case host_assigned_game_port changed in the config.
		await Promise.all(
			[...this.assignedInstances.values()].map(instance => instance.config.save())
		);

		try {
			// Clear silly interval in pidfile library.
			pidusage.clear();
		} catch (err: any) {
			setBlocking(true);
			logger.error(`
+------------------------------------------------------------+
| Unexpected error occured while shutting down host, please  |
| report it to https://github.com/clusterio/clusterio/issues |
+------------------------------------------------------------+
${err.stack}`
			);

			process.exit(1);
		}
	}

	/**
	 * True if the connection to the controller is connected, not in the dropped
	 * state,and not in the process of disconnecting.
	 */
	get connected() {
		return !this._disconnecting && this.connector.connected;
	}
}

// For testing only
export const _discoverInstances = discoverInstances;