Merge branch 'master' into feature/http-logging

packages/controller/src/BaseConnection.ts

@@ -6,7 +6,6 @@ import type WsServerConnector from "./WsServerConnector";
 
 import * as lib from "@clusterio/lib";
 const { logger } = lib;
-import ControllerRouter from "./ControllerRouter";
 import * as routes from "./routes";
 
 
@@ -24,7 +23,7 @@ export default class BaseConnection extends lib.Link {
 		public _controller: Controller
 	) {
 		super(connector);
-		this.router = new ControllerRouter(this._controller);
+		this.router = _controller.router;
 		for (let [Request, handler] of this._controller._registeredRequests) { this.handle(Request, handler); }
 		for (let [Request, handler] of this._controller._fallbackedRequests) { this.fallbackRequest(Request, handler); }
 		for (let [Event, handler] of this._controller._registeredEvents) { this.handle(Event, handler); }

packages/controller/src/BaseControllerPlugin.ts

@@ -13,8 +13,8 @@ import type HostConnection from "./HostConnection";
  * Controller plugins are subclasses of this class which get instantiated by
  * the controller on startup when the plugin is enabled in the config.
  * To be discovered the class must be exported under the name `ControllerPlugin`
- * in the module specified by the `controllerEntrypoint` in the plugin's info.js
- * file.
+ * in the module specified by the `controllerEntrypoint` in the plugin's
+ * `plugin` export.
  */
 export default class BaseControllerPlugin {
 	/**

packages/controller/src/Controller.ts

@@ -27,6 +27,7 @@ import WsServer from "./WsServer";
 import HostConnection from "./HostConnection";
 import HostInfo from "./HostInfo";
 import BaseControllerPlugin from "./BaseControllerPlugin";
+import ControllerRouter from "./ControllerRouter";
 
 const endpointDurationSummary = new Summary(
 	"clusterio_controller_http_endpoint_duration_seconds",
@@ -75,6 +76,7 @@ export default class Controller {
 
 	/** WebSocket server */
 	wsServer: WsServer;
+	router = new ControllerRouter(this);
 	trustedProxies: BlockList;
 	debugEvents: events.EventEmitter = new events.EventEmitter();
 	private _events: events.EventEmitter = new events.EventEmitter();
@@ -273,7 +275,7 @@ export default class Controller {
 		Controller.addAppRoutes(this.app, this.pluginInfos);
 
 		if (!args.dev) {
-			let manifestPath = path.join(__dirname, "..", "web", "manifest.json");
+			let manifestPath = path.join(__dirname, "..", "..", "web", "manifest.json");
 
 			let manifest = await Controller.loadJsonObject(manifestPath);
 			if (!manifest["main.js"]) {
@@ -352,7 +354,7 @@ export default class Controller {
 
 		if (args.dev) {
 			// eslint-disable-next-line node/no-missing-require
-			webpackConfigs.push(require("../../webpack.config")({})); // Path outside of build
+			webpackConfigs.push(require("../../../webpack.config")({})); // Path outside of build
 		}
 		if (args.devPlugin) {
 			let devPlugins = new Map();
@@ -617,7 +619,7 @@ export default class Controller {
 			const host = HostInfo.fromJSON(json);
 			if (host.connected) {
 				host.connected = false;
-				host.updatedAt = Date.now();
+				host.updatedAtMs = Date.now();
 			}
 			hosts.set(id, host);
 		}
@@ -638,13 +640,13 @@ export default class Controller {
 			) as Static<typeof InstanceInfo.jsonSchema>[];
 			for (let json of serialized) {
 				if (!json.config) { // migrate: from pre Alpha 14 format.
-					json = { config: json as any, status: "running" }; // Use running to force updatedAt
+					json = { config: json as any, status: "running" }; // Use running to force updatedAtMs
 				}
 				const instance = InstanceInfo.fromJSON(json, "controller");
 				const status = instance.config.get("instance.assigned_host") === null ? "unassigned" : "unknown";
 				if (instance.status !== status) {
 					instance.status = status;
-					instance.updatedAt = Date.now();
+					instance.updatedAtMs = Date.now();
 				}
 				instances.set(instance.id, instance);
 			}
@@ -771,7 +773,7 @@ export default class Controller {
 		// Set folder to serve static content from (the website)
 		const staticOptions = { immutable: true, maxAge: 1000 * 86400 * 365 };
 		app.use("/static",
-			express.static(path.join(__dirname, "..", "..", "dist", "web", "static"), staticOptions)
+			express.static(path.join(__dirname, "..", "..", "web", "static"), staticOptions)
 		);
 		app.use("/static", express.static("static", staticOptions)); // Used for data export files
 
@@ -795,7 +797,7 @@ export default class Controller {
 
 	async handleSystemInfoSubscription(request: lib.SubscriptionRequest) {
 		const systems = [...this.systems.values()].filter(
-			metric => metric.updatedAt > request.lastRequestTime,
+			metric => metric.updatedAtMs > request.lastRequestTimeMs,
 		);
 		return systems.length ? new lib.SystemInfoUpdateEvent(systems) : null;
 	}
@@ -816,7 +818,7 @@ export default class Controller {
 	hostsUpdated(hosts: HostInfo[]) {
 		const now = Date.now();
 		for (const host of hosts) {
-			host.updatedAt = now;
+			host.updatedAtMs = now;
 		}
 		this.hostsDirty = true;
 		let updates = hosts.map(host => host.toHostDetails());
@@ -825,7 +827,7 @@ export default class Controller {
 
 	async handleHostSubscription(request: lib.SubscriptionRequest) {
 		const hosts = [...this.hosts.values()].filter(
-			host => host.updatedAt > request.lastRequestTime,
+			host => host.updatedAtMs > request.lastRequestTimeMs,
 		).map(host => host.toHostDetails());
 		return hosts.length ? new lib.HostUpdatesEvent(hosts) : null;
 	}
@@ -910,9 +912,11 @@ export default class Controller {
 	 */
 	clearSavesOfInstance(instanceId: number) {
 		const updates = [];
+		const now = Date.now();
 		for (const [id, save] of this.saves) {
 			if (save.instanceId === instanceId) {
 				save.isDeleted = true;
+				save.updatedAtMs = now;
 				updates.push(save);
 				this.saves.delete(id);
 			}
@@ -973,8 +977,9 @@ export default class Controller {
 			);
 		} else {
 			instance.status = "unassigned";
-			this.instanceDetailsUpdated([instance]);
 		}
+		instance.updatedAtMs = Date.now();
+		this.instanceDetailsUpdated([instance]);
 	}
 
 	/**
@@ -995,10 +1000,11 @@ export default class Controller {
 		this.instancesDirty = true;
 
 		this.clearSavesOfInstance(instanceId);
+		this.userManager.clearStatsOfInstance(instanceId);
 
 		let prev = instance.status;
 		instance.status = "deleted";
-		instance.updatedAt = Date.now();
+		instance.updatedAtMs = Date.now();
 		this.instanceDetailsUpdated([instance]);
 		await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance, prev);
 	}
@@ -1026,7 +1032,7 @@ export default class Controller {
 
 	addInstanceHooks(instance: InstanceInfo) {
 		instance.config.on("fieldChanged", (field: string, curr: any, prev: any) => {
-			instance.updatedAt = Date.now();
+			instance.updatedAtMs = Date.now();
 			if (field === "instance.name") {
 				this.instanceDetailsUpdated([instance]);
 			}
@@ -1043,7 +1049,7 @@ export default class Controller {
 
 	async handleInstanceDetailsSubscription(request: lib.SubscriptionRequest) {
 		const instances = [...this.instances.values()].filter(
-			instance => instance.updatedAt > request.lastRequestTime,
+			instance => instance.updatedAtMs > request.lastRequestTimeMs,
 		).map(instance => instance.toInstanceDetails());
 		return instances.length ? new lib.InstanceDetailsUpdatesEvent(instances) : null;
 	}
@@ -1054,7 +1060,7 @@ export default class Controller {
 	}
 
 	async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) {
-		const saves = [...this.saves.values()].filter(save => save.updatedAt > request.lastRequestTime);
+		const saves = [...this.saves.values()].filter(save => save.updatedAtMs > request.lastRequestTimeMs);
 		return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null;
 	}
 
@@ -1062,7 +1068,7 @@ export default class Controller {
 	modPacksUpdated(modPacks: lib.ModPack[]) {
 		const now = Date.now();
 		for (const modPack of modPacks) {
-			modPack.updatedAt = now;
+			modPack.updatedAtMs = now;
 		}
 		this.modPacksDirty = true;
 		this.subscriptions.broadcast(new lib.ModPackUpdatesEvent(modPacks));
@@ -1071,20 +1077,20 @@ export default class Controller {
 
 	async handleModPackSubscription(request: lib.SubscriptionRequest) {
 		const modPacks = [...this.modPacks.values()].filter(
-			modPack => modPack.updatedAt > request.lastRequestTime,
+			modPack => modPack.updatedAtMs > request.lastRequestTimeMs,
 		);
 		return modPacks.length ? new lib.ModPackUpdatesEvent(modPacks) : null;
 	}
 
 	modsUpdated(mods: lib.ModInfo[]) {
-		// ModStore sets updatedAt for mods
+		// ModStore sets updatedAtMs for mods
 		this.subscriptions.broadcast(new lib.ModUpdatesEvent(mods));
 		lib.invokeHook(this.plugins, "onModsUpdated", mods);
 	}
 
 	async handleModSubscription(request: lib.SubscriptionRequest) {
 		const mods = [...this.modStore.files.values()].filter(
-			mod => mod.updatedAt > request.lastRequestTime,
+			mod => mod.updatedAtMs > request.lastRequestTimeMs,
 		);
 		return mods.length ? new lib.ModUpdatesEvent(mods) : null;
 	}
@@ -1092,7 +1098,7 @@ export default class Controller {
 	usersUpdated(users: ControllerUser[]) {
 		const now = Date.now();
 		for (const user of users) {
-			user.updatedAt = now;
+			user.updatedAtMs = now;
 		}
 		this.userManager.dirty = true;
 		this.subscriptions.broadcast(new lib.UserUpdatesEvent(users));
@@ -1100,7 +1106,7 @@ export default class Controller {
 
 	async handleUserSubscription(request: lib.SubscriptionRequest) {
 		const users = [...this.userManager.users.values()].filter(
-			user => user.updatedAt > request.lastRequestTime,
+			user => user.updatedAtMs > request.lastRequestTimeMs,
 		);
 		return users.length ? new lib.UserUpdatesEvent(users) : null;
 	}
@@ -1257,7 +1263,7 @@ export default class Controller {
 				mainBundle = stats.toJson().assetsByChunkName["main"];
 			}
 
-			fs.readFile(path.join(__dirname, "..", "..", "web", "index.html"), "utf8").then((content) => {
+			fs.readFile(path.join(__dirname, "..", "..", "..", "web", "index.html"), "utf8").then((content) => {
 				res.type("text/html");
 				res.send(content
 					.replace(/__CLUSTER_NAME__/g, res.app.locals.controller.config.get("controller.name"))
@@ -1409,7 +1415,11 @@ export default class Controller {
 				}
 
 			} else if (dst.id === lib.Address.instance || dst.id === lib.Address.host) {
+				const plugin = event.constructor.plugin;
 				for (let hostConnection of this.wsServer.hostConnections.values()) {
+					if (plugin && !hostConnection.plugins.has(plugin)) {
+						continue;
+					}
 					hostConnection.sendEvent(event, dst);
 				}
 

packages/controller/src/ControllerRouter.ts

@@ -7,8 +7,6 @@ import ControlConnection from "./ControlConnection";
 const { logger } = lib;
 
 export default class ControllerRouter {
-	pendingRequests: Map<any, any> = new Map();
-
 	constructor(
 		public controller: Controller
 	) {}
@@ -16,12 +14,14 @@ export default class ControllerRouter {
 	/**
 	 * @param origin - Source link of the message.
 	 * @param message - Link the message originated from.
+	 * @param entry - message entry if message is a request or an event.
 	 * @param hasFallback - true if fallback handling is available.
 	 * @returns true if the message was handled
 	 */
 	forwardMessage(
 		origin: ControlConnection | HostConnection,
 		message: lib.MessageRoutable,
+		entry: lib.RequestEntry | lib.EventEntry | undefined,
 		hasFallback: boolean
 	): boolean {
 		if (!["request", "response", "responseError", "event"].includes(message.type)) {
@@ -32,7 +32,7 @@ export default class ControllerRouter {
 		let nextHop: ControlConnection | HostConnection | undefined;
 		let msg: string = "";
 		if (dst.type === lib.Address.broadcast) {
-			this.broadcastMessage(origin, message);
+			this.broadcastMessage(origin, message, entry as lib.EventEntry);
 			return true;
 		} else if (dst.type === lib.Address.host) {
 			nextHop = this.controller.wsServer.hostConnections.get(dst.id);
@@ -84,7 +84,7 @@ export default class ControllerRouter {
 			if (message.type === "request") {
 				nextHop.forwardRequest(message as lib.MessageRequest, origin);
 			} else {
-				nextHop.connector.send(message);
+				nextHop.connector.forward(message);
 			}
 		} else {
 			this.warnUnrouted(message, msg);
@@ -96,20 +96,24 @@ export default class ControllerRouter {
 	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}`);
-		} else if (dst.id === lib.Address.host || dst.id === lib.Address.instance) {
+			return;
+		}
+		const plugin = entry.Event.plugin;
+		if (dst.id === lib.Address.host || dst.id === lib.Address.instance) {
 			for (let hostConnection of this.controller.wsServer.hostConnections.values()) {
-				if (hostConnection !== origin) {
-					hostConnection.connector.send(message);
+				if (hostConnection !== origin && (!plugin || hostConnection.plugins.has(plugin))) {
+					hostConnection.connector.forward(message);
 				}
 			}
 		} else if (dst.id === lib.Address.control) {
 			for (let controlConnection of this.controller.wsServer.controlConnections.values()) {
 				if (controlConnection !== origin) {
-					controlConnection.connector.send(message);
+					controlConnection.connector.forward(message);
 				}
 			}
 		} else {
@@ -126,5 +130,3 @@ export default class ControllerRouter {
 		logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`);
 	}
 }
-
-module.exports = ControllerRouter;

packages/controller/src/ControllerUser.ts

@@ -35,7 +35,7 @@ export default class ControllerUser extends User implements IControllerUser {
 			json.is_banned,
 			json.is_whitelisted,
 			json.ban_reason,
-			json.updated_at,
+			json.updated_at_ms,
 			json.is_deleted,
 			playerStats,
 			instanceStats,

packages/controller/src/HostConnection.ts

@@ -57,6 +57,7 @@ export default class HostConnection extends BaseConnection {
 		}
 
 		this.connector.on("close", () => {
+			const now = Date.now();
 			// Update status to unknown for instances on this host.
 			const instances: InstanceInfo[] = [];
 			for (let instance of this._controller.instances.values()) {
@@ -67,6 +68,7 @@ export default class HostConnection extends BaseConnection {
 				instances.push(instance);
 				let prev = instance.status;
 				instance.status = "unknown";
+				instance.updatedAtMs = now;
 				lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
 			}
 			this._controller.instanceDetailsUpdated(instances);
@@ -176,14 +178,17 @@ export default class HostConnection extends BaseConnection {
 		// changed.  Hosts also send status updates on assignInstance which
 		// for hacky reason is also used to push config changes and
 		// restablish status after a connection loss.
-		if (instance.status === request.status) {
+		if (
+			instance.status === request.status
+			&& instance.gamePort === request.gamePort
+		) {
 			return;
 		}
 
 		let prev = instance.status;
 		instance.status = request.status;
 		instance.gamePort = request.gamePort;
-		instance.updatedAt = Date.now();
+		instance.updatedAtMs = Date.now();
 		logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
 		this._controller.instanceDetailsUpdated([instance]);
 		await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
@@ -220,7 +225,7 @@ export default class HostConnection extends BaseConnection {
 					let prev = controllerInstance.status;
 					controllerInstance.status = instanceData.status;
 					controllerInstance.gamePort = instanceData.gamePort;
-					controllerInstance.updatedAt = Date.now();
+					controllerInstance.updatedAtMs = Date.now();
 					logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`);
 					instanceUpdates.push(controllerInstance);
 					await lib.invokeHook(
@@ -281,7 +286,7 @@ export default class HostConnection extends BaseConnection {
 			if (existingSave && save.equals(existingSave)) {
 				continue;
 			}
-			save.updatedAt = now;
+			save.updatedAtMs = now;
 			this._controller.saves.set(save.id, save);
 			updates.push(save);
 		}
@@ -290,7 +295,7 @@ export default class HostConnection extends BaseConnection {
 			for (const [id, save] of this._controller.saves) {
 				if (save.instanceId === event.instanceId && !updatedSaves.has(id)) {
 					save.isDeleted = true;
-					save.updatedAt = now;
+					save.updatedAtMs = now;
 					updates.push(save);
 					this._controller.saves.delete(id);
 				}

packages/controller/src/HostInfo.ts

@@ -8,7 +8,7 @@ export default class HostInfo {
 	constructor(
 		/** Id of this host */
 		public id: number,
-		/** Version this host last connected with */
+		/** Name of this host */
 		public name: string,
 		/** Version this host last connected with */
 		public version: string,
@@ -23,7 +23,7 @@ export default class HostInfo {
 		/** Unix timestamp in seconds host token must be issued after to be valid */
 		public tokenValidAfter: number = 0,
 		/** Millisecond Unix timestamp this entry was last updated at */
-		public updatedAt: number = 0,
+		public updatedAtMs: number = 0,
 		/** True if this host has been deleted */
 		public isDeleted: boolean = false,
 	) { }
@@ -37,7 +37,7 @@ export default class HostInfo {
 		"remote_address": Type.String(),
 		"public_address": Type.String(),
 		"token_valid_after": Type.Number(),
-		"updated_at": Type.Number(),
+		"updated_at_ms": Type.Number(),
 		"is_deleted": Type.Boolean(),
 	});
 
@@ -51,7 +51,7 @@ export default class HostInfo {
 			json.remote_address,
 			json.public_address,
 			json.token_valid_after,
-			json.updated_at,
+			json.updated_at_ms,
 			json.is_deleted,
 		);
 	}
@@ -66,7 +66,7 @@ export default class HostInfo {
 			remote_address: this.remoteAddress,
 			public_address: this.publicAddress,
 			token_valid_after: this.tokenValidAfter,
-			updated_at: this.updatedAt,
+			updated_at_ms: this.updatedAtMs,
 			is_deleted: this.isDeleted,
 		};
 	}
@@ -80,7 +80,7 @@ export default class HostInfo {
 			this.remoteAddress,
 			this.publicAddress,
 			this.tokenValidAfter,
-			this.updatedAt,
+			this.updatedAtMs,
 			this.isDeleted,
 		);
 	}

packages/controller/src/HttpCloser.ts

@@ -67,10 +67,10 @@ export default class HttpCloser {
 	 * If there's requests in progress that still haven't been completed
 	 * after the timeout value passed they will be aborted.
 	 *
-	 * @param timeout -
+	 * @param timeoutMs -
 	 *     Time in ms to wait for connections to close.
 	 */
-	async close(timeout:number = 5000): Promise<void> {
+	async close(timeoutMs: number = 5000): Promise<void> {
 		let activeSockets = new Set();
 		for (let response of this._responses) {
 			if (!response.headersSent) {
@@ -97,7 +97,7 @@ export default class HttpCloser {
 					// Node.js, see nodejs/node#27428.
 					socket.destroy();
 				}
-			}, timeout);
+			}, timeoutMs);
 			this._server.close(err => {
 				clearTimeout(timer);
 				if (err) {

packages/controller/src/InstanceInfo.ts

@@ -11,7 +11,7 @@ export default class InstanceInfo {
 		public config: lib.InstanceConfig,
 		public status: lib.InstanceStatus,
 		public gamePort?: number,
-		public updatedAt = 0,
+		public updatedAtMs = 0,
 	) {
 		this.config = config;
 		this.status = status;
@@ -21,7 +21,7 @@ export default class InstanceInfo {
 		"config": lib.InstanceConfig.jsonSchema,
 		"status": lib.InstanceStatus,
 		"gamePort": Type.Optional(Type.Number()),
-		"updatedAt": Type.Optional(Type.Number()),
+		"updatedAtMs": Type.Optional(Type.Number()),
 	});
 
 	static fromJSON(json: Static<typeof this.jsonSchema>, location: lib.ConfigLocation) {
@@ -29,7 +29,7 @@ export default class InstanceInfo {
 			lib.InstanceConfig.fromJSON(json.config, location),
 			json.status,
 			json.gamePort,
-			json.updatedAt,
+			json.updatedAtMs,
 		);
 	}
 
@@ -40,7 +40,7 @@ export default class InstanceInfo {
 			this.config.get("instance.assigned_host") ?? undefined,
 			this.gamePort,
 			this.status,
-			this.updatedAt,
+			this.updatedAtMs,
 		);
 	}
 

packages/controller/src/UserManager.ts

@@ -113,4 +113,13 @@ export default class UserManager {
 			this.onlineUsers.delete(user);
 		}
 	}
+
+	clearStatsOfInstance(instanceId: number) {
+		for (const user of this.users.values()) {
+			this.notifyLeave(user, instanceId);
+			user.instanceStats.delete(instanceId);
+			user.recalculatePlayerStats();
+		}
+		this.dirty = true;
+	}
 }

packages/controller/src/routes.ts

@@ -56,7 +56,7 @@ async function getMetrics(req: Request, res: Response, next: any) {
 	}
 
 	let requests: Promise<InstanceType<typeof lib.HostMetricsRequest["Response"]> | null>[] = [];
-	let timeout = controller.config.get("controller.metrics_timeout") * 1000;
+	let timeoutMs = controller.config.get("controller.metrics_timeout") * 1000;
 	for (let [hostId, hostConnection] of controller.wsServer.hostConnections) {
 		if (!hostConnection.connected) {
 			continue;
@@ -68,7 +68,7 @@ async function getMetrics(req: Request, res: Response, next: any) {
 				}
 				return null;
 			}),
-			timeout, null
+			timeoutMs, null
 		));
 	}