Merge pull request #1115 from ILLISIS/rs-queue-sync

Add a cluster research queue to research_sync

README.md

@@ -73,7 +73,7 @@ Plugins add functionality to Factorio servers, Clusterio itself or both.
 These are the plugins supported and maintained by the Clusterio developers:
 
 - [Global Chat](/plugins/global_chat/README.md): share the in-game chat between servers.
-- [Research Sync](/plugins/research_sync/README.md): synchronize research progress and technologies unlocked between servers.
+- [Research Sync](/plugins/research_sync/README.md): synchronize research progress and technologies unlocked and the research queue between servers.
 - [Statistics Exporter](/plugins/statistics_exporter/README.md): collect in-game statistics from all the servers and makes it available to the Prometheus endpoint on the controller.
 - [Subspace Storage](https://github.com/clusterio/subspace_storage): Provide shared storage that can transport items between servers via teleport chests.
 - [Player Auth](/plugins/player_auth/README.md): Provides authentication to the cluster via logging into a Factorio server.

plugins/research_sync/README.md

@@ -2,6 +2,32 @@
 
 Plugin synchronizing research progress between instances and the cluster.
 
+## Research queue
+
+Each server can follow one cluster research queue.
+It is set from the normal research screen: queueing, moving or cancelling a technology there changes it for every following server.
+So does a lab whose circuit "set research" picks a technology, and so can another mod.
+
+The "Follow the cluster research queue" checkbox on the lab window turns this on or off per server.
+Turning it on replaces the server's queue with the cluster's, or, if the cluster queue is empty, shares the server's queue as the cluster queue.
+Turning it off clears the server's queue.
+Flipping it prints a chat line on that server.
+It is on by default.
+The first server to turn following on with something queued shares its queue as the cluster queue, and servers that turn it on after that take the cluster's.
+A server's first start counts as turning it on; a later restart takes the cluster queue, even an empty one.
+Whenever the cluster queue is empty (for example, everything in it has finished), the next server that turns following on shares its queue again.
+Every change to the cluster queue prints a chat line on every server.
+
+It needs Factorio 2.0.56 or later, older servers ignore the queue.
+To turn it off without the GUI, run `/sc research_sync.set_follow(false)` on that server.
+
+Known gaps:
+
+- A script that assigns the queue directly to `{}` isn't seen.
+- Two following servers whose labs want different research keep flipping the cluster queue, as two labs do on one server.
+- A technology a server can't queue yet is dropped there, and an edit on that server then drops it for everyone.
+- An edit made on a server in the moment before it hears that a technology finished can put that technology back in the cluster queue. Every server skips it, since the game won't queue finished research, and the next edit removes it.
+
 
 ## Installation
 

plugins/research_sync/controller.ts

@@ -13,6 +13,9 @@ import {
 	TechnologySync,
 	SyncTechnologiesRequest,
 	TechnologyProgress,
+	QueueUpdateEvent,
+	QueueEvent,
+	QueueSyncRequest,
 } from "./messages.js";
 
 type Technology = {
@@ -49,12 +52,71 @@ async function saveTechnologies(
 	await lib.safeOutputFile(filePath, JSON.stringify([...technologies.entries()], null, "\t"));
 }
 
+const MAX_QUEUE = 7;
+
+async function loadQueue(controllerConfig: lib.ControllerConfig, logger: lib.Logger): Promise<string[]> {
+	let filePath = path.join(controllerConfig.get("controller.database_directory"), "research_queue.json");
+	logger.verbose(`Loading ${filePath}`);
+	try {
+		return JSON.parse(await fs.readFile(filePath, "utf8"));
+	} catch (err: any) {
+		if (err.code === "ENOENT") {
+			logger.verbose("Creating new research queue");
+			return [];
+		}
+		throw err;
+	}
+}
+
+async function saveQueue(controllerConfig: lib.ControllerConfig, queue: string[], logger: lib.Logger) {
+	let filePath = path.join(controllerConfig.get("controller.database_directory"), "research_queue.json");
+	logger.verbose(`writing ${filePath}`);
+	await lib.safeOutputFile(filePath, JSON.stringify(queue, null, "\t"));
+}
+
 export default async function(context: ControllerPluginContext) {
 	const { controller, logger, plugin } = context;
 	const technologies = await loadTechnologies(controller.config, logger);
 	let technologiesDirty = true;
+	let queue = await loadQueue(controller.config, logger);
+	let queueDirty = false;
 	const progressToBroadcast = new Set<string>();
 
+	function instanceName(src: lib.Address) {
+		return controller.instances.get(src.id)?.config.get("instance.name") ?? String(src.id);
+	}
+
+	// Removes one entry per level that finished, returns whether any was removed
+	function removeFinished(name: string, levels = 1): boolean {
+		let removed = false;
+		for (let i = 0; i < levels; i++) {
+			let index = queue.indexOf(name);
+			if (index === -1) {
+				break;
+			}
+			queue.splice(index, 1);
+			queueDirty = true;
+			removed = true;
+		}
+		return removed;
+	}
+
+	// Instances only advance the technology on a finish, so they need the queue without it.
+	// No player or instance: the finished research prints its own chat line.
+	function broadcastQueueSilently() {
+		controller.sendTo("allInstances", new QueueEvent(queue, null, null));
+	}
+
+	function setQueue(newQueue: string[], player: string | null, src: lib.Address) {
+		newQueue = newQueue.slice(0, MAX_QUEUE);
+		if (newQueue.length === queue.length && newQueue.every((name, i) => name === queue[i])) {
+			return;
+		}
+		queue = newQueue;
+		queueDirty = true;
+		controller.sendTo("allInstances", new QueueEvent(queue, player, instanceName(src)));
+	}
+
 	function broadcastProgress() {
 		let techs = [];
 		for (let name of progressToBroadcast) {
@@ -109,8 +171,12 @@ export default async function(context: ControllerPluginContext) {
 			tech.researched = true;
 			tech.progress = null;
 			progressToBroadcast.delete(name);
+			let removed = removeFinished(name);
 
 			controller.sendTo("allInstances", new FinishedEvent(name, tech.level));
+			if (removed) {
+				broadcastQueueSilently();
+			}
 		}
 		technologiesDirty = true;
 	});
@@ -118,12 +184,19 @@ export default async function(context: ControllerPluginContext) {
 	controller.handle(FinishedEvent, async (event: FinishedEvent) => {
 		let { name, level } = event;
 		let tech = technologies.get(name);
+		let removed = false;
+		if (!tech || tech.level < level || (tech.level === level && !tech.researched)) {
+			removed = removeFinished(name);
+		}
 		if (!tech || tech.level <= level) {
 			controller.sendTo("allInstances", event);
 			progressToBroadcast.delete(name);
 			technologies.set(name, { level, progress: null, researched: true });
 			technologiesDirty = true;
 		}
+		if (removed) {
+			broadcastQueueSilently();
+		}
 	});
 
 	controller.handle(SyncTechnologiesRequest, async (request: SyncTechnologiesRequest): Promise<TechnologySync[]> => {
@@ -135,6 +208,7 @@ export default async function(context: ControllerPluginContext) {
 			return Number.parseInt(match[1], 10);
 		}
 
+		let removed = false;
 		for (let instanceTech of request.technologies) {
 			let { name, level, progress, researched } = instanceTech;
 			let tech = technologies.get(name);
@@ -145,6 +219,9 @@ export default async function(context: ControllerPluginContext) {
 					progressToBroadcast.add(name);
 				} else if (researched || baseLevel(name) !== level) {
 					controller.sendTo("allInstances", new FinishedEvent(name, level));
+					// One queue entry per level unlocked
+					let levels = Math.max(level - Number(!researched) - baseLevel(name) + 1, 1);
+					removed = removeFinished(name, levels) || removed;
 				}
 
 			} else {
@@ -154,8 +231,10 @@ export default async function(context: ControllerPluginContext) {
 
 				if (tech.level < level || researched) {
 					// Send update if the unlocked level is greater
-					if (level - Number(!researched) > tech.level - Number(!tech.researched)) {
+					let unlocked = level - Number(!researched) - (tech.level - Number(!tech.researched));
+					if (unlocked > 0) {
 						controller.sendTo("allInstances", new FinishedEvent(name, level - Number(!researched)));
+						removed = removeFinished(name, unlocked) || removed;
 					}
 					tech.level = level;
 					tech.progress = progress;
@@ -176,6 +255,9 @@ export default async function(context: ControllerPluginContext) {
 			}
 		}
 		progressRateLimiter.activate();
+		if (removed) {
+			broadcastQueueSilently();
+		}
 
 		let result = [];
 		for (let [name, tech] of technologies) {
@@ -185,7 +267,22 @@ export default async function(context: ControllerPluginContext) {
 		return result;
 	});
 
+	controller.handle(QueueUpdateEvent, async (event: QueueUpdateEvent, src: lib.Address) => {
+		setQueue(event.queue, event.player, src);
+	});
+
+	controller.handle(QueueSyncRequest, async (request: QueueSyncRequest, src: lib.Address): Promise<string[]> => {
+		if (request.queue?.length && !queue.length) {
+			setQueue(request.queue, request.player, src);
+		}
+		return queue;
+	});
+
 	controller.hooks.save.attach(plugin.name, async () => {
+		if (queueDirty) {
+			queueDirty = false;
+			await saveQueue(controller.config, queue, logger);
+		}
 		if (technologiesDirty) {
 			technologiesDirty = false;
 			await saveTechnologies(controller.config, technologies, logger);

plugins/research_sync/index.ts

@@ -19,5 +19,8 @@ export const plugin: lib.PluginDeclaration = {
 		messages.ProgressEvent,
 		messages.FinishedEvent,
 		messages.SyncTechnologiesRequest,
+		messages.QueueUpdateEvent,
+		messages.QueueEvent,
+		messages.QueueSyncRequest,
 	],
 };

plugins/research_sync/instance.ts

@@ -6,6 +6,9 @@ import {
 	FinishedEvent,
 	TechnologySync,
 	SyncTechnologiesRequest,
+	QueueUpdateEvent,
+	QueueEvent,
+	QueueSyncRequest,
 } from "./messages.js";
 
 // ./module/sync.lua
@@ -18,10 +21,12 @@ type IpcFinished = {
 	name: string,
 	level: number,
 };
+type IpcQueue = { queue: unknown, player?: string };
 
 export default async function(context: InstancePluginContext) {
 	const { instance, logger, plugin } = context;
 	let syncStarted = false;
+	let queueEventsApplied = 0;
 
 	// Commands are sent one at a time so they execute in order
 	let rconQueue: Promise<unknown> = Promise.resolve();
@@ -35,6 +40,34 @@ export default async function(context: InstancePluginContext) {
 		logger.error(`Unexpected error:\n${err.stack}`);
 	}
 
+	// An empty Lua table serialises as {}, so anything that isn't an array is an empty queue
+	function asArray(v: unknown): string[] {
+		return Array.isArray(v) ? v : [];
+	}
+
+	async function applyQueue(data: { queue: string[], instance?: string, player?: string }) {
+		await sendOrderedRcon(`/sc research_sync.apply_queue("${lib.escapeString(JSON.stringify(data))}")`, true);
+	}
+
+	// queue null asks for the cluster queue without seeding it
+	async function followQueue(queue: string[] | null, player: string | null) {
+		let applied = queueEventsApplied;
+		let result = await instance.sendTo("controller", new QueueSyncRequest(queue, player));
+		// A QueueEvent handled meanwhile is at least as new as this reply, since the controller broadcasts every
+		// queue change in order with its replies. Arriving in the same read, it is applied before the reply.
+		if (queueEventsApplied !== applied) {
+			return;
+		}
+		await applyQueue({ queue: result });
+	}
+
+	async function syncQueue(seed: boolean) {
+		let state = JSON.parse(await sendOrderedRcon("/sc research_sync.queue_state()"));
+		if (state.follow) {
+			await followQueue(seed && state.first ? asArray(state.queue) : null, null);
+		}
+	}
+
 	async function researchContribution(tech: IpcContribution) {
 		instance.sendTo("controller", new ContributionEvent(tech.name, tech.level, tech.contribution));
 	}
@@ -50,6 +83,17 @@ export default async function(context: InstancePluginContext) {
 		researchFinished(tech).catch(err => unexpectedError(err));
 	});
 
+	instance.server.on("ipc-research_sync:queue", (data: IpcQueue) => {
+		try {
+			instance.sendTo("controller", new QueueUpdateEvent(asArray(data.queue), data.player ?? null));
+		} catch (err: any) {
+			unexpectedError(err);
+		}
+	});
+	instance.server.on("ipc-research_sync:follow", (data: IpcQueue) => {
+		followQueue(asArray(data.queue), data.player ?? null).catch(err => unexpectedError(err));
+	});
+
 	instance.handle(ProgressEvent, async (event: ProgressEvent) => {
 		if (!syncStarted || !["starting", "running"].includes(instance.status)) {
 			return;
@@ -68,6 +112,21 @@ export default async function(context: InstancePluginContext) {
 		);
 	});
 
+	instance.handle(QueueEvent, async (event: QueueEvent) => {
+		if (!syncStarted || !["starting", "running"].includes(instance.status)) {
+			return;
+		}
+		let data: { queue: string[], instance?: string, player?: string } = { queue: event.queue };
+		if (event.instance !== null) {
+			data.instance = event.instance;
+		}
+		if (event.player !== null) {
+			data.player = event.player;
+		}
+		queueEventsApplied += 1;
+		await applyQueue(data);
+	});
+
 	instance.hooks.start.attach(plugin.name, async () => {
 		let dumpJson = await sendOrderedRcon("/sc research_sync.dump_technologies()");
 		let techsToSend = [];
@@ -102,5 +161,21 @@ export default async function(context: InstancePluginContext) {
 			let syncJson = lib.escapeString(JSON.stringify(techsToSync));
 			await sendOrderedRcon(`/sc research_sync.sync_technologies("${syncJson}")`, true);
 		}
+
+		// A first start counts as turning following on, which seeds an empty cluster queue
+		await syncQueue(true);
+	});
+
+	// QueueEvents sent while the controller session was down are lost, so take the cluster queue again.
+	// An edit made on this instance during the outage was never sent and is replaced: the cluster queue wins.
+	instance.hooks.controllerConnectionEvent.attach(plugin.name, async (event) => {
+		if (event !== "connect" || !syncStarted || !["starting", "running"].includes(instance.status)) {
+			return;
+		}
+		try {
+			await syncQueue(false);
+		} catch (err: any) {
+			unexpectedError(err);
+		}
 	});
 }

plugins/research_sync/messages.ts

@@ -136,3 +136,76 @@ export class SyncTechnologiesRequest {
 
 	static Response = lib.jsonArray(TechnologySync);
 }
+
+export class QueueUpdateEvent {
+	declare ["constructor"]: typeof QueueUpdateEvent;
+	static type = "event" as const;
+	static src = "instance" as const;
+	static dst = "controller" as const;
+	static plugin = "research_sync" as const;
+
+	constructor(
+		public queue: string[],
+		public player: string | null,
+	) {
+	}
+
+	static jsonSchema = Type.Object({
+		"queue": Type.Array(Type.String()),
+		"player": Type.Union([Type.String(), Type.Null()]),
+	});
+
+	static fromJSON(json: Static<typeof QueueUpdateEvent.jsonSchema>): QueueUpdateEvent {
+		return new this(json.queue, json.player);
+	}
+}
+
+export class QueueEvent {
+	declare ["constructor"]: typeof QueueEvent;
+	static type = "event" as const;
+	static src = "controller" as const;
+	static dst = "instance" as const;
+	static plugin = "research_sync" as const;
+
+	constructor(
+		public queue: string[],
+		public player: string | null,
+		public instance: string | null,
+	) {
+	}
+
+	static jsonSchema = Type.Object({
+		"queue": Type.Array(Type.String()),
+		"player": Type.Union([Type.String(), Type.Null()]),
+		"instance": Type.Union([Type.String(), Type.Null()]),
+	});
+
+	static fromJSON(json: Static<typeof QueueEvent.jsonSchema>): QueueEvent {
+		return new this(json.queue, json.player, json.instance);
+	}
+}
+
+export class QueueSyncRequest {
+	declare ["constructor"]: typeof QueueSyncRequest;
+	static type = "request" as const;
+	static src = "instance" as const;
+	static dst = "controller" as const;
+	static plugin = "research_sync" as const;
+
+	constructor(
+		public queue: string[] | null,
+		public player: string | null,
+	) {
+	}
+
+	static jsonSchema = Type.Object({
+		"queue": Type.Union([Type.Array(Type.String()), Type.Null()]),
+		"player": Type.Union([Type.String(), Type.Null()]),
+	});
+
+	static fromJSON(json: Static<typeof QueueSyncRequest.jsonSchema>): QueueSyncRequest {
+		return new this(json.queue, json.player);
+	}
+
+	static Response = lib.plainJson(Type.Array(Type.String()));
+}

plugins/research_sync/module/sync.lua

@@ -2,6 +2,8 @@ local clusterio_api = require("modules/clusterio/api")
 local compat = require("modules/clusterio/compat")
 
 local v2_tech_progress = compat.version_ge("2.0.0")
+-- on_research_queued and player_index on moved/cancelled arrived in 2.0.56
+local research_queue_api = compat.version_ge("2.0.56")
 
 local sync = {}
 
@@ -38,6 +40,10 @@ end
 --- @class ResearchSyncData
 --- @field technologies table<string, ResearchSyncTechnology>
 --- @field ignore_research_finished boolean
+--- @field follow boolean? nil until the first queue_state sets it: shown ticked, but doesn't send or apply
+--- @field ignore_queue_events boolean
+--- @field queue_dirty boolean
+--- @field queue_player string?
 
 --- @param no_early_return boolean?
 --- @return ResearchSyncData
@@ -53,6 +59,12 @@ local function get_script_data(no_early_return)
 	if research_sync.ignore_research_finished == nil then
 		research_sync.ignore_research_finished = false
 	end
+	if research_sync.ignore_queue_events == nil then
+		research_sync.ignore_queue_events = false
+	end
+	if research_sync.queue_dirty == nil then
+		research_sync.queue_dirty = false
+	end
 
 	local force = game.forces["player"]
 	for _, tech in pairs(force.technologies) do
@@ -143,6 +155,131 @@ sync.on_nth_tick[79] = function(event)
 	end
 end
 
+--- @param force LuaForce
+--- @return string[]
+local function queue_names(force)
+	local names = {}
+	for _, tech in ipairs(force.research_queue) do
+		table.insert(names, tech.name)
+	end
+	return names
+end
+
+--- @param force LuaForce
+--- @param names string[]
+local function write_queue(force, names)
+	local script_data = get_script_data()
+	script_data.ignore_queue_events = true
+	local ok, err = pcall(function() force.research_queue = names end)
+	script_data.ignore_queue_events = false
+	if not ok then
+		log("research_sync: failed to write research queue: " .. tostring(err))
+	end
+end
+
+--- @param names string[]
+--- @return string
+local function format_queue(names)
+	if #names == 0 then
+		return "(empty)"
+	end
+	local tags = {}
+	for _, name in ipairs(names) do
+		table.insert(tags, "[technology=" .. name .. "]")
+	end
+	return table.concat(tags, " ")
+end
+
+--- @param player LuaPlayer
+local function ensure_gui(player)
+	local frame = player.gui.relative["research_sync_queue"]
+	if not frame then
+		frame = player.gui.relative.add {
+			type = "frame",
+			name = "research_sync_queue",
+			caption = "Research sync",
+			anchor = {
+				gui = defines.relative_gui_type.lab_gui,
+				position = defines.relative_gui_position.right,
+			},
+		}
+		frame.add {
+			type = "checkbox",
+			name = "research_sync_follow",
+			caption = "Follow the cluster research queue",
+			tooltip = "On: this server's research queue is replaced by the cluster's, "
+				.. "and changes here change it on every following server.\n"
+				.. "Off: this server's research queue is cleared and it picks its own research.\n"
+				.. "Either way, all research progress is counted, whatever technology is selected.",
+			state = true,
+		}
+	end
+	frame["research_sync_follow"].state = get_script_data().follow ~= false
+end
+
+--- @param event EventData.on_player_created
+sync.events[defines.events.on_player_created] = function(event)
+	if not research_queue_api then
+		return
+	end
+	ensure_gui(game.get_player(event.player_index) --[[@as LuaPlayer]])
+end
+sync.events[defines.events.on_player_joined_game] = sync.events[defines.events.on_player_created]
+
+--- @param event EventData.on_gui_checked_state_changed
+sync.events[defines.events.on_gui_checked_state_changed] = function(event)
+	if not research_queue_api then
+		return
+	end
+	if event.element.name == "research_sync_follow" then
+		local player = game.get_player(event.player_index) --[[@as LuaPlayer]]
+		research_sync.set_follow(event.element.state, player.name)
+	end
+end
+
+-- The version check stays outside here: these defines don't exist before 2.0.56, so indexing them there would error.
+if research_queue_api then
+	--- @param event EventData.on_research_queued|EventData.on_research_moved|EventData.on_research_cancelled
+	local function on_queue_event(event)
+		if event.force.name ~= "player" then
+			return
+		end
+		local script_data = get_script_data()
+		if script_data.follow ~= true or script_data.ignore_queue_events or script_data.ignore_research_finished then
+			return
+		end
+		script_data.queue_dirty = true
+		-- the last event of the tick decides who gets named
+		local player = event.player_index and game.get_player(event.player_index)
+		script_data.queue_player = player and player.name or nil
+	end
+	sync.events[defines.events.on_research_queued] = on_queue_event
+	sync.events[defines.events.on_research_moved] = on_queue_event
+	sync.events[defines.events.on_research_cancelled] = on_queue_event
+end
+
+-- One send per tick however many events fired (a lab raises several at once).
+--- @param event EventData.on_tick
+sync.events[defines.events.on_tick] = function(event)
+	if not research_queue_api then
+		return
+	end
+	local script_data = get_script_data()
+	if not script_data.queue_dirty then
+		return
+	end
+	script_data.queue_dirty = false
+	local player = script_data.queue_player
+	script_data.queue_player = nil
+	if script_data.follow ~= true then
+		return
+	end
+	clusterio_api.send_json("research_sync:queue", {
+		queue = queue_names(game.forces["player"]),
+		player = player,
+	})
+end
+
 research_sync = {}
 function research_sync.dump_technologies()
 	local force = game.forces["player"]
@@ -269,5 +406,94 @@ function research_sync.research_technology(name, level)
 	}
 end
 
+function research_sync.queue_state()
+	if not research_queue_api then
+		rcon.print('{"follow":false}')
+		return
+	end
+	local script_data = get_script_data()
+	local first = script_data.follow == nil
+	if first then
+		script_data.follow = true
+	end
+	rcon.print(compat.table_to_json({
+		follow = script_data.follow,
+		first = first,
+		queue = queue_names(game.forces["player"]),
+	}))
+end
+
+--- @param data string JSON {queue: string[], player?: string, instance?: string}
+function research_sync.apply_queue(data)
+	if not research_queue_api then
+		return
+	end
+	local msg = compat.json_to_table(data) --[[@as table]]
+	local queue = msg.queue or {}
+	if msg.instance then
+		local list = format_queue(queue)
+		if msg.player then
+			game.print(msg.player .. " changed the research queue on " .. msg.instance .. ": " .. list)
+		else
+			game.print("The research queue was changed on " .. msg.instance .. ": " .. list)
+		end
+	end
+
+	if get_script_data().follow ~= true then
+		return
+	end
+	local force = game.forces["player"]
+	-- Names this tile has no technology for (a mod it lacks) would throw.
+	local known = {}
+	for _, name in ipairs(queue) do
+		if force.technologies[name] then
+			table.insert(known, name)
+		end
+	end
+	local current = queue_names(force)
+	local same = #known == #current
+	for i = 1, #known do
+		if known[i] ~= current[i] then
+			same = false
+			break
+		end
+	end
+	if not same then
+		write_queue(force, known)
+	end
+end
+
+--- @param enabled boolean
+--- @param player_name string?
+function research_sync.set_follow(enabled, player_name)
+	if not research_queue_api then
+		return
+	end
+	local script_data = get_script_data()
+	if script_data.follow == enabled then
+		return
+	end
+	script_data.follow = enabled
+	script_data.queue_dirty = false
+	script_data.queue_player = nil
+	for _, player in pairs(game.players) do
+		ensure_gui(player)
+	end
+
+	local state = enabled and "on" or "off"
+	if player_name then
+		game.print(player_name .. " turned " .. state .. " following the cluster research queue")
+	else
+		game.print("Following the cluster research queue was turned " .. state)
+	end
+
+	local force = game.forces["player"]
+	if enabled then
+		clusterio_api.send_json("research_sync:follow", { queue = queue_names(force), player = player_name })
+	else
+		write_queue(force, {})
+	end
+end
+
 
 return sync

plugins/research_sync/test/controller.js

@@ -0,0 +1,173 @@
+import assert from "node:assert/strict";
+import fs from "node:fs/promises";
+import path from "node:path";
+
+import * as lib from "@clusterio/lib";
+import * as mock from "../../../test/mock.js";
+import controllerEntrypoint from "../dist/node/controller.js";
+import { plugin as info } from "../dist/node/index.js";
+import {
+	ContributionEvent, FinishedEvent, QueueEvent, QueueSyncRequest, QueueUpdateEvent,
+	SyncTechnologiesRequest, TechnologySync,
+} from "../dist/node/messages.js";
+
+describe("research_sync controller", function() {
+	const databaseDir = path.join("temp", "test", "research_sync");
+	let controller, sent;
+
+	before(function() {
+		try {
+			lib.registerPluginMessages([info]);
+		} catch (err) {
+			// Already registered by the full test suite
+		}
+	});
+	beforeEach(async function() {
+		await fs.rm(databaseDir, { recursive: true, force: true });
+		await fs.mkdir(databaseDir, { recursive: true });
+		controller = await load();
+	});
+
+	async function load() {
+		const c = await mock.loadControllerPlugin(async (context) => {
+			context.controller.mockConfigEntries.set("controller.database_directory", databaseDir);
+			await controllerEntrypoint(context);
+		}, info);
+		sent = [];
+		c.sendTo = (dst, message) => { sent.push([dst, message]); };
+		c.instances = new Map([[1, { config: { get: () => "rs-1" } }]]);
+		return c;
+	}
+
+	const src = { id: 1 };
+	const update = (queue, player = null) => mock.getHandler(controller, QueueUpdateEvent)(
+		new QueueUpdateEvent(queue, player), src
+	);
+	const sync = (queue = null, player = null) => mock.getHandler(controller, QueueSyncRequest)(
+		new QueueSyncRequest(queue, player), src
+	);
+	const queueEvents = () => sent.filter(([, m]) => m instanceof QueueEvent);
+
+	it("should store and broadcast an update", async function() {
+		await update(["a", "b"], "Bob");
+		assert.equal(sent.length, 1);
+		assert.equal(sent[0][0], "allInstances");
+		assert.ok(sent[0][1] instanceof QueueEvent);
+		assert.deepEqual(sent[0][1].queue, ["a", "b"]);
+		assert.equal(sent[0][1].player, "Bob");
+		assert.equal(sent[0][1].instance, "rs-1");
+		assert.deepEqual(await sync(), ["a", "b"]);
+	});
+	it("should not broadcast an unchanged update", async function() {
+		await update(["a", "b"]);
+		await update(["a", "b"]);
+		assert.equal(queueEvents().length, 1);
+	});
+	it("should broadcast a clear to an empty queue", async function() {
+		await update(["a"], "Bob");
+		sent.length = 0;
+		await update([], null);
+		assert.equal(queueEvents().length, 1);
+		assert.deepEqual(queueEvents()[0][1].queue, []);
+		assert.equal(queueEvents()[0][1].player, null);
+		assert.deepEqual(await sync(null, null), []);
+	});
+	it("should cap the queue at 7", async function() {
+		await update(["1", "2", "3", "4", "5", "6", "7", "8", "9"]);
+		assert.deepEqual(queueEvents()[0][1].queue, ["1", "2", "3", "4", "5", "6", "7"]);
+		assert.deepEqual(await sync(), ["1", "2", "3", "4", "5", "6", "7"]);
+	});
+	it("should seed an empty queue from a sync request", async function() {
+		assert.deepEqual(await sync(["x"], "Bob"), ["x"]);
+		assert.equal(queueEvents().length, 1);
+		assert.deepEqual(queueEvents()[0][1].queue, ["x"]);
+		assert.equal(queueEvents()[0][1].player, "Bob");
+		assert.equal(queueEvents()[0][1].instance, "rs-1");
+	});
+	it("should not replace a non-empty queue from a sync request", async function() {
+		await update(["a"]);
+		sent.length = 0;
+		assert.deepEqual(await sync(["x"]), ["a"]);
+		assert.equal(sent.length, 0);
+		assert.deepEqual(await sync(), ["a"]);
+	});
+	it("should not broadcast when seeding with an empty queue", async function() {
+		assert.deepEqual(await sync([]), []);
+		assert.equal(sent.length, 0);
+	});
+
+	describe("finished technologies", function() {
+		it("should drop one entry when a contribution completes a technology", async function() {
+			await update(["m", "m", "x"]);
+			await mock.getHandler(controller, ContributionEvent)(new ContributionEvent("m", 3, 1), src);
+			assert.deepEqual(await sync(), ["m", "x"]);
+		});
+		it("should drop one entry per advancing FinishedEvent only", async function() {
+			await update(["m", "m", "x"]);
+			const finished = mock.getHandler(controller, FinishedEvent);
+			await finished(new FinishedEvent("m", 3), src);
+			assert.deepEqual(await sync(), ["m", "x"]);
+			await finished(new FinishedEvent("m", 3), src);
+			assert.deepEqual(await sync(), ["m", "x"]);
+		});
+		it("should send the queue silently after a completion behind the current research", async function() {
+			await update(["automation", "m", "m"]);
+			sent.length = 0;
+			await mock.getHandler(controller, ContributionEvent)(new ContributionEvent("m", 3, 1), src);
+			assert.equal(queueEvents().length, 1);
+			assert.deepEqual(queueEvents()[0][1], new QueueEvent(["automation", "m"], null, null));
+		});
+		it("should send the queue after an advancing FinishedEvent only", async function() {
+			await update(["m", "x"]);
+			sent.length = 0;
+			const finished = mock.getHandler(controller, FinishedEvent);
+			await finished(new FinishedEvent("m", 3), src);
+			assert.equal(queueEvents().length, 1);
+			assert.deepEqual(queueEvents()[0][1], new QueueEvent(["x"], null, null));
+			sent.length = 0;
+			await finished(new FinishedEvent("m", 3), src);
+			assert.equal(queueEvents().length, 0);
+		});
+		it("should send no queue when the finished technology is not queued", async function() {
+			await update(["x"]);
+			sent.length = 0;
+			await mock.getHandler(controller, FinishedEvent)(new FinishedEvent("m", 3), src);
+			await mock.getHandler(controller, ContributionEvent)(new ContributionEvent("n", 3, 1), src);
+			assert.equal(queueEvents().length, 0);
+		});
+		it("should drop an entry finished through a technology sync", async function() {
+			await update(["m", "x"]);
+			await mock.getHandler(controller, SyncTechnologiesRequest)(
+				new SyncTechnologiesRequest([new TechnologySync("m", 1, null, true)]), src
+			);
+			assert.deepEqual(await sync(), ["x"]);
+		});
+	});
+
+	describe("technology sync unlocking several levels", function() {
+		const techSync = (name, level, researched) => mock.getHandler(controller, SyncTechnologiesRequest)(
+			new SyncTechnologiesRequest([new TechnologySync(name, level, null, researched)]), src
+		);
+		it("should drop one entry per level unlocked on a known technology", async function() {
+			await techSync("m", 4, true);
+			await update(["m", "m", "m", "x"]);
+			sent.length = 0;
+			await techSync("m", 7, true);
+			assert.deepEqual(await sync(), ["x"]);
+			assert.equal(queueEvents().length, 1);
+			assert.deepEqual(queueEvents()[0][1], new QueueEvent(["x"], null, null));
+		});
+		it("should drop one entry per level unlocked on a new technology", async function() {
+			await update(["m", "m", "x"]);
+			await techSync("m", 2, true);
+			assert.deepEqual(await sync(), ["x"]);
+		});
+	});
+
+	it("should persist the queue across loads", async function() {
+		await update(["a", "b"]);
+		await controller.hooks.save.invoke();
+		controller = await load();
+		assert.deepEqual(await sync(), ["a", "b"]);
+	});
+});

plugins/research_sync/test/instance.js

@@ -0,0 +1,140 @@
+import assert from "node:assert/strict";
+
+import * as lib from "@clusterio/lib";
+import * as mock from "../../../test/mock.js";
+import loadInstancePlugin from "../dist/node/instance.js";
+import { plugin as info } from "../dist/node/index.js";
+import {
+	QueueEvent, QueueSyncRequest, QueueUpdateEvent, SyncTechnologiesRequest,
+} from "../dist/node/messages.js";
+
+describe("research_sync instance", function() {
+	let instance, sent;
+	const clusterQueue = ["a", "b"];
+
+	before(function() {
+		try {
+			lib.registerPluginMessages([info]);
+		} catch (err) {
+			// Already registered by the full test suite
+		}
+	});
+	beforeEach(async function() {
+		({ instance } = await mock.loadInstancePlugin(loadInstancePlugin, info));
+		sent = [];
+		instance.sendTo = async (dst, message) => {
+			sent.push(message);
+			if (message instanceof SyncTechnologiesRequest) {
+				return [];
+			}
+			if (message instanceof QueueSyncRequest) {
+				return clusterQueue;
+			}
+			return undefined;
+		};
+		instance.server.rconCommandResults.set("/sc research_sync.dump_technologies()", "[]");
+	});
+
+	function queueState(state) {
+		instance.server.rconCommandResults.set("/sc research_sync.queue_state()", JSON.stringify(state));
+	}
+	const applyCommand = data => `/sc research_sync.apply_queue("${lib.escapeString(JSON.stringify(data))}")`;
+	const lastCommand = () => instance.server.rconCommands[instance.server.rconCommands.length - 1];
+	const queueSyncs = () => sent.filter(m => m instanceof QueueSyncRequest);
+
+	describe("start", function() {
+		it("should seed the cluster queue on a first start", async function() {
+			queueState({ follow: true, first: true, queue: {} });
+			await instance.hooks.start.invoke();
+			assert.equal(queueSyncs().length, 1);
+			assert.deepEqual(queueSyncs()[0].queue, []);
+			assert.equal(lastCommand(), applyCommand({ queue: clusterQueue }));
+		});
+		it("should only fetch the cluster queue after the first start", async function() {
+			queueState({ follow: true, first: false });
+			await instance.hooks.start.invoke();
+			assert.equal(queueSyncs()[0].queue, null);
+		});
+		it("should not sync the queue when not following", async function() {
+			queueState({ follow: false });
+			await instance.hooks.start.invoke();
+			assert.equal(queueSyncs().length, 0);
+		});
+	});
+
+	describe("QueueEvent", function() {
+		beforeEach(async function() {
+			queueState({ follow: false });
+			await instance.hooks.start.invoke();
+			instance.server.reset();
+		});
+		const handler = () => mock.getHandler(instance, QueueEvent);
+
+		it("should apply the queue with the player", async function() {
+			await handler()(new QueueEvent(["a"], "Bob", "rs-2"));
+			assert.equal(lastCommand(), applyCommand({ queue: ["a"], instance: "rs-2", player: "Bob" }));
+		});
+		it("should leave out a null player", async function() {
+			await handler()(new QueueEvent(["a"], null, "rs-2"));
+			assert.equal(lastCommand(), applyCommand({ queue: ["a"], instance: "rs-2" }));
+		});
+		it("should leave out a null instance", async function() {
+			await handler()(new QueueEvent(["a"], null, null));
+			assert.equal(lastCommand(), applyCommand({ queue: ["a"] }));
+		});
+		it("should ignore the event while stopped", async function() {
+			instance.status = "stopped";
+			await handler()(new QueueEvent(["a"], null, "rs-2"));
+			assert.equal(instance.server.rconCommands.length, 0);
+		});
+	});
+
+	describe("controller connection", function() {
+		beforeEach(async function() {
+			queueState({ follow: true, first: false });
+			await instance.hooks.start.invoke();
+			sent.length = 0;
+			instance.server.reset();
+			queueState({ follow: true, first: false });
+		});
+
+		it("should fetch the cluster queue again on a new session", async function() {
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			assert.equal(queueSyncs().length, 1);
+			assert.equal(queueSyncs()[0].queue, null);
+			assert.equal(lastCommand(), applyCommand({ queue: clusterQueue }));
+		});
+		it("should do nothing on resume", async function() {
+			await instance.hooks.controllerConnectionEvent.invoke("resume");
+			assert.equal(sent.length, 0);
+		});
+		it("should do nothing while stopped", async function() {
+			instance.status = "stopped";
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			assert.equal(sent.length, 0);
+		});
+		it("should not throw when the fetch fails", async function() {
+			instance.sendTo = async () => { throw new Error("gone"); };
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+		});
+		it("should not let the reply overwrite a newer QueueEvent delivered with it", async function() {
+			// As if the reply and the broadcast came in one socket read: the event is handled before the
+			// reply resolves the request's await
+			instance.sendTo = async (dst, message) => {
+				sent.push(message);
+				mock.getHandler(instance, QueueEvent)(new QueueEvent(["new"], null, "B"));
+				return ["old"];
+			};
+			await instance.hooks.controllerConnectionEvent.invoke("connect");
+			assert.equal(lastCommand(), applyCommand({ queue: ["new"], instance: "B" }));
+		});
+	});
+
+	it("should forward queue changes from the game", async function() {
+		instance.server.emit("ipc-research_sync:queue", { queue: {}, player: "Bob" });
+		assert.equal(sent.length, 1);
+		assert.ok(sent[0] instanceof QueueUpdateEvent);
+		assert.deepEqual(sent[0].queue, []);
+		assert.equal(sent[0].player, "Bob");
+	});
+});

plugins/research_sync/test/missing.js

@@ -1,5 +0,0 @@
-// These are libraries that does not have any units test, they are
-// required here in order to track coverage for them.
-
-import "../dist/node/instance.js";
-import "../dist/node/controller.js";