Implement hosts downloading needed mods

Download and cache needed mods to run instance mod packs on instance
startup.  To avoid opening too many download streams at once only one
mod is downloaded at a time.

packages/host/host.ts

@@ -203,7 +203,14 @@ async function startHost() {
 	}
 
 	let hostConnector = new HostConnector(hostConfig, tlsCa, pluginInfos);
-	let host = new Host(hostConnector, args.config, hostConfig, tlsCa, pluginInfos);
+	let host = new Host(
+		hostConnector,
+		args.config,
+		hostConfig,
+		tlsCa,
+		pluginInfos,
+		...await Host.bootstrap(hostConfig)
+	);
 
 	// Handle interrupts
 	let secondSigint = false;

packages/host/src/Host.ts

@@ -263,12 +263,21 @@ export default class Host extends lib.Link {
 	_disconnecting = false;
 	_shuttingDown = false;
 
+	static async bootstrap(hostConfig: lib.HostConfig) {
+		const modsDirectory = hostConfig.get("host.mods_directory");
+		await fs.ensureDir(modsDirectory);
+		return [
+			await lib.ModStore.fromDirectory(modsDirectory),
+		] as const;
+	}
+
 	constructor(
 		connector: HostConnector,
 		hostConfigPath: string,
 		hostConfig: lib.HostConfig,
 		tlsCa: string | undefined,
-		pluginInfos: lib.PluginNodeEnvInfo[]
+		pluginInfos: lib.PluginNodeEnvInfo[],
+		public modStore: lib.ModStore,
 	) {
 		super(connector);
 		this.tlsCa = tlsCa;
@@ -478,6 +487,77 @@ export default class Host extends lib.Link {
 		}
 	}
 
+	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}`;
+		let response = await phin({
+			url, method: "GET",
+			core: { ca: this.tlsCa } as object,
+			stream: true,
+		});
+
+		const file = lib.ModInfo.filename(mod.name, mod.version);
+		const filePath = path.join(this.modStore.modsDirectory, file);
+		const tempFilePath = filePath.replace(/(\.zip)?$/, ".tmp.zip");
+		const writeStream = fs.createWriteStream(tempFilePath, { flags: "w" });
+		await events.once(writeStream, "open");
+		response.pipe(writeStream);
+		await finished(writeStream);
+		await fs.rename(tempFilePath, filePath);
+
+		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) {
+			if (err.code !== "ENOENT") {
+				throw err;
+			}
+		}
+		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>) {
+		const modInfos: Promise<lib.ModInfo>[] = [];
+		for (const mod of mods) {
+			// XXX These mods ship with Factorio, but the list of mods may
+			// change and should not be hardcoded here.
+			if (["core", "base"].includes(mod.name)) {
+				continue;
+			}
+			modInfos.push(this.fetchMod(mod));
+		}
+		return await Promise.all(modInfos);
+	}
+
 	async handleInstanceAssignInternalRequest(request: lib.InstanceAssignInternalRequest) {
 		let { instanceId, config } = request;
 		let instanceInfo = this.instanceInfos.get(instanceId);

packages/host/src/Instance.ts

@@ -646,9 +646,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		// TODO validate factorioVersion
 
-		if (!this._host.config.get("host.mods_directory_is_shared")) {
-			throw new Error("Fetching mods is not implemented");
-		}
+		const mods = await this._host.fetchMods(modPack.mods.values());
 
 		await fs.ensureDir(this.path("mods"));
 
@@ -668,8 +666,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		// Add mods from mod the pack
 		const modsDir = this._host.config.get("host.mods_directory");
-		for (let mod of this.activeModPack.mods.values()) {
-			const modFile = `${mod.name}_${mod.version}.zip`;
+		for (let mod of mods) {
+			const modFile = mod.filename;
 			const target = path.join(modsDir, modFile);
 			const link = this.path("mods", modFile);
 

packages/lib/src/config/definitions.ts

@@ -21,6 +21,7 @@ ControllerGroup.define({
 	name: "mods_directory",
 	title: "Mods Directory",
 	description: "Path to directory where mods shared with the cluster are stored.",
+	restartRequired: true,
 	type: "string",
 	initial_value: "mods",
 });
@@ -202,16 +203,10 @@ HostGroup.define({
 	name: "mods_directory",
 	title: "Mods Directory",
 	description: "Path to directory where mods for instances are cached.",
+	restartRequired: true,
 	type: "string",
 	initial_value: "mods",
 });
-HostGroup.define({
-	name: "mods_directory_is_shared",
-	title: "Mods Directory is Shared",
-	description: "True if the mods directory is shared with the controller.",
-	type: "boolean",
-	initial_value: true,
-});
 HostGroup.define({
 	name: "instances_directory",
 	description: "Path to directory to store instances in.",
@@ -264,7 +259,6 @@ export class HostConfig extends classes.Config {
 	get(name: "host.id"): number;
 	get(name: "host.factorio_directory"): string;
 	get(name: "host.mods_directory"): string;
-	get(name: "host.mods_directory_is_shared"): boolean;
 	get(name: "host.instances_directory"): string;
 	get(name: "host.controller_url"): string;
 	get(name: "host.controller_token"): string;

test/integration/clusterio.js

@@ -32,9 +32,12 @@ async function startAltHost() {
 	let configPath = path.join("temp", "test", config);
 	await fs.remove(configPath);
 	await fs.remove(path.join("temp", "test", "alt-instances"));
+	await fs.remove(path.join("temp", "test", "alt-mods"));
 	await execCtl(`host create-config --id 5 --name alt-host --generate-token --output ${config}`);
 	await exec(`node ../../packages/host --config ${config} config set host.tls_ca ../../test/file/tls/cert.pem`);
 	await exec(`node ../../packages/host --config ${config} config set host.instances_directory alt-instances`);
+	await exec(`node ../../packages/host --config ${config} config set host.factorio_directory ../../factorio`);
+	await exec(`node ../../packages/host --config ${config} config set host.mods_directory alt-mods`);
 	return await spawn("alt-host:", `node ../../packages/host run --config ${config}`, /Started host/);
 }
 
@@ -176,6 +179,23 @@ describe("Integration of Clusterio", function() {
 				assert(hosts.has(5), "Host list was not updated");
 			});
 		});
+		it("should download mods from controller", async function() {
+			// On windows there's currently no way to automate graceful shutdown of the host
+			// process as CTRL+C is some weird terminal thing and SIGINT isn't a thing.
+			if (process.platform === "win32") {
+				this.skip();
+			}
+
+			slowTest(this);
+			await runWithAltHost(async () => {
+				const clusterioLib = path.join("temp", "test", "alt-mods", "clusterio_lib_0.1.2.zip");
+				assert(!await fs.pathExists(clusterioLib), "mod was downloaded before the test");
+				await execCtl("instance create alt-mod --id 98");
+				await execCtl("instance assign alt-mod 5");
+				await execCtl("instance save create alt-mod");
+				assert(await fs.pathExists(clusterioLib), "mod was not downloaded by the test");
+			});
+		});
 	});
 
 	describe("clusterioctl", function() {