diff --git a/.eslintrc.js b/.eslintrc.js index b316b1ed..0764886b 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -350,8 +350,9 @@ module.exports = { "node/exports-style": "off", "node/file-extension-in-import": "off", "node/global-require": "off", - "node/handle-callback-err": "error", - "node/no-callback-literal": "error", + // Node.js callback passing style of (err, value) is not used in this project + "node/handle-callback-err": "off", + "node/no-callback-literal": "off", "node/no-deprecated-api": "error", "node/no-exports-assign": "error", "node/no-extraneous-import": "off", diff --git a/packages/controller/src/BaseConnection.ts b/packages/controller/src/BaseConnection.ts index 91f2a5ab..493ebbde 100644 --- a/packages/controller/src/BaseConnection.ts +++ b/packages/controller/src/BaseConnection.ts @@ -33,6 +33,11 @@ export default class BaseConnection extends lib.Link { this.handle(lib.ModPackGetRequest, this.handleModPackGetRequest.bind(this)); this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this)); this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this)); + + this.handle(lib.SubscriptionRequest, this.handleSubscriptionRequest.bind(this)); + this.connector.on("close", () => { + this._controller.subscriptions.unsubscribe(this); + }); } async disconnect(code: number, reason: string) { @@ -98,4 +103,8 @@ export default class BaseConnection extends lib.Link { return stream.id; } + + async handleSubscriptionRequest(request: lib.SubscriptionRequest, src: lib.Address, dst: lib.Address) { + return await this._controller.subscriptions.handleRequest(this, request, src, dst); + } } diff --git a/packages/controller/src/BaseControllerPlugin.ts b/packages/controller/src/BaseControllerPlugin.ts index 659001d2..9aec578b 100644 --- a/packages/controller/src/BaseControllerPlugin.ts +++ b/packages/controller/src/BaseControllerPlugin.ts @@ -229,29 +229,29 @@ export default class BaseControllerPlugin { async onPrepareHostDisconnect(connection: HostConnection) { } /** - * Called when a mod pack is updated + * Called when one or more mod packs are updated * - * Invoked when a mod pack has been added, updated or deleted in the - * cluster. + * Invoked when one or more mod pack have been added, updated or deleted + * in the cluster. * - * If the mod pack has been deleted its `.isDeleted` property will be + * If a mod pack has been deleted its `.isDeleted` property will be * true. * - * @param modPack - Mod pack that updated. + * @param modPacks - Mod packs that updated. */ - async onModPackUpdated(modPack: ModPack) { } + async onModPacksUpdated(modPacks: ModPack[]) { } /** - * Called when a mod stored on the controller is updated + * Called when one or more mod stored on the controller are updated * - * Invoked when a mod has been added, updated or deleted from the pool - * of shared mods stored on the cluster. + * Invoked when one or more mods have been added, updated or deleted + * from the pool of shared mods stored on the cluster. * * If a mod has been deleted its `.isDeleted` property will be true. * - * @param mod - Mod that updated. + * @param mods - Mods that updated. */ - async onModUpdated(mod: ModInfo) { } + async onModsUpdated(mods: ModInfo[]) { } /** * Called when a player joins or leaves an instance diff --git a/packages/controller/src/ControlConnection.ts b/packages/controller/src/ControlConnection.ts index abd955b7..b1740ea5 100644 --- a/packages/controller/src/ControlConnection.ts +++ b/packages/controller/src/ControlConnection.ts @@ -1,5 +1,4 @@ import type WsServerConnector from "./WsServerConnector"; -import type { HostInfo } from "./HostConnection"; import events from "events"; @@ -86,6 +85,7 @@ export default class ControlConnection extends BaseConnection { this.handle(lib.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this)); this.handle(lib.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this)); this.handle(lib.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this)); + this.handle(lib.InstanceSaveDetailsListRequest, this.handleInstanceSaveDetailsListRequest.bind(this)); this.handle(lib.InstanceRenameSaveRequest, controller.sendRequestToHostByInstanceId.bind(controller)); this.handle(lib.InstanceCopySaveRequest, controller.sendRequestToHostByInstanceId.bind(controller)); this.handle(lib.InstanceDeleteSaveRequest, controller.sendRequestToHostByInstanceId.bind(controller)); @@ -188,30 +188,18 @@ export default class ControlConnection extends BaseConnection { throw new Error(`Unknown host id (${request.hostId})`); } - host.token_valid_after = Math.floor(Date.now() / 1000); + host.tokenValidAfter = Math.floor(Date.now() / 1000); const hostConnection = this._controller.wsServer.hostConnections.get(request.hostId); if (hostConnection) { hostConnection.connector.terminate(); } - this._controller.hostUpdated(host as HostInfo); + this._controller.hostsUpdated([host]); } async handleHostListRequest(): Promise { - let list = []; - for (let host of this._controller.hosts.values()) { - list.push(new lib.HostDetails( - host.agent, - host.version, - host.name, - host.id, - this._controller.wsServer.hostConnections.has(host.id), - host.public_address, - host.token_valid_after, - )); - } - return list; + return [...this._controller.hosts.values()].map(host => host.toHostDetails()); } async handleHostGenerateTokenRequest(message: lib.HostGenerateTokenRequest) { @@ -241,49 +229,11 @@ export default class ControlConnection extends BaseConnection { } async handleInstanceDetailsGetRequest(request: lib.InstanceDetailsGetRequest) { - let instance = this._controller.getRequestInstance(request.instanceId); - - let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); - if (assigned_host === null) { - assigned_host = undefined; - } - - let game_port: number|null|undefined = instance.game_port; - if (game_port === null) { - game_port = undefined; - } - - return new lib.InstanceDetails( - instance.config.get("instance.name"), - instance.id, - assigned_host, - game_port, - instance.status, - ); + return this._controller.getRequestInstance(request.instanceId).toInstanceDetails(); } async handleInstanceDetailsListRequest() { - let list = []; - for (let instance of this._controller.instances.values()) { - let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); - if (assigned_host === null) { - assigned_host = undefined; - } - - let game_port: number|null|undefined = instance.game_port; - if (game_port === null) { - game_port = undefined; - } - - list.push(new lib.InstanceDetails( - instance.config.get("instance.name"), - instance.id, - assigned_host, - game_port, - instance.status, - )); - } - return list; + return [...this._controller.instances.values()].map(instance => instance.toInstanceDetails()); } // XXX should probably add a hook for host reuqests? @@ -328,6 +278,10 @@ export default class ControlConnection extends BaseConnection { await this._controller.instanceAssign(request.instanceId, request.hostId); } + async handleInstanceSaveDetailsListRequest() { + return [...this._controller.saves.values()]; + } + async handleInstanceDownloadSaveRequest(request: lib.InstanceDownloadSaveRequest) { let { instanceId, name } = request; let stream = await routes.createProxyStream(this._controller.app); @@ -434,7 +388,7 @@ export default class ControlConnection extends BaseConnection { throw new lib.RequestError(`Mod pack with ID ${modPack.id} already exist`); } this._controller.modPacks.set(modPack.id, modPack); - this._controller.modPackUpdated(modPack); + this._controller.modPacksUpdated([modPack]); } async handleModPackUpdateRequest(request: lib.ModPackUpdateRequest) { @@ -443,7 +397,7 @@ export default class ControlConnection extends BaseConnection { throw new lib.RequestError(`Mod pack with ID ${modPack.id} does not exist`); } this._controller.modPacks.set(modPack.id, modPack); - this._controller.modPackUpdated(modPack); + this._controller.modPacksUpdated([modPack]); } async handleModPackDeleteRequest(request: lib.ModPackDeleteRequest) { @@ -454,7 +408,7 @@ export default class ControlConnection extends BaseConnection { } modPack.isDeleted = true; this._controller.modPacks.delete(id); - this._controller.modPackUpdated(modPack); + this._controller.modPacksUpdated([modPack]); } async handleModGetRequest(request: lib.ModGetRequest) { @@ -680,7 +634,7 @@ export default class ControlConnection extends BaseConnection { async handleUserCreateRequest(request: lib.UserCreateRequest) { let user = this._controller.userManager.createUser(request.name); - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); } async handleUserRevokeTokenRequest(request: lib.UserRevokeTokenRequest) { @@ -698,7 +652,7 @@ export default class ControlConnection extends BaseConnection { controlConnection.connector.terminate(); } } - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); } async handleUserUpdateRolesRequest(request: lib.UserUpdateRolesRequest) { @@ -715,7 +669,7 @@ export default class ControlConnection extends BaseConnection { user.roleIds = new Set(request.roles); this._controller.userPermissionsUpdated(user); - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); } async handleUserSetAdminRequest(request: lib.UserSetAdminRequest) { @@ -731,7 +685,7 @@ export default class ControlConnection extends BaseConnection { } user.isAdmin = admin; - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, admin)); } @@ -749,7 +703,7 @@ export default class ControlConnection extends BaseConnection { user.isBanned = banned; user.banReason = reason; - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); this._controller.sendTo("allInstances", new lib.InstanceBanlistUpdateEvent(name, banned, reason)); } @@ -766,7 +720,7 @@ export default class ControlConnection extends BaseConnection { } user.isWhitelisted = whitelisted; - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); this._controller.sendTo("allInstances", new lib.InstanceWhitelistUpdateEvent(name, whitelisted)); } @@ -779,7 +733,7 @@ export default class ControlConnection extends BaseConnection { user.isDeleted = true; this._controller.userManager.users.delete(name); - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); if (user.isAdmin) { this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, false)); diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index be26d4b0..28750608 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -2,6 +2,7 @@ import type winston from "winston"; import type { AddressInfo } from "net"; import type { ControllerArgs } from "../controller"; import express, { type Request, type Response, type NextFunction, type Application } from "express"; +import type { Static } from "@sinclair/typebox"; import compression from "compression"; import events, { EventEmitter } from "events"; @@ -16,13 +17,14 @@ import * as lib from "@clusterio/lib"; const { logger, Summary, Gauge } = lib; import HttpCloser from "./HttpCloser"; -import InstanceInfo, { InstanceStatus } from "./InstanceInfo"; +import InstanceInfo from "./InstanceInfo"; import * as metrics from "./metrics"; import * as routes from "./routes"; import ControllerUser from "./ControllerUser"; import UserManager from "./UserManager"; import WsServer from "./WsServer"; -import HostConnection, { type HostInfo } from "./HostConnection"; +import HostConnection from "./HostConnection"; +import HostInfo from "./HostInfo"; import BaseControllerPlugin from "./BaseControllerPlugin"; const endpointDurationSummary = new Summary( @@ -55,6 +57,8 @@ export default class Controller { hostsDirty = false; /** True if {@link Controller.instances} has changed since last time it was saved */ instancesDirty = false; + /** True if {@link Controller.saves} has changed since last time it was saved */ + savesDirty = false; /** True if {@link Controller.modPacks} has changed since last time it was saved */ modPacksDirty = false; @@ -72,7 +76,7 @@ export default class Controller { private _events: events.EventEmitter = new events.EventEmitter(); /** Event subscription controller */ - subscriptions: lib.SubscriptionController; + subscriptions = new lib.SubscriptionController(); // Possible states are new, starting, running, stopping, stopped private _state: string = "new"; @@ -96,6 +100,7 @@ export default class Controller { const hosts = await Controller.loadHosts(path.join(databaseDirectory, "hosts.json")); const instances = await Controller.loadInstances(path.join(databaseDirectory, "instances.json")); + const saves = await Controller.loadSaves(path.join(databaseDirectory, "saves.json")); const modPacks = await Controller.loadModPacks(path.join(databaseDirectory, "mod-packs.json")); const userManager = new UserManager(config); await userManager.load(path.join(databaseDirectory, "users.json")); @@ -107,6 +112,7 @@ export default class Controller { return [ hosts, instances, + saves, modPacks, modStore, userManager, @@ -123,6 +129,8 @@ export default class Controller { public hosts = new Map(), /** Mapping of instance id to instance info */ public instances = new Map(), + /** Mapping of save id to save details */ + public saves = new Map(), /** Mapping of mod pack id to mod pack */ public modPacks = new Map(), /** Mods stored on the controller */ @@ -140,19 +148,20 @@ export default class Controller { this.app.locals.streams = new Map(); this.wsServer = new WsServer(this); - this.subscriptions = new lib.SubscriptionController(this); this.modStore.on("change", mod => { - this.modUpdated(mod); + this.modsUpdated([mod]); }); // Handle subscriptions for all internal properties - this.subscriptions.handle(lib.HostUpdateEvent); - this.subscriptions.handle(lib.InstanceDetailsUpdateEvent); - this.subscriptions.handle(lib.InstanceSaveListUpdateEvent); - this.subscriptions.handle(lib.ModPackUpdateEvent); - this.subscriptions.handle(lib.ModUpdateEvent); - this.subscriptions.handle(lib.UserUpdateEvent); + this.subscriptions.handle(lib.HostUpdatesEvent, this.handleHostSubscription.bind(this)); + this.subscriptions.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsSubscription.bind(this)); + this.subscriptions.handle( + lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsSubscription.bind(this) + ); + this.subscriptions.handle(lib.ModPackUpdatesEvent, this.handleModPackSubscription.bind(this)); + this.subscriptions.handle(lib.ModUpdatesEvent, this.handleModSubscription.bind(this)); + this.subscriptions.handle(lib.UserUpdatesEvent, this.handleUserSubscription.bind(this)); } async start(args: ControllerArgs) { @@ -426,6 +435,11 @@ export default class Controller { await Controller.saveInstances(path.join(databaseDirectory, "instances.json"), this.instances); } + if (this.savesDirty) { + this.savesDirty = false; + await Controller.saveSaves(path.join(databaseDirectory, "saves.json"), this.saves); + } + if (this.modPacksDirty) { this.modPacksDirty = false; await Controller.saveModPacks(path.join(databaseDirectory, "mod-packs.json"), this.modPacks); @@ -439,7 +453,7 @@ export default class Controller { } static async loadHosts(filePath: string): Promise> { - let serialized: [number, HostInfo][]; + let serialized: [number, Static][]; try { serialized = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" })); @@ -450,12 +464,21 @@ export default class Controller { return new Map(); } - // TODO: Remove after release. + // TODO: migrate remove after release. if (serialized.length && !(serialized[0] instanceof Array)) { return new Map(); // Discard old format. } - return new Map(serialized); + const hosts = new Map(); + for (let [id, json] of serialized) { + const host = HostInfo.fromJSON(json); + if (host.connected) { + host.connected = false; + host.updatedAt = Date.now(); + } + hosts.set(id, host); + } + return hosts; } static async saveHosts(filePath: string, hosts: Map) { @@ -467,16 +490,22 @@ export default class Controller { let instances = new Map(); try { - let serialized = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" })); - for (let serializedConfig of serialized) { + let serialized = JSON.parse( + await fs.readFile(filePath, { encoding: "utf8" }) + ) as Static[]; + for (let json of serialized) { + if (!json.config) { // migrate: from pre Alpha 14 format. + json = { config: json, status: "running" }; // Use running to force updatedAt + } let instanceConfig = new lib.InstanceConfig("controller"); - await instanceConfig.load(serializedConfig); - let status = instanceConfig.get("instance.assigned_host") === null ? "unassigned" : "unknown"; - let instance = new InstanceInfo({ - config: instanceConfig, - status: status as InstanceStatus, - }); - instances.set(instanceConfig.get("instance.id"), instance); + await instanceConfig.load(json.config as lib.SerializedConfig); + let instance = InstanceInfo.fromJSON(json, instanceConfig); + const status = instance.config.get("instance.assigned_host") === null ? "unassigned" : "unknown"; + if (instance.status !== status) { + instance.status = status; + instance.updatedAt = Date.now(); + } + instances.set(instance.id, instance); } } catch (err: any) { @@ -490,14 +519,28 @@ export default class Controller { } static async saveInstances(filePath: string, instances: Map) { - let serialized = []; - for (const instance of instances.values()) { - serialized.push(instance.config.serialize()); - } - + let serialized = [...instances.values()].map(instance => instance.toJSON()); await lib.safeOutputFile(filePath, JSON.stringify(serialized, null, "\t")); } + static async loadSaves(filePath: string): Promise> { + let json: Static[]; + try { + json = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" })); + } catch (err: any) { + if (err.code !== "ENOENT") { + throw err; + } + return new Map(); + } + return new Map(json.map(s => lib.SaveDetails.fromJSON(s)).map(s => [s.id, s])); + } + + static async saveSaves(filePath: string, saves: Map) { + await lib.safeOutputFile(filePath, JSON.stringify([...saves.values()], null, "\t")); + } + + static async loadModPacks(filePath: string): Promise> { let json; try { @@ -607,18 +650,21 @@ export default class Controller { ); } - hostUpdated(host: HostInfo) { - let update = new lib.HostDetails( - host.agent, - host.version, - host.name, - host.id, - this.wsServer.hostConnections.has(host.id), - host.public_address, - host.token_valid_after, - ); + hostsUpdated(hosts: HostInfo[]) { + const now = Date.now(); + for (const host of hosts) { + host.updatedAt = now; + } + this.hostsDirty = true; + let updates = hosts.map(host => host.toHostDetails()); + this.subscriptions.broadcast(new lib.HostUpdatesEvent(updates)); + } - this.subscriptions.broadcast(new lib.HostUpdateEvent(update)); + async handleHostSubscription(request: lib.SubscriptionRequest) { + const hosts = [...this.hosts.values()].filter( + host => host.updatedAt > request.lastRequestTime, + ).map(host => host.toHostDetails()); + return hosts.length ? new lib.HostUpdatesEvent(hosts) : null; } /** @@ -685,15 +731,34 @@ export default class Controller { }; instanceConfig.set("factorio.settings", settings); - let instance = new InstanceInfo({ config: instanceConfig, status: "unassigned" }); + let instance = new InstanceInfo(instanceConfig, "unassigned", undefined, Date.now()); this.instances.set(instanceId, instance); this.instancesDirty = true; await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance); this.addInstanceHooks(instance); - this.instanceDetailsUpdated(instance); + this.instanceDetailsUpdated([instance]); return instance; } + /** + * Removes all saves currently stored for the given instance, if any + * @param instanceId - Id of Instance to clear saves for. + * @internal + */ + clearSavesOfInstance(instanceId: number) { + const updates = []; + for (const [id, save] of this.saves) { + if (save.instanceId === instanceId) { + save.isDeleted = true; + updates.push(save); + this.saves.delete(id); + } + } + if (updates.length) { + this.savesUpdated(updates); + } + } + /** * Change assigned host of an instance * @@ -733,16 +798,19 @@ export default class Controller { } } + // Remove saves recorded from currently assigned host if any + this.clearSavesOfInstance(instanceId); + // Assign to target instance.config.set("instance.assigned_host", hostId); - this.instancesDirty = true; + // "fieldChanged" event handler will set this.instancesDirty if (hostId !== undefined && newHostConnection) { await newHostConnection.send( new lib.InstanceAssignInternalRequest(instanceId, instance.config.serialize("host")) ); } else { instance.status = "unassigned"; - this.instanceDetailsUpdated(instance); + this.instanceDetailsUpdated([instance]); } } @@ -763,9 +831,12 @@ export default class Controller { this.instances.delete(instanceId); this.instancesDirty = true; + this.clearSavesOfInstance(instanceId); + let prev = instance.status; instance.status = "deleted"; - this.instanceDetailsUpdated(instance); + instance.updatedAt = Date.now(); + this.instanceDetailsUpdated([instance]); await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance, prev); } @@ -792,8 +863,9 @@ export default class Controller { addInstanceHooks(instance: InstanceInfo) { instance.config.on("fieldChanged", (group: lib.ConfigGroup, field: string, prev: any) => { + instance.updatedAt = Date.now(); if (group.name === "instance" && field === "name") { - this.instanceDetailsUpdated(instance); + this.instanceDetailsUpdated([instance]); } this.instancesDirty = true; @@ -801,46 +873,73 @@ export default class Controller { }); } - instanceDetailsUpdated(instance: InstanceInfo) { - let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); - if (assigned_host === null) { - assigned_host = undefined; - } - - let game_port: number|null|undefined = instance.game_port; - if (game_port === null) { - game_port = undefined; - } - - this.subscriptions.broadcast(new lib.InstanceDetailsUpdateEvent( - new lib.InstanceDetails( - instance.config.get("instance.name"), - instance.id, - assigned_host, - game_port, - instance.status, - ) - )); + instanceDetailsUpdated(instances: InstanceInfo[]) { + const updates = instances.map(instance => instance.toInstanceDetails()); + this.subscriptions.broadcast(new lib.InstanceDetailsUpdatesEvent(updates)); } - saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) { - this.subscriptions.broadcast(new lib.InstanceSaveListUpdateEvent(instanceId, saves)); + async handleInstanceDetailsSubscription(request: lib.SubscriptionRequest) { + const instances = [...this.instances.values()].filter( + instance => instance.updatedAt > request.lastRequestTime, + ).map(instance => instance.toInstanceDetails()); + return instances.length ? new lib.InstanceDetailsUpdatesEvent(instances) : null; } - modPackUpdated(modPack: lib.ModPack) { + savesUpdated(saves: lib.SaveDetails[]) { + this.savesDirty = true; + this.subscriptions.broadcast(new lib.InstanceSaveDetailsUpdatesEvent(saves)); + } + + async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) { + const saves = [...this.saves.values()].filter(save => save.updatedAt > request.lastRequestTime); + return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null; + } + + + modPacksUpdated(modPacks: lib.ModPack[]) { + const now = Date.now(); + for (const modPack of modPacks) { + modPack.updatedAt = now; + } this.modPacksDirty = true; - this.subscriptions.broadcast(new lib.ModPackUpdateEvent(modPack)); - lib.invokeHook(this.plugins, "onModPackUpdated", modPack); + this.subscriptions.broadcast(new lib.ModPackUpdatesEvent(modPacks)); + lib.invokeHook(this.plugins, "onModPacksUpdated", modPacks); } - modUpdated(mod: lib.ModInfo) { - this.subscriptions.broadcast(new lib.ModUpdateEvent(mod)); - lib.invokeHook(this.plugins, "onModUpdated", mod); + async handleModPackSubscription(request: lib.SubscriptionRequest) { + const modPacks = [...this.modPacks.values()].filter( + modPack => modPack.updatedAt > request.lastRequestTime, + ); + return modPacks.length ? new lib.ModPackUpdatesEvent(modPacks) : null; } - userUpdated(user: ControllerUser) { + modsUpdated(mods: lib.ModInfo[]) { + // ModStore sets updatedAt 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, + ); + return mods.length ? new lib.ModUpdatesEvent(mods) : null; + } + + usersUpdated(users: ControllerUser[]) { + const now = Date.now(); + for (const user of users) { + user.updatedAt = now; + } this.userManager.dirty = true; - this.subscriptions.broadcast(new lib.UserUpdateEvent(user)); + this.subscriptions.broadcast(new lib.UserUpdatesEvent(users)); + } + + async handleUserSubscription(request: lib.SubscriptionRequest) { + const users = [...this.userManager.users.values()].filter( + user => user.updatedAt > request.lastRequestTime, + ); + return users.length ? new lib.UserUpdatesEvent(users) : null; } /** diff --git a/packages/controller/src/ControllerUser.ts b/packages/controller/src/ControllerUser.ts index 7c79dfb7..5af3b1c2 100644 --- a/packages/controller/src/ControllerUser.ts +++ b/packages/controller/src/ControllerUser.ts @@ -35,6 +35,7 @@ export default class ControllerUser extends User implements IControllerUser { json.is_banned, json.is_whitelisted, json.ban_reason, + json.updated_at, json.is_deleted, playerStats, instanceStats, diff --git a/packages/controller/src/HostConnection.ts b/packages/controller/src/HostConnection.ts index 140bd124..164f3e82 100644 --- a/packages/controller/src/HostConnection.ts +++ b/packages/controller/src/HostConnection.ts @@ -5,19 +5,10 @@ import * as lib from "@clusterio/lib"; const { logger } = lib; import BaseConnection from "./BaseConnection"; -import InstanceInfo, { InstanceStatus } from "./InstanceInfo"; +import InstanceInfo from "./InstanceInfo"; +import HostInfo from "./HostInfo"; -export interface HostInfo { - agent: string; - id: number; - name: string; - version: string; - plugins: Record; - public_address: string | undefined; - token_valid_after?: number; -} - /** * Represents the connection to a host * @@ -25,10 +16,10 @@ export interface HostInfo { * @alias module:controller/src/HostConnection */ export default class HostConnection extends BaseConnection { - private _agent: any; - private _id: any; - private _name: any; - private _version: any; + private _agent: string; + private _id: number; + private _name: string; + private _version: string; plugins: Map; constructor( @@ -45,18 +36,21 @@ export default class HostConnection extends BaseConnection { this.plugins = new Map(Object.entries(registerData.plugins)); this._checkPluginVersions(); - let currentHostInfo = this._controller.hosts.get(this._id); - - this._controller.hosts.set(this._id, { - agent: this._agent, - id: this._id, - name: this._name, - version: this._version, - public_address: registerData.publicAddress, - plugins: registerData.plugins, - token_valid_after: currentHostInfo?.token_valid_after, - }); - this._controller.hostsDirty = true; + const previousHostInfo = this._controller.hosts.get(this._id); + const currentHostInfo = new HostInfo( + this._agent, + this._id, + this._name, + this._version, + this.plugins, + true, + registerData.publicAddress ?? "", + previousHostInfo?.tokenValidAfter, + Date.now(), + false, + ); + this._controller.hosts.set(this._id, currentHostInfo); + this._controller.hostsUpdated([currentHostInfo]); for (let event of ["connect", "drop", "resume", "close"] as const) { // eslint-disable-next-line no-loop-func @@ -69,21 +63,26 @@ export default class HostConnection extends BaseConnection { this.connector.on("close", () => { // Update status to unknown for instances on this host. + const instances: InstanceInfo[] = []; for (let instance of this._controller.instances.values()) { if (instance.config.get("instance.assigned_host") !== this._id) { continue; } + instances.push(instance); let prev = instance.status; instance.status = "unknown"; - this._controller.instanceDetailsUpdated(instance); lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev); } + this._controller.instanceDetailsUpdated(instances); + + currentHostInfo.connected = false; + this._controller.hostsUpdated([currentHostInfo]); }); this.handle(lib.InstanceStatusChangedEvent, this.handleInstanceStatusChangedEvent.bind(this)); this.handle(lib.InstancesUpdateRequest, this.handleInstancesUpdateRequest.bind(this)); - this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this)); + this.handle(lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsUpdatesEvent.bind(this)); this.handle(lib.LogMessageEvent, this.handleLogMessageEvent.bind(this)); this.handle(lib.InstancePlayerUpdateEvent, this.handleInstancePlayerUpdateEvent.bind(this)); } @@ -177,10 +176,11 @@ export default class HostConnection extends BaseConnection { } let prev = instance.status; - instance.status = request.status as lib.InstanceStatus; - instance.game_port = request.gamePort || null; + instance.status = request.status; + instance.gamePort = request.gamePort; + instance.updatedAt = Date.now(); logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`); - this._controller.instanceDetailsUpdated(instance); + this._controller.instanceDetailsUpdated([instance]); await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev); } @@ -195,6 +195,7 @@ export default class HostConnection extends BaseConnection { } // Assign instances the host has but controller does not + const instanceUpdates: InstanceInfo[] = []; for (let instanceData of request.instances) { let instanceConfig = new lib.InstanceConfig("controller"); await instanceConfig.load(instanceData.config as lib.SerializedConfig, "host"); @@ -212,9 +213,11 @@ export default class HostConnection extends BaseConnection { // Already have this instance, update state instead if (controllerInstance.status !== instanceData.status) { let prev = controllerInstance.status; - controllerInstance.status = instanceData.status as InstanceStatus; + controllerInstance.status = instanceData.status; + controllerInstance.gamePort = instanceData.gamePort; + controllerInstance.updatedAt = Date.now(); logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`); - this._controller.instanceDetailsUpdated(controllerInstance); + instanceUpdates.push(controllerInstance); await lib.invokeHook( this._controller.plugins, "onInstanceStatusChanged", controllerInstance, prev ); @@ -224,7 +227,10 @@ export default class HostConnection extends BaseConnection { instanceConfig.set("instance.assigned_host", this._id); let newInstance = new InstanceInfo( - { config: instanceConfig, status: instanceData.status as InstanceStatus } + instanceConfig, + instanceData.status, + instanceData.gamePort, + Date.now(), ); this._controller.instances.set(instanceConfig.get("instance.id"), newInstance); this._controller.instancesDirty = true; @@ -234,10 +240,14 @@ export default class HostConnection extends BaseConnection { instanceConfig.get("instance.id"), instanceConfig.serialize("host") ) ); - this._controller.instanceDetailsUpdated(newInstance); + instanceUpdates.push(newInstance); await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", newInstance); } + if (instanceUpdates.length) { + this._controller.instanceDetailsUpdated(instanceUpdates); + } + // Push lists to make sure they are in sync. let adminlist: Set = new Set(); let banlist: Map = new Map(); @@ -258,8 +268,34 @@ export default class HostConnection extends BaseConnection { await this.send(new lib.SyncUserListsEvent(adminlist, banlist, whitelist)); } - async handleInstanceSaveListUpdateEvent(event: lib.InstanceSaveListUpdateEvent) { - this._controller.saveListUpdate(event.instanceId, event.saves); + async handleInstanceSaveDetailsUpdatesEvent(event: lib.InstanceSaveDetailsUpdatesEvent) { + const now = Date.now(); + const updates: lib.SaveDetails[] = []; + for (const save of event.updates) { + const existingSave = this._controller.saves.get(save.id); + if (existingSave && save.equals(existingSave)) { + continue; + } + save.updatedAt = now; + this._controller.saves.set(save.id, save); + updates.push(save); + } + if (event.instanceId !== undefined) { + const updatedSaves = new Set(event.updates.map(s => s.id)); + for (const [id, save] of this._controller.saves) { + if (save.instanceId === event.instanceId && !updatedSaves.has(id)) { + save.isDeleted = true; + save.updatedAt = now; + updates.push(save); + this._controller.saves.delete(id); + } + } + } + // Hosts eagerly send updates, which means we may get an update + // where nothing actualy changed. + if (updates.length) { + this._controller.savesUpdated(updates); + } } async handleLogMessageEvent(event: lib.LogMessageEvent) { @@ -286,7 +322,7 @@ export default class HostConnection extends BaseConnection { user.instanceStats.set(instanceId, event.stats); user.recalculatePlayerStats(); - this._controller.userUpdated(user); + this._controller.usersUpdated([user]); let instance = this._controller.instances.get(instanceId)!; await lib.invokeHook(this._controller.plugins, "onPlayerEvent", instance, { diff --git a/packages/controller/src/HostInfo.ts b/packages/controller/src/HostInfo.ts new file mode 100644 index 00000000..7a817c53 --- /dev/null +++ b/packages/controller/src/HostInfo.ts @@ -0,0 +1,88 @@ +import { Type, Static } from "@sinclair/typebox"; +import { HostDetails } from "@clusterio/lib"; + +/** + * Info about a host known to the controller. + */ +export default class HostInfo { + constructor( + /** Agent this host last connected with */ + public agent: string, + /** Id of this host */ + public id: number, + /** Version this host last connected with */ + public name: string, + /** Version this host last connected with */ + public version: string, + /** Plugins this host last connected with */ + public plugins: Map, + /** True if this host is currently connected to controller */ + public connected: boolean = false, + /** Value of host.public_address configured for this host */ + public publicAddress: string = "", + /** Unix timestamp in seconds host token must be issued afterto be valid */ + public tokenValidAfter: number = 0, + /** Millisecond Unix timestamp this entry was last updated at */ + public updatedAt: number = 0, + /** True if this host has been deleted */ + public isDeleted: boolean = false, + ) { } + + static jsonSchema = Type.Object({ + "agent": Type.String(), + "id": Type.Number(), + "name": Type.String(), + "version": Type.String(), + "plugins": Type.Record(Type.String(), Type.String()), + "connected": Type.Boolean(), + "public_address": Type.String(), + "token_valid_after": Type.Number(), + "updated_at": Type.Number(), + "is_deleted": Type.Boolean(), + }); + + static fromJSON(json: Static) { + return new this( + json.agent, + json.id, + json.name, + json.version, + new Map(Object.entries(json.plugins)), + json.connected, + json.public_address, + json.token_valid_after, + json.updated_at, + json.is_deleted, + ); + } + + toJSON(): Static { + return { + agent: this.agent, + id: this.id, + name: this.name, + version: this.version, + plugins: Object.fromEntries(this.plugins), + connected: this.connected, + public_address: this.publicAddress, + token_valid_after: this.tokenValidAfter, + updated_at: this.updatedAt, + is_deleted: this.isDeleted, + }; + } + + toHostDetails() { + return new HostDetails( + this.agent, + this.version, + this.name, + this.id, + this.connected, + this.publicAddress, + this.tokenValidAfter, + this.updatedAt, + this.isDeleted, + ); + } +} + diff --git a/packages/controller/src/InstanceInfo.ts b/packages/controller/src/InstanceInfo.ts index e33f43f3..320fa32c 100644 --- a/packages/controller/src/InstanceInfo.ts +++ b/packages/controller/src/InstanceInfo.ts @@ -1,22 +1,5 @@ -import type * as lib from "@clusterio/lib"; - -/** - * Current status of the instance. One of: - * - `unknown`: Instance is assigned to a host but this host is currently - * not connected to the contreller. - * - `unassigned`: Instance is not assigned to a a host and exists only on - * the controller. - * - `stopped`: Instance is stopped. - * - `starting`: Instance is in the process of starting up. - * - `running`: Instance is running normally. - * - `stopping`: Instance is in the process of stopping. - * - `creating_save`: Instance is in the process of creating a save. - * - `exporting_data`: Instance is in the process of exporting game data. - * - `deleted`: Instance has been deleted. - */ -export type InstanceStatus = - "unknown" | "unassigned" | "stopped" | "starting" | "running" | - "stopping" | "creating_save" | "exporting_data" | "deleted"; +import * as lib from "@clusterio/lib"; +import { Static, Type } from "@sinclair/typebox"; /** @@ -24,25 +7,56 @@ export type InstanceStatus = * @alias module:controller/src/InstanceInfo */ export default class InstanceInfo { - config: lib.InstanceConfig; - status: InstanceStatus; - game_port: number | null = null; - constructor( - { config, status }: - { config: lib.InstanceConfig, status: InstanceStatus } + public config: lib.InstanceConfig, + public status: lib.InstanceStatus, + public gamePort?: number, + public updatedAt = 0, ) { this.config = config; this.status = status; } - toJSON() { + static jsonSchema = Type.Object({ + "config": Type.Object({}), + "status": lib.InstanceStatus, + "gamePort": Type.Optional(Type.Number()), + "updatedAt": Type.Optional(Type.Number()), + }); + + static fromJSON(json: Static, config: lib.InstanceConfig) { + return new this( + config, + json.status, + json.gamePort, + json.updatedAt, + ); + } + + toJSON(): Static { return { config: this.config.serialize(), status: this.status, + gamePort: this.gamePort, + updatedAt: this.updatedAt, }; } + toInstanceDetails() { + return new lib.InstanceDetails( + this.config.get("instance.name"), + this.id, + this.config.get("instance.assigned_host") ?? undefined, + this.gamePort, + this.status, + this.updatedAt, + ); + } + + get isDeleted() { + return this.status === "deleted"; + } + /** Shorthand for `instance.config.get("instance.id")` */ get id():number { return this.config.get("instance.id"); diff --git a/packages/controller/src/WsServer.ts b/packages/controller/src/WsServer.ts index bc63a56f..fb10cc66 100644 --- a/packages/controller/src/WsServer.ts +++ b/packages/controller/src/WsServer.ts @@ -9,7 +9,7 @@ import * as lib from "@clusterio/lib"; const { logger } = lib; import ControlConnection from "./ControlConnection"; -import HostConnection, { HostInfo } from "./HostConnection"; +import HostConnection from "./HostConnection"; import WsServerConnector from "./WsServerConnector"; import { version as packageVersion } from "../package.json"; @@ -243,7 +243,7 @@ ${err.stack}` } let host = this.controller.hosts.get(data.id); - if (!tokenPayload.iat || tokenPayload.iat < (host?.token_valid_after??0)) { + if (!tokenPayload.iat || tokenPayload.iat < (host?.tokenValidAfter ?? 0)) { throw new Error("invalid token"); } } catch (err: any) { @@ -274,13 +274,11 @@ ${err.stack}` connector.on("close", () => { if (this.hostConnections.get(data.id) === connection) { this.hostConnections.delete(data.id); - this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo); } else { logger.warn("Unlisted HostConnection closed"); } }); this.hostConnections.set(data.id, connection); - this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo); let src = new lib.Address(lib.Address.host, data.id); connector.ready(socket, src, sessionToken, undefined); } diff --git a/packages/controller/src/routes.ts b/packages/controller/src/routes.ts index 99446069..5fceb72a 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -152,7 +152,7 @@ function validateHostToken(req: Request, res: Response, next: any) { if (!host) { throw new Error("invalid host"); } - if (!tokenPayload.iat || tokenPayload.iat < (host.token_valid_after??0)) { + if (!tokenPayload.iat || tokenPayload.iat < (host.tokenValidAfter ?? 0)) { throw new Error("invalid token"); } } catch (err) { @@ -256,7 +256,7 @@ async function uploadExport(req: Request, res: Response) { modPack.exportManifest = new lib.ExportManifest(assets); modPack.fillDefaultSettings(settingPrototypes, logger); - res.app.locals.controller.modPackUpdated(modPack); + res.app.locals.controller.modPacksUpdated([modPack]); res.sendStatus(200); } diff --git a/packages/ctl/src/commands.ts b/packages/ctl/src/commands.ts index a1cfdcfe..c112c162 100644 --- a/packages/ctl/src/commands.ts +++ b/packages/ctl/src/commands.ts @@ -532,7 +532,8 @@ instanceSaveCommands.add(new lib.Command({ }], handler: async function(args: { instance: string }, control: Control) { let instanceId = await lib.resolveInstance(control, args.instance); - let saves = await control.sendTo({ instanceId }, new lib.InstanceListSavesRequest()); + let saves = await control.sendTo("controller", new lib.InstanceSaveDetailsListRequest()); + saves = saves.filter(save => save.instanceId === instanceId); print(asTable(saves.map( ({ mtimeMs, ...rest }) => ({ ...rest, mtime: new Date(mtimeMs).toLocaleString() }) ))); @@ -866,7 +867,7 @@ modPackCommands.add(new lib.Command({ }], handler: async function(args: { modPack: string }, control: Control) { let mods = new Map( - (await control.send(new lib.ModListRequest())).map(m => [`${m.name}_${m.version}`, m]) + (await control.send(new lib.ModListRequest())).map(m => [m.id, m]) ); let modPack = await control.send( new lib.ModPackGetRequest(await lib.resolveModPack(control, args.modPack)) diff --git a/packages/host/src/Host.ts b/packages/host/src/Host.ts index c9395724..a2c700e3 100644 --- a/packages/host/src/Host.ts +++ b/packages/host/src/Host.ts @@ -347,7 +347,9 @@ export default class Host extends lib.Link { this.handle(lib.InstanceAssignInternalRequest, this.handleInstanceAssignInternalRequest.bind(this)); this.handle(lib.InstanceUnassignInternalRequest, this.handleInstanceUnassignInternalRequest.bind(this)); this.handle(lib.HostMetricsRequest, this.handleHostMetricsRequest.bind(this)); - this.fallbackRequest(lib.InstanceListSavesRequest, this.fallbackInstanceListSavesRequest.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)); @@ -600,6 +602,9 @@ export default class Host extends lib.Link { ) ); + // Send the new list of saves for this assigned instance to the controller. + await this.sendSaveListUpdate(instanceId, path.join(instanceInfo.path, "saves")); + // save a copy of the instance config let warnedOutput = { _warning: "Changes to this file will be overwritten by the controller's copy.", @@ -700,9 +705,13 @@ export default class Host extends lib.Link { return { results }; } - async fallbackInstanceListSavesRequest(request: lib.InstanceListSavesRequest, src: lib.Address, dst: lib.Address) { + async fallbackInstanceSaveDetailsListRequest( + request: lib.InstanceSaveDetailsListRequest, + src: lib.Address, + dst: lib.Address + ) { let instanceInfo = this.getRequestInstanceInfo(dst.id); - return await Instance.listSaves(path.join(instanceInfo.path, "saves"), null); + return await Instance.listSaves(dst.id, path.join(instanceInfo.path, "saves"), null); } async handleInstanceRenameSaveRequest(request: lib.InstanceRenameSaveRequest) { @@ -788,12 +797,12 @@ export default class Host extends lib.Link { let instanceConnection = this.instanceConnections.get(instanceId); let saveList: lib.SaveDetails[]; if (instanceConnection) { - saveList = await instanceConnection.send(new lib.InstanceListSavesRequest()); + saveList = await instanceConnection.send(new lib.InstanceSaveDetailsListRequest()); } else { - saveList = await Instance.listSaves(savesDir, null); + saveList = await Instance.listSaves(instanceId, savesDir, null); } - this.send(new lib.InstanceSaveListUpdateEvent(instanceId, saveList)); + this.send(new lib.InstanceSaveDetailsUpdatesEvent(saveList, instanceId)); } async handleInstanceDeleteSaveRequest(request: lib.InstanceDeleteSaveRequest) { @@ -913,6 +922,7 @@ export default class Host extends lib.Link { list.push(new lib.RawInstanceInfo( instanceInfo.config.serialize("controller"), instanceConnection ? instanceConnection.instance.status : "stopped", + instanceConnection ? instanceConnection.instance.server.gamePort : undefined, )); } await this.send(new lib.InstancesUpdateRequest(list)); diff --git a/packages/host/src/Instance.ts b/packages/host/src/Instance.ts index aab54cd1..22c43cf7 100644 --- a/packages/host/src/Instance.ts +++ b/packages/host/src/Instance.ts @@ -108,7 +108,7 @@ export default class Instance extends lib.Link { /** * Mod pack currently running on this instance */ - activeModPack = new lib.ModPack(); + activeModPack!: lib.ModPack; // This is set in syncMods /** * Per player statistics recorded by this instance. */ @@ -228,7 +228,7 @@ export default class Instance extends lib.Link { this.handle(lib.InstanceMetricsRequest, this.handleInstanceMetricsRequest.bind(this)); this.handle(lib.InstanceStartRequest, this.handleInstanceStartRequest.bind(this)); this.handle(lib.InstanceLoadScenarioRequest, this.handleInstanceLoadScenarioRequest.bind(this)); - this.handle(lib.InstanceListSavesRequest, this.handleInstanceListSavesRequest.bind(this)); + this.handle(lib.InstanceSaveDetailsListRequest, this.handleInstanceSaveDetailsListRequest.bind(this)); this.handle(lib.InstanceCreateSaveRequest, this.handleInstanceCreateSaveRequest.bind(this)); this.handle(lib.InstanceExportDataRequest, this.handleInstanceExportDataRequest.bind(this)); this.handle(lib.InstanceStopRequest, this.handleInstanceStopRequest.bind(this)); @@ -374,7 +374,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " "); } } - static async listSaves(savesDir: string, loadedSave: string | null) { + static async listSaves(instanceId: number, savesDir: string, loadedSave: string | null) { let defaultSave = null; if (loadedSave === null) { defaultSave = await lib.getNewestFile( @@ -395,12 +395,15 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " "); } list.push(new lib.SaveDetails( + instanceId, type, name, stat.size, stat.mtimeMs, name === loadedSave, name === defaultSave, + 0, // Set by controller + false, )); } @@ -410,9 +413,9 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " "); async sendSaveListUpdate() { this.sendTo( "controller", - new lib.InstanceSaveListUpdateEvent( + new lib.InstanceSaveDetailsUpdatesEvent( + await Instance.listSaves(this.id, this.path("saves"), this._loadedSave), this.id, - await Instance.listSaves(this.path("saves"), this._loadedSave), ), ); } @@ -446,7 +449,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " "); new lib.InstanceStatusChangedEvent( this.id, status, - this.server.gamePort || this.config.get("factorio.game_port") || undefined, + status !== "stopped" ? this.server.gamePort : undefined, ), ); } @@ -1034,8 +1037,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " "); } } - async handleInstanceListSavesRequest() { - return await Instance.listSaves(this.path("saves"), this._loadedSave); + async handleInstanceSaveDetailsListRequest() { + return await Instance.listSaves(this.id, this.path("saves"), this._loadedSave); } async handleInstanceCreateSaveRequest(request: lib.InstanceCreateSaveRequest) { diff --git a/packages/lib/src/ModStore.ts b/packages/lib/src/ModStore.ts index 54eda548..83cb0167 100644 --- a/packages/lib/src/ModStore.ts +++ b/packages/lib/src/ModStore.ts @@ -66,6 +66,7 @@ export default class ModStore extends TypedEventEmitter) { @@ -32,6 +37,8 @@ export class HostDetails { json.connected, json.publicAddress, json.tokenValidAfter, + json.updatedAt, + json.isDeleted, ); } } @@ -45,27 +52,23 @@ export class HostListRequest { static Response = jsonArray(HostDetails); } -export class HostUpdateEvent { - declare ["constructor"]: typeof HostUpdateEvent; +export class HostUpdatesEvent { + declare ["constructor"]: typeof HostUpdatesEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; static permission = "core.host.subscribe" as const; constructor( - public update: HostDetails, + public updates: HostDetails[], ) { } static jsonSchema = Type.Object({ - "update": HostDetails.jsonSchema, + "updates": Type.Array(HostDetails.jsonSchema), }); static fromJSON(json: Static) { - return new this(HostDetails.fromJSON(json.update)); - } - - get subscriptionChannel() { - return this.update.id; + return new this(json.updates.map(update => HostDetails.fromJSON(update))); } } diff --git a/packages/lib/src/data/messages_instance.ts b/packages/lib/src/data/messages_instance.ts index c5ce1609..dbc32be1 100644 --- a/packages/lib/src/data/messages_instance.ts +++ b/packages/lib/src/data/messages_instance.ts @@ -7,10 +7,34 @@ import type { MessageRequest } from "./messages_core"; import { CollectorResultSerialized } from "../prometheus"; -export type InstanceStatus = - "unknown" | "unassigned" | "stopped" | "starting" | "running" - | "stopping" | "creating_save" | "exporting_data" | "deleted" -; +export const InstanceStatus = StringEnum([ + "unknown", + "unassigned", + "stopped", + "starting", + "running", + "stopping", + "creating_save", + "exporting_data", + "deleted", +]); + +/** + * Current status of the instance. One of: + * - `unknown`: Instance is assigned to a host but this host is currently + * not connected to the contreller. + * - `unassigned`: Instance is not assigned to a a host and exists only on + * the controller. + * - `stopped`: Instance is stopped. + * - `starting`: Instance is in the process of starting up. + * - `running`: Instance is running normally. + * - `stopping`: Instance is in the process of stopping. + * - `creating_save`: Instance is in the process of creating a save. + * - `exporting_data`: Instance is in the process of exporting game data. + * - `deleted`: Instance has been deleted. + */ +export type InstanceStatus = Static; + export class InstanceDetails { constructor( public name: string, @@ -18,6 +42,8 @@ export class InstanceDetails { public assignedHost: number | undefined, public gamePort: number | undefined, public status: InstanceStatus, + /** Millisecond Unix timestamp this entry was last updated at */ + public updatedAt = 0, ) { } static jsonSchema = Type.Object({ @@ -29,10 +55,22 @@ export class InstanceDetails { "unknown", "unassigned", "stopped", "starting", "running", "stopping", "creating_save", "exporting_data", "deleted", ]), + "updatedAt": Type.Optional(Type.Number()), }); static fromJSON(json: Static) { - return new this(json.name, json.id, json.assignedHost, json.gamePort, json.status); + return new this( + json.name, + json.id, + json.assignedHost, + json.gamePort, + json.status, + json.updatedAt, + ); + } + + get isDeleted() { + return this.status === "deleted"; } } @@ -67,29 +105,23 @@ export class InstanceDetailsListRequest { static Response = jsonArray(InstanceDetails); }; -export class InstanceDetailsUpdateEvent { - declare ["constructor"]: typeof InstanceDetailsUpdateEvent; +export class InstanceDetailsUpdatesEvent { + declare ["constructor"]: typeof InstanceDetailsUpdatesEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; static permission = "core.instance.subscribe" as const; constructor( - public details: InstanceDetails, + public updates: InstanceDetails[], ) { } - static jsonSchema = InstanceDetails.jsonSchema; - - toJSON() { - return this.details; - } + static jsonSchema = Type.Object({ + "updates": Type.Array(InstanceDetails.jsonSchema), + }); static fromJSON(json: Static) { - return new this(InstanceDetails.fromJSON(json)); - } - - get subscriptionChannel() { - return this.details.id; + return new this(json.updates.map(update => InstanceDetails.fromJSON(update))); } }; @@ -251,60 +283,104 @@ export class InstanceStartRequest { export class SaveDetails { constructor( - public type: string, + public instanceId: number, + public type: "file" | "directory" | "special", public name: string, public size: number, public mtimeMs: number, public loaded: boolean, public loadByDefault: boolean, + /** Millisecond Unix timestamp this entry was last updated at */ + public updatedAt: number, + public isDeleted: boolean, ) { } static jsonSchema = Type.Object({ + "instanceId": Type.Integer(), "type": StringEnum(["file", "directory", "special"]), "name": Type.String(), "size": Type.Integer(), "mtimeMs": Type.Number(), "loaded": Type.Boolean(), "loadByDefault": Type.Boolean(), + "updatedAt": Type.Number(), + "isDeleted": Type.Boolean(), }); static fromJSON(json: Static) { - return new this(json.type, json.name, json.size, json.mtimeMs, json.loaded, json.loadByDefault); + return new this( + json.instanceId, + json.type, + json.name, + json.size, + json.mtimeMs, + json.loaded, + json.loadByDefault, + json.updatedAt, + json.isDeleted, + ); + } + + /** Concatenation of instance id and name forming a unique string for this save */ + get id() { + return `${this.instanceId}/${this.name}`; + } + + /** + * Compare this save with another save. + * @param other - save to compare with. + * @returns true if this save is identical to the provided save + */ + equals(other: SaveDetails) { + // Note that updatedAt is not included in the equality check because + // instances send the whole list of saves with updatedAt set to zero + // to the controller whenever any save changes and this equals check + // is used to filter whether a save was updated. + return ( + this.instanceId === other.instanceId + && this.type === other.type + && this.name === other.name + && this.size === other.size + && this.mtimeMs === other.mtimeMs + && this.loaded === other.loaded + && this.loadByDefault === other.loadByDefault + && this.isDeleted === other.isDeleted + ); } } -export class InstanceListSavesRequest { - declare ["constructor"]: typeof InstanceListSavesRequest; +export class InstanceSaveDetailsListRequest { + declare ["constructor"]: typeof InstanceSaveDetailsListRequest; static type = "request" as const; - static src = "control" as const; - static dst = "instance" as const; + static src = ["control", "host"] as const; + static dst = ["controller", "instance"] as const; static permission = "core.instance.save.list" as const; static Response = jsonArray(SaveDetails); } -export class InstanceSaveListUpdateEvent { - declare ["constructor"]: typeof InstanceSaveListUpdateEvent; +export class InstanceSaveDetailsUpdatesEvent { + declare ["constructor"]: typeof InstanceSaveDetailsUpdatesEvent; static type = "event" as const; static src = ["instance", "host", "controller"] as const; static dst = ["controller", "control"] as const; - static permission = "core.instance.save.list_subscribe" as const; + static permission = "core.instance.save.list.subscribe" as const; constructor( - public instanceId: number, - public saves: SaveDetails[], + public updates: SaveDetails[], + /** + * Present if this update was sent by a host and updates contains all + * saves of the given instance. + */ + public instanceId?: number, ) { } static jsonSchema = Type.Object({ - "instanceId": Type.Integer(), - "saves": Type.Array(SaveDetails.jsonSchema), + "updates": Type.Array(SaveDetails.jsonSchema), + "instanceId": Type.Optional(Type.Integer()), }); static fromJSON(json: Static) { - return new this(json.instanceId, json.saves.map(i => SaveDetails.fromJSON(i))); - } - - get subscriptionChannel() { - return this.instanceId; + return new this(json.updates.map(i => SaveDetails.fromJSON(i)), json.instanceId); } } @@ -643,7 +719,8 @@ export class RawInstanceInfo { constructor( public config: object, - public status: string, + public status: InstanceStatus, + public gamePort: undefined | number, ) { } static jsonSchema = Type.Object({ @@ -651,10 +728,11 @@ export class RawInstanceInfo { "status": StringEnum([ "stopped", "starting", "running", "stopping", "creating_save", "exporting_data", ]), + "gamePort": Type.Optional(Type.Number()), }); static fromJSON(json: Static) { - return new this(json.config, json.status); + return new this(json.config, json.status, json.gamePort); } } @@ -747,7 +825,7 @@ export class InstanceStatusChangedEvent { constructor( public instanceId: number, - public status: string, + public status: InstanceStatus, public gamePort?: number, ) { } diff --git a/packages/lib/src/data/messages_mod.ts b/packages/lib/src/data/messages_mod.ts index b64c86cf..60fc6563 100644 --- a/packages/lib/src/data/messages_mod.ts +++ b/packages/lib/src/data/messages_mod.ts @@ -245,50 +245,42 @@ export class ModDeleteRequest { } } -export class ModPackUpdateEvent { - declare ["constructor"]: typeof ModPackUpdateEvent; +export class ModPackUpdatesEvent { + declare ["constructor"]: typeof ModPackUpdatesEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; static permission = "core.mod_pack.subscribe" as const; constructor( - public modPack: ModPack, + public updates: ModPack[], ) { } static jsonSchema = Type.Object({ - "modPack": ModPack.jsonSchema, + "updates": Type.Array(ModPack.jsonSchema), }); static fromJSON(json: Static) { - return new this(ModPack.fromJSON(json.modPack)); - } - - get subscriptionChannel() { - return this.modPack.id; + return new this(json.updates.map(update => ModPack.fromJSON(update))); } } -export class ModUpdateEvent { - declare ["constructor"]: typeof ModUpdateEvent; +export class ModUpdatesEvent { + declare ["constructor"]: typeof ModUpdatesEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; static permission = "core.mod.subscribe" as const; constructor( - public mod: ModInfo, + public updates: ModInfo[], ) { } static jsonSchema = Type.Object({ - "mod": ModInfo.jsonSchema, + "updates": Type.Array(ModInfo.jsonSchema), }); static fromJSON(json: Static) { - return new this(ModInfo.fromJSON(json.mod)); - } - - get subscriptionChannel() { - return this.mod.name; + return new this(json.updates.map(update => ModInfo.fromJSON(update))); } } diff --git a/packages/lib/src/data/messages_user.ts b/packages/lib/src/data/messages_user.ts index 4e8b658b..17fcae89 100644 --- a/packages/lib/src/data/messages_user.ts +++ b/packages/lib/src/data/messages_user.ts @@ -301,26 +301,22 @@ export class UserDeleteRequest { } } -export class UserUpdateEvent { - declare ["constructor"]: typeof UserUpdateEvent; +export class UserUpdatesEvent { + declare ["constructor"]: typeof UserUpdatesEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; static permission = "core.user.subscribe" as const; constructor( - public user: User, + public updates: User[], ) { } static jsonSchema = Type.Object({ - "user": User.jsonSchema, + "updates": Type.Array(User.jsonSchema), }); static fromJSON(json: Static) { - return new this(User.fromJSON(json.user)); - } - - get subscriptionChannel() { - return this.user.name; + return new this(json.updates.map(update => User.fromJSON(update))); } } diff --git a/packages/lib/src/link/messages.ts b/packages/lib/src/link/messages.ts index e9b1cc5e..13743844 100644 --- a/packages/lib/src/link/messages.ts +++ b/packages/lib/src/link/messages.ts @@ -31,7 +31,7 @@ export const dataClasses: (RequestClass | EventClass) subscriptions.SubscriptionRequest, host.HostListRequest, - host.HostUpdateEvent, + host.HostUpdatesEvent, host.HostMetricsRequest, host.ControllerConnectionEvent, host.PrepareControllerDisconnectRequest, @@ -40,7 +40,7 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceDetailsGetRequest, instance.InstanceDetailsListRequest, - instance.InstanceDetailsUpdateEvent, + instance.InstanceDetailsUpdatesEvent, instance.InstanceCreateRequest, instance.InstanceConfigGetRequest, instance.InstanceConfigSetFieldRequest, @@ -48,8 +48,8 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceAssignRequest, instance.InstanceMetricsRequest, instance.InstanceStartRequest, - instance.InstanceListSavesRequest, - instance.InstanceSaveListUpdateEvent, + instance.InstanceSaveDetailsListRequest, + instance.InstanceSaveDetailsUpdatesEvent, instance.InstanceCreateSaveRequest, instance.InstanceRenameSaveRequest, instance.InstanceCopySaveRequest, @@ -88,8 +88,8 @@ export const dataClasses: (RequestClass | EventClass) mod.ModSearchRequest, mod.ModDownloadRequest, mod.ModDeleteRequest, - mod.ModPackUpdateEvent, - mod.ModUpdateEvent, + mod.ModPackUpdatesEvent, + mod.ModUpdatesEvent, user.PermissionListRequest, user.RoleListRequest, @@ -106,5 +106,5 @@ export const dataClasses: (RequestClass | EventClass) user.UserSetWhitelistedRequest, user.UserSetBannedRequest, user.UserDeleteRequest, - user.UserUpdateEvent, + user.UserUpdatesEvent, ]; diff --git a/packages/lib/src/permissions.ts b/packages/lib/src/permissions.ts index 28a7cf89..55a2ca5b 100644 --- a/packages/lib/src/permissions.ts +++ b/packages/lib/src/permissions.ts @@ -201,7 +201,7 @@ definePermission({ description: "List the saves currently on the instance.", }); definePermission({ - name: "core.instance.save.list_subscribe", + name: "core.instance.save.list.subscribe", title: "Subscribe to save list updatess", description: "Subscribe to be notifed on updates to the list of saves of instances.", }); diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 7a176f55..abf237cd 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,73 +1,21 @@ import { Type, Static } from "@sinclair/typebox"; import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link"; +import { logger } from "./logging"; import { Address, MessageRequest, IControllerUser } from "./data"; export type SubscriptionRequestHandler = RequestHandler | null>; -export type EventSubscriberCallback = (value: T) => void; - -export type ChannelEvent = Event & { - get subscriptionChannel(): number | string; -} - -export type ChannelEventClass = EventClass & { - new(...args: any): ChannelEvent -} - -/** - * Response received by the subscriber after a request - * It can contain an eventReplay value if the event sender implements a subscription handler - * This replay will be send to the handlers as if an update had just occurred - */ -export class SubscriptionResponse { - constructor( - public readonly eventReplay: Event | null = null, - ) { - if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) { - throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`); - } - } - - static jsonSchema = Type.Union([ - Type.Tuple([ - Type.String(), - Type.Unknown(), - ]), - Type.Null(), - ]); - - toJSON() { - if (this.eventReplay) { - const entry = Link._eventsByClass.get(this.eventReplay.constructor)!; - return [entry.name, this.eventReplay]; - } - return null; - } - - static fromJSON(json: Static): SubscriptionResponse { - if (json) { - const entry = Link._eventsByName.get(json[0]); - if (!entry) { - throw new Error(`Unregistered Event class ${json[0]}`); - } else { - return new SubscriptionResponse(entry.eventFromJSON(json[1])); - } - } else { - return new SubscriptionResponse(); - } - } -} +export type EventSubscriberCallback = (updates: T[], synced: boolean) => void /** * A subscription request sent by a subscriber, this updates what events the subscriber will be sent * The permission for this request copies the permission from the event being subscribed to - * allChannels: false, channels: [] will unsubscribe the subscriber from all notifications + * subscribe: false will unsubscribe the subscriber from all notifications */ export class SubscriptionRequest { declare ["constructor"]: typeof SubscriptionRequest; static type = "request" as const; static src = ["control", "instance"] as const; static dst = "controller" as const; - static Response = SubscriptionResponse; static permission(user: IControllerUser, message: MessageRequest) { if (typeof message.data === "object" && message.data !== null) { const data = message.data as Static; @@ -84,8 +32,7 @@ export class SubscriptionRequest { constructor( public eventName: string, - public allChannels: boolean, - public channels: Array = [], + public subscribe: boolean, public lastRequestTime: number = 0, ) { if (!Link._eventsByName.has(eventName)) { @@ -96,12 +43,11 @@ export class SubscriptionRequest { static jsonSchema = Type.Tuple([ Type.String(), Type.Boolean(), - Type.Array(Type.Union([Type.String(), Type.Number()])), Type.Number(), ]); toJSON() { - return [this.eventName, this.allChannels, this.channels, this.lastRequestTime]; + return [this.eventName, this.subscribe, this.lastRequestTime]; } static fromJSON(json: Static): SubscriptionRequest { @@ -111,7 +57,7 @@ export class SubscriptionRequest { type EventData = { subscriptionUpdate?: SubscriptionRequestHandler, - subscriptions: Map, onceClose: () => void }>, + subscriptions: Set, }; /** @@ -121,18 +67,12 @@ type EventData = { export class SubscriptionController { _events = new Map(); - constructor( - private controller: any, // Controller | Link - ) { - this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this)); - } - /** * Allow clients to subscribe to an event by telling the subscription controller to accept them - * Has an optional subscription update handler which is called when any client updates their subscription + * Has an optional subscription update handler which is called when a client subscribes * @param Event - Event class which is sent out as updates. * @param subscriptionUpdate - - * Optional handler called when a client updates its subscriptions. + * Optional handler called when a client subscribes. */ handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler) { const entry = Link._eventsByClass.get(Event); @@ -144,15 +84,15 @@ export class SubscriptionController { } this._events.set(entry.name, { subscriptionUpdate: subscriptionUpdate, - subscriptions: new Map(), + subscriptions: new Set(), }); } /** - * Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided + * Broadcast an event to all subscribers of that event * @param event - Event to broadcast. */ - broadcast(event: Event | ChannelEvent) { + broadcast(event: Event) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { throw new Error(`Unregistered Event class ${event.constructor.name}`); @@ -161,25 +101,34 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } - const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null; - for (let [link, subscription] of eventData.subscriptions) { + for (let link of eventData.subscriptions) { if ((link.connector as WebSocketBaseConnector).closing) { eventData.subscriptions.delete(link); - } else if ((subscription.all || (channel && subscription.channels.includes(channel)))) { + } else { link.send(event); } } } /** - * Handle incoming event subscription requests + * Unsubscribe from all events of a given link. + * Used when a link is closed to stop all active subscriptions. + * @param link - Link to stop sending events to. + */ + unsubscribe(link: Link) { + for (let eventData of this._events.values()) { + eventData.subscriptions.delete(link); + } + } + + /** + * Handle incoming subscription requests on a link + * @param link - Link message was received on * @param event - incomming event. * @param src - Source address of incomming request. * @param dst - destination address of incomming request. - * @returns Response to subscription request. - * @internal */ - async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { + async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) { if (!Link._eventsByName.has(event.eventName)) { throw new Error(`Event ${event.eventName} is not a registered event`); } @@ -187,184 +136,158 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${event.eventName} is not a registered as subscribable`); } - const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null; - const link: Link = this.controller.wsServer.controlConnections.get(src.id); - if (event.allChannels === false && event.channels.length === 0) { - let onceClose = eventData.subscriptions.get(link)?.onceClose; - if (onceClose) { - link.connector.off("close", onceClose); - eventData.subscriptions.delete(link); - } + if (event.subscribe === false) { + eventData.subscriptions.delete(link); } else { - let onceClose = eventData.subscriptions.get(link)?.onceClose; - if (!onceClose) { - onceClose = () => eventData.subscriptions.delete(link); - link.connector.once("close", onceClose); - } - eventData.subscriptions.set( - link, { all: event.allChannels, channels: event.channels, onceClose: onceClose }, - ); - } - return new SubscriptionResponse(eventReplay); - } -} - -/** - * A class component to allow subscribing and unsubscribing to/from an event and its channels - * Channels are an optional feature which can be implemented to only get notifications of reliant updates - * Multiple handlers can be subscribed at the same time, on the same or different channels - */ -export class EventSubscriber, V=T> { - _callbacks = new Array>(); - _channelCallbacks = new Map>>(); - lastResponse: Event | null = null; - lastResponseTime = 0; - control?: Link; - - constructor( - private event: EventClass | ChannelEventClass, - private prehandler?: (e: T) => V, - control?: Link - ) { - const entry = Link._eventsByClass.get(this.event); - if (!entry) { - throw new Error(`Unregistered Event class ${this.event.name}`); - } - if (control) { - this.connectControl(control); - } - } - - /** - * Handle incoming events and distribute it to the correct callbacks - * @param response - event from subscribed resource - * @internal - */ - async _handle(response: Event | ChannelEvent) { - this.lastResponse = response; - this.lastResponseTime = Date.now(); - const value = this.prehandler ? this.prehandler(response as T) : response as any; - for (let callback of this._callbacks) { - callback(value); - } - const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null; - if (channel) { - const callbacks = this._channelCallbacks.get(channel); - if (callbacks) { - for (let callback of callbacks) { - callback(value); + eventData.subscriptions.add(link); + if (eventData.subscriptionUpdate) { + const eventReplay = await eventData.subscriptionUpdate(event, src, dst); + if (eventReplay) { + link.send(eventReplay); } } } } +} + +export interface SubscribableValue { + id: number | string, + updatedAt: number, + isDeleted: boolean, +} + +export interface EventSubscribable extends Event { + updates: V[], +} + +/** + * Component for subscribing to and tracking updates of a remote resource + * Multiple handlers can be subscribed at the same time + */ +export class EventSubscriber< + T extends EventSubscribable, + K extends string | number = T["updates"][number]["id"], + V extends SubscribableValue = T["updates"][number], +> { + _callbacks = new Array>(); + /** Values of the subscribed resource */ + values = new Map(); + /** True if this subscriber is currently synced with the source */ + synced = false; + _snapshot?: readonly [ReadonlyMap>, boolean]; + _snapshotLastUpdated = 0; + lastResponseTime = -1; + + constructor( + private Event: EventClass, + public control: Link, + ) { + control.handle(Event, this._handle.bind(this)); + control.connector.on("connect", () => { + this._updateSubscription(); + }); + control.connector.on("close", () => { + if (this.synced) { + this.synced = false; + for (let callback of this._callbacks) { + callback([], this.synced); + } + } + }); + } /** - * If a control link was not connected at creation, it can be connected - * here (normally during onControllerConnectionEvent) - * @param control - Control link to associate with. + * Handle incoming events and distribute it to the correct callbacks + * @param event - event from subscribed resource + * @internal */ - async connectControl(control: Link) { - if (this.control === control) { - return; + async _handle(event: EventSubscribable) { + for (const value of event.updates) { + this.lastResponseTime = Math.max(this.lastResponseTime, value.updatedAt); + if (value.isDeleted) { + this.values.delete(value.id as K); + } else { + this.values.set(value.id as K, value); + } + } + for (let callback of this._callbacks) { + callback(event.updates, this.synced); } - this.control = control; - this.control.handle(this.event, this._handle.bind(this)); - await this._updateSubscription(); } /** * Subscribe to receive all event notifications * @param handler - - * callback invoked whenever the subscribed resource changes. + * callback invoked whenever the subscribed resource changes or the + * synced property changes, in which case the updates will be empty. + * @returns function that will unsubscribe from notifications */ - async subscribe(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); - await this._updateSubscription(); + if (this._callbacks.length === 1) { + this._updateSubscription(); + } + return () => { + // During a page transition the components currently rendered + // are unmounted and then the components for the new page is + // mounted. This means that if a resource is used by both pages + // it is first unsubscribed by the unmounted component causing + // the callbacks count to go to zero and a subscription update + // to be sent, and then subscribed by the mounted component + // causing another subscription update to be sent. + + // By delaying the unsubscription here the subscription happens + // before the unsubscription, thus preventing the redundant + // updates from being sent out. + setImmediate(() => { + let index = this._callbacks.lastIndexOf(handler); + if (index === -1) { + return; + } + this._callbacks.splice(index, 1); + if (this._callbacks.length === 0) { + this._updateSubscription(); + } + }); + }; } /** - * Subscribe to receive event notifications for the given channel - * @param channel - Channel to subscribe to. - * @param handler - - * callback invoked whenever the subscribed resource changes on the - * given channel. + * Obtain a snapshot of the current state of the tracked resource + * @returns tuple of values map snapshot and synced property. */ - async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { - if (!this._channelCallbacks.get(channel)) { - this._channelCallbacks.set(channel, []); + getSnapshot() { + if (this._snapshotLastUpdated !== this.lastResponseTime) { + this._snapshotLastUpdated = this.lastResponseTime; + this._snapshot = [new Map(this.values), this.synced]; } - this._channelCallbacks.get(channel)!.push(handler); - await this._updateSubscription(); - } - - /** - * Unsubscribe from receiving all event notifications, does not effect - * the subscriptions of other handlers. - * - * @param handler - A callback previously passed to {@link subscribe} - * - * @throws {Error} - * If your handler is not subscribed to all channels, does not - * accept anonymous functions - */ - async unsubscribe(handler: EventSubscriberCallback) { - let index = this._callbacks.lastIndexOf(handler); - if (index === -1) { - throw new Error("handler is not registered"); - } - - this._callbacks.splice(index, 1); - await this._updateSubscription(); - } - - /** - * Unsubscribe from receiving event notifications for a given channel, - * does not effect the subscriptions of other handlers. - * - * @param channel - A channel previously passed to {@link subscribeToChannel} - * @param handler - A callback previously passed to {@link subscribeToChannel} - * - * @throws {Error} - * If your handler is not subscribed to the channel or does not - * accept anonymous functions - */ - async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { - const channelCallbacks = this._channelCallbacks.get(channel); - if (!channelCallbacks) { - throw new Error("handler is not registered"); - } - - let index = channelCallbacks.lastIndexOf(handler); - if (index === -1) { - throw new Error("handler is not registered"); - } - - if (channelCallbacks.length === 1) { - this._channelCallbacks.delete(channel); - } else { - channelCallbacks.splice(index, 1); - } - await this._updateSubscription(); + return this._snapshot!; } /** * Update the subscription with the controller based on current handler counts */ async _updateSubscription() { - if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) { + if (!(this.control.connector as WebSocketClientConnector).connected) { return; } - const entry = Link._eventsByClass.get(this.event)!; + const entry = Link._eventsByClass.get(this.Event)!; - const response = await this.control.send(new SubscriptionRequest( - entry.name, - this._callbacks.length > 0, - [...this._channelCallbacks.keys()], - this.lastResponseTime - )); - - if (response.eventReplay) { - await this._handle(response.eventReplay as any); + try { + await this.control.send(new SubscriptionRequest( + entry.name, + this._callbacks.length > 0, + this.lastResponseTime + )); + this.synced = this._callbacks.length > 0; + this._snapshotLastUpdated = 0; + for (let callback of this._callbacks) { + callback([], this.synced); + } + } catch (err: any) { + logger.error( + `Unexpected error updating ${entry.name} subscription:\n${err.stack}` + ); } } } diff --git a/packages/web_ui/src/components/AssignInstanceModal.tsx b/packages/web_ui/src/components/AssignInstanceModal.tsx index b033469f..f6525522 100644 --- a/packages/web_ui/src/components/AssignInstanceModal.tsx +++ b/packages/web_ui/src/components/AssignInstanceModal.tsx @@ -6,7 +6,7 @@ import * as lib from "@clusterio/lib"; import ControlContext from "./ControlContext"; import { notifyErrorHandler } from "../util/notify"; -import { useHostList } from "../model/host"; +import { useHosts } from "../model/host"; const { Paragraph } = Typography; @@ -20,7 +20,7 @@ type AssignInstanceModalProps = { }; export default function AssignInstanceModal(props: AssignInstanceModalProps) { let [open, setOpen] = useState(false); - let [hostList] = useHostList(); + let [hosts] = useHosts(); let [applying, setApplying] = useState(false); let [form] = Form.useForm(); let control = useContext(ControlContext); @@ -80,7 +80,7 @@ export default function AssignInstanceModal(props: AssignInstanceModalProps) { Unassigned - {hostList.map((host) => instance.assignedHost === hostId)); - instanceList = instanceList.filter(instance => instance.assignedHost === hostId); + let nav = [{ name: "Hosts", path: "/hosts" }, { name: host?.name ?? String(hostId) }]; + if (!host) { + if (!synced) { + return ; + } - let nav = [{ name: "Hosts", path: "/hosts" }, { name: host.name || String(hostId) }]; - if (host.loading) { - return ; - } - - if (host.missing) { return

Host with id {hostId} was not found on the controller.

@@ -83,7 +81,7 @@ export default function HostViewPage() { {account.hasPermission("core.instance.list") && <> Instances - + } {account.hasPermission("core.log.follow") && <> Console diff --git a/packages/web_ui/src/components/HostsPage.tsx b/packages/web_ui/src/components/HostsPage.tsx index 5d59c4af..cbec5dc2 100644 --- a/packages/web_ui/src/components/HostsPage.tsx +++ b/packages/web_ui/src/components/HostsPage.tsx @@ -10,7 +10,7 @@ import ControlContext from "./ControlContext"; import PageHeader from "./PageHeader"; import PageLayout from "./PageLayout"; import PluginExtra from "./PluginExtra"; -import { useHostList } from "../model/host"; +import { useHosts } from "../model/host"; import notify, { notifyErrorHandler } from "../util/notify"; const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare; @@ -184,7 +184,7 @@ function CopyButton({ text, message }: { text:string, message:string }) { export default function HostsPage() { let account = useAccount(); let navigate = useNavigate(); - let [hostList] = useHostList(); + let [hosts] = useHosts(); return Number(a.connected) - Number(b.connected), }, ]} - dataSource={hostList} + dataSource={[...hosts.values()]} rowKey={host => host.id} pagination={false} onRow={(record, rowIndex) => ({ diff --git a/packages/web_ui/src/components/InstanceList.tsx b/packages/web_ui/src/components/InstanceList.tsx index feec3335..09868175 100644 --- a/packages/web_ui/src/components/InstanceList.tsx +++ b/packages/web_ui/src/components/InstanceList.tsx @@ -6,7 +6,7 @@ import type { SizeType } from "antd/es/config-provider/SizeContext"; import type { ColumnsType } from "antd/es/table"; import { useAccount } from "../model/account"; -import { useHostList } from "../model/host"; +import { useHosts } from "../model/host"; import InstanceStatusTag from "./InstanceStatusTag"; import StartStopInstanceButton from "./StartStopInstanceButton"; import { InstanceDetails } from "@clusterio/lib"; @@ -14,7 +14,7 @@ import { InstanceDetails } from "@clusterio/lib"; const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare; type InstanceListProps = { - instances: InstanceDetails[]; + instances: ReadonlyMap>; size?: SizeType; hideAssignedHost?: boolean; }; @@ -22,28 +22,27 @@ type InstanceListProps = { export default function InstanceList(props: InstanceListProps) { let account = useAccount(); let navigate = useNavigate(); - let [hostList] = useHostList(); + let [hosts] = useHosts(); - function hostName(hostId?: number|null) { - if (hostId === null ||hostId === undefined) { + function hostName(hostId?: number) { + if (hostId === undefined) { return ""; } - let host = hostList.find(s => s.id === hostId); - if (host) { - return host.name; - } - return String(hostId); + return hosts.get(hostId)?.name ?? String(hostId); } function instancePublicAddress(instance: InstanceDetails) { - let host = hostList.find(s => s.id === instance.assignedHost); - if (host && host.publicAddress) { - if (instance.gamePort) { - return `${host.publicAddress}:${instance.gamePort}`; - } + if (instance.assignedHost === undefined) { + return ""; + } + let host = hosts.get(instance.assignedHost); + if (!host || !host.publicAddress) { + return ""; + } + if (instance.gamePort === undefined) { return host.publicAddress; } - return null; + return `${host.publicAddress}:${instance.gamePort}`; } let columns: ColumnsType = [ @@ -78,7 +77,7 @@ export default function InstanceList(props: InstanceListProps) { /> : ""; }, - sorter: (a, b) => strcmp(instancePublicAddress(a)||"", instancePublicAddress(b)||""), + sorter: (a, b) => strcmp(instancePublicAddress(a), instancePublicAddress(b)), responsive: ["lg"], }, { @@ -109,7 +108,7 @@ export default function InstanceList(props: InstanceListProps) { return instance["id"]} pagination={false} onRow={(record, rowIndex) => ({ diff --git a/packages/web_ui/src/components/InstanceViewPage.tsx b/packages/web_ui/src/components/InstanceViewPage.tsx index e8c72614..13258761 100644 --- a/packages/web_ui/src/components/InstanceViewPage.tsx +++ b/packages/web_ui/src/components/InstanceViewPage.tsx @@ -20,15 +20,15 @@ import StartStopInstanceButton from "./StartStopInstanceButton"; import LoadScenarioModal from "./LoadScenarioModal"; import SavesList from "./SavesList"; import { notifyErrorHandler } from "../util/notify"; -import { InstanceState, useInstance } from "../model/instance"; -import { HostState, useHost } from "../model/host"; +import { useInstance } from "../model/instance"; +import { useHost } from "../model/host"; import InstanceStatusTag from "./InstanceStatusTag"; const { Title } = Typography; type InstanceDescriptionProps = { - host: HostState; - instance: InstanceState; + host?: Readonly; + instance: Readonly; }; function InstanceDescription(props: InstanceDescriptionProps) { let account = useAccount(); @@ -42,7 +42,7 @@ function InstanceDescription(props: InstanceDescriptionProps) { {!assigned ? Unassigned - : host.name || instance.assignedHost + : host?.name ?? instance.assignedHost } {account.hasPermission("core.instance.assign") && ; } -function InstanceButtons(props: { instance: InstanceState }) { +function InstanceButtons(props: { instance: lib.InstanceDetails }) { let account = useAccount(); let control = useContext(ControlContext); let navigate = useNavigate(); @@ -169,18 +169,18 @@ export default function InstanceViewPage() { let navigate = useNavigate(); let account = useAccount(); - let [instance] = useInstance(instanceId); - let [host] = useHost(Number(instance.assignedHost)); + let [instance, synced] = useInstance(instanceId); + let [host] = useHost(instance?.assignedHost); - let nav = [{ name: "Instances", path: "/instances" }, { name: instance.name || "Unknown" }]; - if (instance.loading) { - return ; - } + let nav = [{ name: "Instances", path: "/instances" }, { name: instance?.name ?? String(instanceId) }]; + if (!instance) { + if (!synced) { + return ; + } - if (instance.missing || instance.status === "deleted") { return Instance with id {instanceId} was not found on the controller.} type="warning" diff --git a/packages/web_ui/src/components/InstancesPage.tsx b/packages/web_ui/src/components/InstancesPage.tsx index eba88f67..351d6fad 100644 --- a/packages/web_ui/src/components/InstancesPage.tsx +++ b/packages/web_ui/src/components/InstancesPage.tsx @@ -9,7 +9,7 @@ import ControlContext from "./ControlContext"; import PageHeader from "./PageHeader"; import PageLayout from "./PageLayout"; import PluginExtra from "./PluginExtra"; -import { useInstanceList } from "../model/instance"; +import { useInstances } from "../model/instance"; import InstanceList from "./InstanceList"; import { notifyErrorHandler } from "../util/notify"; @@ -62,7 +62,7 @@ function CreateInstanceButton() { export default function InstancesPage() { let control = useContext(ControlContext); let account = useAccount(); - let [instanceList] = useInstanceList(); + let [instances] = useInstances(); return {account.hasPermission("core.instance.create") && } {account.hasPermission("core.instance.start") - && } {account.hasPermission("core.instance.stop") - && } {account.hasPermission("core.role.delete") && @@ -153,17 +153,17 @@ function TransferModal(props: ModalProps) { (option?.title.toLowerCase().indexOf(input.toLowerCase())??-1) >= 0 )} > - {instanceList.filter( + {[...instances.values()].filter( instance => instance["id"] !== props.instanceId ).map((instance) => - {instance["name"]} - {instance["status"] === "unassigned" && " (unassigned)"} - {instance["status"] === "unknown" && " (offline)"} + {instance.name} + {instance.status === "unassigned" && " (unassigned)"} + {instance.status === "unknown" && " (offline)"} )} @@ -198,10 +198,10 @@ type File = { status: string; } -export default function SavesList(props: { instance: InstanceState }) { +export default function SavesList(props: { instance: lib.InstanceDetails }) { let account = useAccount(); let control = useContext(ControlContext); - let saves = useSaves(props.instance.id); + let [saves] = useSavesOfInstance(props.instance.id); let [starting, setStarting] = useState(false); let [uploadingFiles, setUploadingFiles] = useState([]); @@ -234,7 +234,7 @@ export default function SavesList(props: { instance: InstanceState }) { defaultSortOrder: "descend", }, ]} - dataSource={saves} + dataSource={[...saves.values()]} rowKey={save => save.name} pagination={false} expandable={{ diff --git a/packages/web_ui/src/components/StartStopInstanceButton.tsx b/packages/web_ui/src/components/StartStopInstanceButton.tsx index 2920721a..2948675a 100644 --- a/packages/web_ui/src/components/StartStopInstanceButton.tsx +++ b/packages/web_ui/src/components/StartStopInstanceButton.tsx @@ -5,11 +5,10 @@ import * as lib from "@clusterio/lib"; import ControlContext from "./ControlContext"; import { notifyErrorHandler } from "../util/notify"; -import type { InstanceState } from "../model/instance"; import { ButtonProps } from "antd/es/button"; type StartStopInstanceButtonProps = { - instance: InstanceState; + instance: lib.InstanceDetails; onFinish?: () => void; buttonProps?: ButtonProps; }; @@ -23,15 +22,15 @@ export default function StartStopInstanceButton(props: StartStopInstanceButtonPr let action; if (props.instance.status === "stopped") { action = control.sendTo( - { instanceId: props.instance.id! }, + { instanceId: props.instance.id }, new lib.InstanceStartRequest(undefined), ).catch( notifyErrorHandler("Error starting instance") ); - } else if (["starting", "running"].includes(props.instance.status!)) { + } else if (["starting", "running"].includes(props.instance.status)) { action = control.sendTo( - { instanceId: props.instance.id! }, + { instanceId: props.instance.id }, new lib.InstanceStopRequest(), ).catch( notifyErrorHandler("Error stopping instance") @@ -54,7 +53,7 @@ export default function StartStopInstanceButton(props: StartStopInstanceButtonPr {...(props.buttonProps || {})} loading={switching} type="primary" - disabled={!["starting", "running", "stopped"].includes(props.instance.status!)} + disabled={!["starting", "running", "stopped"].includes(props.instance.status)} onClick={onClick} > {props.instance.status === "stopped" ? "Start" : "Stop"} diff --git a/packages/web_ui/src/components/UserViewPage.tsx b/packages/web_ui/src/components/UserViewPage.tsx index 5d41a4ac..07620da2 100644 --- a/packages/web_ui/src/components/UserViewPage.tsx +++ b/packages/web_ui/src/components/UserViewPage.tsx @@ -9,7 +9,7 @@ import DeleteOutlined from "@ant-design/icons/DeleteOutlined"; import * as lib from "@clusterio/lib"; import { useAccount } from "../model/account"; -import { useInstanceList } from "../model/instance"; +import { useInstances } from "../model/instance"; import ControlContext from "./ControlContext"; import PageHeader from "./PageHeader"; import PageLayout from "./PageLayout"; @@ -33,8 +33,8 @@ export default function UserViewPage() { let account = useAccount(); let control = useContext(ControlContext); - let [instanceList] = useInstanceList(); - let [user] = useUser(userName); + let [instances] = useInstances(); + const [user, synced] = useUser(userName); let [roles, setRoles] = useState>(new Map()); let [form] = Form.useForm(); let [rolesDirty, setRolesDirty] = useState(false); @@ -54,6 +54,7 @@ export default function UserViewPage() { if ( account.hasPermission("core.user.set_banned") && !banReasonDirty + && user && user.banReason !== form.getFieldValue("banReason") ) { form.setFieldsValue({ "banReason": user.banReason }); @@ -61,14 +62,14 @@ export default function UserViewPage() { }, [account, form, user, banReasonDirty]); let nav = [{ name: "Users", path: "/users" }, { name: userName }]; - if (user.loading) { - return - - - ; - } + if (!user) { + if (!synced) { + return + + + ; + } - if (user.missing) { return

User with name {userName} was not found on the controller.

@@ -78,7 +79,7 @@ export default function UserViewPage() { let roleSelector = (rolesError ? Promise.reject(rolesError) : Promise.resolve()) }, ]} @@ -100,9 +101,8 @@ export default function UserViewPage() { ; - function instanceName(id: number): string { - let instance = instanceList.find(i => i.id === id); - return instance ? instance.name : String(id); + function instanceName(id: number) { + return instances.get(id)?.name ?? String(id); } return diff --git a/packages/web_ui/src/components/UsersPage.tsx b/packages/web_ui/src/components/UsersPage.tsx index e0966187..b7469d40 100644 --- a/packages/web_ui/src/components/UsersPage.tsx +++ b/packages/web_ui/src/components/UsersPage.tsx @@ -11,7 +11,7 @@ import { formatDuration } from "../util/time_format"; import PageHeader from "./PageHeader"; import PageLayout from "./PageLayout"; import PluginExtra from "./PluginExtra"; -import { formatLastSeen, sortLastSeen, useUserList } from "../model/user"; +import { formatLastSeen, sortLastSeen, useUsers } from "../model/user"; const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare; @@ -60,7 +60,7 @@ export default function UsersPage() { let account = useAccount(); let control = useContext(ControlContext); let navigate = useNavigate(); - let [userList] = useUserList(); + let [users] = useUsers(); let [roles, setRoles] = useState(new Map()); @@ -117,7 +117,7 @@ export default function UsersPage() { responsive: ["lg"], }, ]} - dataSource={userList} + dataSource={[...users.values()]} pagination={false} rowKey={user => user.name} onRow={(user, rowIndex) => ({ diff --git a/packages/web_ui/src/model/host.ts b/packages/web_ui/src/model/host.ts index d211b428..c6e28ff3 100644 --- a/packages/web_ui/src/model/host.ts +++ b/packages/web_ui/src/model/host.ts @@ -1,87 +1,13 @@ -import { useEffect, useContext, useState } from "react"; +import { useCallback, useContext, useSyncExternalStore } from "react"; import ControlContext from "../components/ControlContext"; -import * as lib from "@clusterio/lib"; - - -export type HostState = Partial & { - loading?: boolean, - present?: boolean, - missing?: boolean, +export function useHost(id?: number) { + const [hosts, synced] = useHosts(); + return [id !== undefined ? hosts.get(id) : undefined, synced] as const; } -export function useHost(id: number): [ HostState, () => void ] { - let control = useContext(ControlContext); - let [host, setHost] = useState({ loading: true }); - - function updateHost() { - // XXX optimize by requesting only the host in question - control.send( - new lib.HostListRequest() - ).then((hosts: lib.HostDetails[]) => { - let match = hosts.find(i => i.id === id); - if (!match) { - setHost({ missing: true }); - } else { - setHost({ ...match, present: true }); - } - }); - } - - useEffect(() => { - if (!Number.isInteger(id)) { - setHost({ missing: true }); - return undefined; - } - updateHost(); - - function updateHandler(newHost: lib.HostDetails) { - setHost({ ...newHost, loading: false, missing: false, present: true }); - } - - control.hostUpdate.subscribeToChannel(id, updateHandler); - return () => { - control.hostUpdate.unsubscribeFromChannel(id, updateHandler); - }; - }, [id]); - - return [host, updateHost]; -} - -export function useHostList() { - let control = useContext(ControlContext); - let [hostList, setHostList] = useState([]); - - function updateHostList() { - control.send( - new lib.HostListRequest() - ).then((hosts: lib.HostDetails[]) => { - setHostList(hosts); - }); - } - - useEffect(() => { - updateHostList(); - - function updateHandler(newHost: lib.HostDetails) { - setHostList(oldList => { - let newList = oldList.concat(); - let index = newList.findIndex(s => s.id === newHost.id); - if (index !== -1) { - newList[index] = newHost; - } else { - newList.push(newHost); - } - return newList; - }); - } - - control.hostUpdate.subscribe(updateHandler); - return () => { - control.hostUpdate.unsubscribe(updateHandler); - }; - }, []); - - - return [hostList]; +export function useHosts() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.hosts.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.hosts.getSnapshot()); } diff --git a/packages/web_ui/src/model/instance.ts b/packages/web_ui/src/model/instance.ts index c262c2b9..41bb8b99 100644 --- a/packages/web_ui/src/model/instance.ts +++ b/packages/web_ui/src/model/instance.ts @@ -1,86 +1,13 @@ -import { useEffect, useContext, useState } from "react"; +import { useCallback, useContext, useSyncExternalStore } from "react"; import ControlContext from "../components/ControlContext"; -import * as lib from "@clusterio/lib"; - -const { logger } = lib; - -export type InstanceState = Partial & { - loading?: boolean, - present?: boolean, - missing?: boolean, +export function useInstance(id?: number) { + const [instances, synced] = useInstances(); + return [id !== undefined ? instances.get(id) : undefined, synced] as const; } -export function useInstance(id: number): [InstanceState, ()=>void] { - let control = useContext(ControlContext); - let [instance, setInstance] = useState({ loading: true }); - - function updateInstance() { - control.send(new lib.InstanceDetailsGetRequest(id)).then(updatedInstance => { - setInstance({ ...updatedInstance, present: true }); - }).catch((err: any) => { - logger.error(`Failed to get instance: ${err}`); - setInstance({ missing: true }); - }); - } - - useEffect(() => { - if (!Number.isInteger(id)) { - setInstance({ missing: true }); - return undefined; - } - updateInstance(); - - function updateHandler(newInstance: lib.InstanceDetails) { - setInstance({ ...newInstance, present: true }); - } - - control.instanceUpdate.subscribeToChannel(id, updateHandler); - return () => { - control.instanceUpdate.unsubscribeFromChannel(id, updateHandler); - }; - }, [id]); - - return [instance, updateInstance]; -} - -export function useInstanceList() { - let control = useContext(ControlContext); - let [instanceList, setInstanceList] = useState([]); - - function updateInstanceList() { - control.send(new lib.InstanceDetailsListRequest()).then(instances => { - setInstanceList(instances); - }).catch(err => { - logger.error(`Failed to list instances:\n${err}`); - }); - } - - useEffect(() => { - updateInstanceList(); - - function updateHandler(newInstance: lib.InstanceDetails) { - setInstanceList(oldList => { - let newList = oldList.concat(); - let index = newList.findIndex(s => s.id === newInstance.id); - if (newInstance.status !== "deleted") { - if (index !== -1) { - newList[index] = newInstance; - } else { - newList.push(newInstance); - } - } else if (index !== -1) { - newList.splice(index, 1); - } - return newList; - }); - } - - control.instanceUpdate.subscribe(updateHandler); - return () => { - control.instanceUpdate.unsubscribe(updateHandler); - }; - }, []); - - return [instanceList]; +export function useInstances() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.instances.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.instances.getSnapshot()); } diff --git a/packages/web_ui/src/model/mod_pack.ts b/packages/web_ui/src/model/mod_pack.ts index 0d610f29..927a9a75 100644 --- a/packages/web_ui/src/model/mod_pack.ts +++ b/packages/web_ui/src/model/mod_pack.ts @@ -1,85 +1,13 @@ -import { useContext, useEffect, useState } from "react"; - -import * as lib from "@clusterio/lib"; - +import { useCallback, useContext, useSyncExternalStore } from "react"; import ControlContext from "../components/ControlContext"; -const { logger } = lib; - - -export type StateLoadStatus = { - loading?: boolean; - present?: boolean; - missing?: boolean; -}; - -export function useModPack(id: number) { - let control = useContext(ControlContext); - let [modPack, setModPack] = useState({ loading: true }); - - function updateModPack() { - control.send(new lib.ModPackGetRequest(id)).then(updatedModPack => { - setModPack(updatedModPack); - }).catch(err => { - logger.error(`Failed to get mod pack: ${err}`); - setModPack({ missing: true }); - }); - } - - useEffect(() => { - if (typeof id !== "number") { - setModPack({ missing: true }); - return undefined; - } - updateModPack(); - - control.modPackUpdate.subscribeToChannel(id, setModPack); - return () => { - control.modPackUpdate.unsubscribeFromChannel(id, setModPack); - }; - }, [id]); - - return [modPack]; +export function useModPack(id?: number) { + const [modPacks, synced] = useModPacks(); + return [id !== undefined ? modPacks.get(id) : undefined, synced] as const; } -export function useModPackList() { - let control = useContext(ControlContext); - let [modPackList, setModPackList] = useState([]); - - function updateModPackList() { - control.send(new lib.ModPackListRequest()).then(modPacks => { - setModPackList(modPacks); - }).catch(err => { - logger.error(`Failed to list mod packs:\n${err}`); - }); - } - - useEffect(() => { - updateModPackList(); - - function updateHandler(newModPack: lib.ModPack) { - setModPackList(oldList => { - let newList = oldList.concat(); - let index = newList.findIndex(u => u.id === newModPack.id); - if (!newModPack.isDeleted) { - if (index !== -1) { - newList[index] = newModPack; - } else { - newList.push(newModPack); - } - } else if (index !== -1) { - newList.splice(index, 1); - } - return newList; - }); - } - - control.modPackUpdate.subscribe(updateHandler); - return () => { - control.modPackUpdate.unsubscribe(updateHandler); - }; - }, []); - - return [modPackList]; - +export function useModPacks() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.modPacks.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.modPacks.getSnapshot()); } diff --git a/packages/web_ui/src/model/mods.ts b/packages/web_ui/src/model/mods.ts index 0d131fae..4c618d38 100644 --- a/packages/web_ui/src/model/mods.ts +++ b/packages/web_ui/src/model/mods.ts @@ -1,49 +1,13 @@ -import { useContext, useEffect, useState } from "react"; - -import * as lib from "@clusterio/lib"; - +import { useCallback, useContext, useSyncExternalStore } from "react"; import ControlContext from "../components/ControlContext"; -const { logger } = lib; - -export function useModList() { - let control = useContext(ControlContext); - let [modList, setModList] = useState([]); - - function updateModList() { - control.send(new lib.ModListRequest()).then(mods => { - setModList(mods); - }).catch(err => { - logger.error(`Failed to list mods:\n${err}`); - }); - } - - useEffect(() => { - updateModList(); - - function updateHandler(newMod: lib.ModInfo) { - setModList(oldList => { - let newList = oldList.concat(); - let index = newList.findIndex(mod => mod.name === newMod.name && mod.version === newMod.version); - if (!newMod.isDeleted) { - if (index !== -1) { - newList[index] = newMod; - } else { - newList.push(newMod); - } - } else if (index !== -1) { - newList.splice(index, 1); - } - return newList; - }); - } - - control.modUpdate.subscribe(updateHandler); - return () => { - control.modUpdate.unsubscribe(updateHandler); - }; - }, []); - - return [modList]; +export function useMod(id?: string) { + const [mods, synced] = useMods(); + return [id !== undefined ? mods.get(id) : undefined, synced] as const; } +export function useMods() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.mods.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.mods.getSnapshot()); +} diff --git a/packages/web_ui/src/model/saves.ts b/packages/web_ui/src/model/saves.ts index 9abb1cfd..f92b5ce5 100644 --- a/packages/web_ui/src/model/saves.ts +++ b/packages/web_ui/src/model/saves.ts @@ -1,39 +1,14 @@ -import { useEffect, useContext, useState } from "react"; +import { useCallback, useContext, useSyncExternalStore } from "react"; import ControlContext from "../components/ControlContext"; -import * as lib from "@clusterio/lib"; - -const { logger } = lib; - -export function useSaves(instanceId?: number): lib.SaveDetails[] { - let control = useContext(ControlContext); - let [saves, setSaves] = useState([]); - - function updateSaves() { - control.sendTo({ instanceId: instanceId! }, new lib.InstanceListSavesRequest()).then(updatedSaves => { - setSaves(updatedSaves); - }).catch(err => { - logger.error(`Failed to list instance saves: ${err}`); - setSaves([]); - }); - } - - useEffect(() => { - if (!Number.isInteger(instanceId)) { - setSaves([]); - return undefined; - } - updateSaves(); - - function updateHandler(data: lib.InstanceSaveListUpdateEvent) { - setSaves(data.saves); - } - - control.saveListUpdate.subscribeToChannel(instanceId!, updateHandler); - return () => { - control.saveListUpdate.unsubscribeFromChannel(instanceId!, updateHandler); - }; - }, [instanceId]); - - return saves; +export function useSavesOfInstance(instanceId?: number) { + const [saves, synced] = useSaves(); + const savesOfInstance = new Map([...saves].filter(([_id, save]) => save.instanceId === instanceId)); + return [savesOfInstance, synced] as const; +} + +export function useSaves() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.saves.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.saves.getSnapshot()); } diff --git a/packages/web_ui/src/model/user.tsx b/packages/web_ui/src/model/user.tsx index 6ed8431b..f0ff2b7e 100644 --- a/packages/web_ui/src/model/user.tsx +++ b/packages/web_ui/src/model/user.tsx @@ -1,31 +1,29 @@ -import React, { useEffect, useContext, useState } from "react"; +import React, { useCallback, useContext, useSyncExternalStore } from "react"; import { Tag } from "antd"; import ControlContext from "../components/ControlContext"; import * as lib from "@clusterio/lib"; -const { logger } = lib; - -function calculateLastSeen(user: RawUserState, instanceId?: number) { +function calculateLastSeen(user: lib.User, instanceId?: number) { let stats; if (instanceId === undefined) { stats = user.playerStats; } else { - stats = (user.instanceStats || new Map()).get(instanceId); + stats = user.instanceStats.get(instanceId); if (!stats) { return undefined; } } - if (stats.lastLeaveAt > stats.lastJoinAt) { - return stats.lastLeaveAt; + if ((stats.lastLeaveAt?.getTime() ?? 0) > (stats.lastJoinAt?.getTime() ?? 0)) { + return stats.lastLeaveAt?.getTime(); } if (stats.lastJoinAt) { - return stats.lastLeaveAt; + return stats.lastLeaveAt?.getTime(); } return undefined; } -export function formatLastSeen(user: RawUserState, instanceId?: number) { +export function formatLastSeen(user: lib.User, instanceId?: number) { if (user.instances && [...user.instances].some(id => instanceId === undefined || id === instanceId)) { return Online; } @@ -36,8 +34,8 @@ export function formatLastSeen(user: RawUserState, instanceId?: number) { return new Date(lastSeen).toLocaleString(); } -export function sortLastSeen(userA:RawUserState, userB:RawUserState, instanceIdA?: number, instanceIdB?: number) { - function epoch(user:RawUserState, instanceId?: number) { +export function sortLastSeen(userA: lib.User, userB: lib.User, instanceIdA?: number, instanceIdB?: number) { + function epoch(user: lib.User, instanceId?: number) { return user.instances && [...user.instances].some(id => instanceId === undefined || id === instanceId); } @@ -52,82 +50,13 @@ export function sortLastSeen(userA:RawUserState, userB:RawUserState, instanceIdA return lastSeenA - lastSeenB; } -export type RawUserState = Partial & { - loading?: boolean; - present?: boolean; - missing?: boolean; -}; - -export function useUser(name: string): [RawUserState, () => void] { - let control = useContext(ControlContext); - let [user, setUser] = useState({ loading: true }); - - function updateUser() { - control.send(new lib.UserGetRequest(name)).then(updatedUser => { - setUser({ ...updatedUser, present: true }); - }).catch(err => { - logger.error(`Failed to get user: ${err}`); - setUser({ missing: true }); - }); - } - - useEffect(() => { - if (typeof name !== "string") { - setUser({ missing: true }); - return undefined; - } - updateUser(); - - function updateHandler(newUser: lib.User) { - setUser({ ...newUser, present: true }); - } - - control.userUpdate.subscribeToChannel(name, updateHandler); - return () => { - control.userUpdate.unsubscribeFromChannel(name, updateHandler); - }; - }, [name]); - - return [user, updateUser]; +export function useUser(name?: string) { + const [users, synced] = useUsers(); + return [name !== undefined ? users.get(name) : undefined, synced] as const; } -export function useUserList() { - let control = useContext(ControlContext); - let [userList, setUserList] = useState([]); - - function updateUserList() { - control.send(new lib.UserListRequest()).then(users => { - setUserList(users); - }).catch(err => { - logger.error(`Failed to list users:\n${err}`); - }); - } - - useEffect(() => { - updateUserList(); - - function updateHandler(newUser: lib.User) { - setUserList(oldList => { - let newList = oldList.concat(); - let index = newList.findIndex(u => u.name === newUser.name); - if (!newUser.isDeleted) { - if (index !== -1) { - newList[index] = newUser; - } else { - newList.push(newUser); - } - } else if (index !== -1) { - newList.splice(index, 1); - } - return newList; - }); - } - - control.userUpdate.subscribe(updateHandler); - return () => { - control.userUpdate.unsubscribe(updateHandler); - }; - }, []); - - return [userList]; +export function useUsers() { + const control = useContext(ControlContext); + const subscribe = useCallback((callback: () => void) => control.users.subscribe(callback), [control]); + return useSyncExternalStore(subscribe, () => control.users.getSnapshot()); } diff --git a/packages/web_ui/src/util/websocket.ts b/packages/web_ui/src/util/websocket.ts index 02b24fbd..06f796e9 100644 --- a/packages/web_ui/src/util/websocket.ts +++ b/packages/web_ui/src/util/websocket.ts @@ -7,7 +7,7 @@ const { logFilter, logger } = lib; type accountHandler = (account: lib.AccountDetails) => void; type hostHandler = (hostDetails: lib.HostDetails) => void; type instanceHandler = (instanceDetails: lib.InstanceDetails) => void; -type saveListHandler = (saveListEvent: lib.InstanceSaveListUpdateEvent) => void; +type saveListHandler = (saveListEvent: lib.SaveDetails) => void; type modPackHandler = (modPack: lib.ModPack) => void; type modInfoHandler = (modInfo: lib.ModInfo) => void; type userHandler = (rawUser: lib.User) => void; @@ -60,29 +60,12 @@ export class Control extends lib.Link { logHandlers: Map = new Map(); /** Updates handled by the subscription service */ - hostUpdate = new lib.EventSubscriber( - lib.HostUpdateEvent, event => event.update, - ); - - instanceUpdate = new lib.EventSubscriber( - lib.InstanceDetailsUpdateEvent, event => event.details, - ); - - saveListUpdate = new lib.EventSubscriber( - lib.InstanceSaveListUpdateEvent, - ); - - modPackUpdate = new lib.EventSubscriber( - lib.ModPackUpdateEvent, event => event.modPack, - ); - - modUpdate = new lib.EventSubscriber( - lib.ModUpdateEvent, event => event.mod, - ); - - userUpdate = new lib.EventSubscriber( - lib.UserUpdateEvent, event => event.user, - ); + hosts = new lib.EventSubscriber(lib.HostUpdatesEvent, this); + instances = new lib.EventSubscriber(lib.InstanceDetailsUpdatesEvent, this); + saves = new lib.EventSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this); + modPacks = new lib.EventSubscriber(lib.ModPackUpdatesEvent, this); + mods = new lib.EventSubscriber(lib.ModUpdatesEvent, this); + users = new lib.EventSubscriber(lib.UserUpdatesEvent, this); declare connector: ControlConnector; @@ -95,24 +78,6 @@ export class Control extends lib.Link { this.accountName = data.account.name; this.accountRoles = data.account.roles; this.emitAccountUpdate(); - this.hostUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating host subscriptions:\n${err.stack}` - )); - this.instanceUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating instance subscriptions:\n${err.stack}` - )); - this.saveListUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating save list subscriptions:\n${err.stack}` - )); - this.modPackUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating mod pack subscriptions:\n${err.stack}` - )); - this.modUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating mod subscriptions:\n${err.stack}` - )); - this.userUpdate.connectControl(this).catch(err => logger.error( - `Unexpected error updating user subscriptions:\n${err.stack}` - )); this.updateLogSubscriptions().catch(err => logger.error( `Unexpected error updating log subscriptions:\n${err.stack}` )); diff --git a/test/controller/InstanceInfo.js b/test/controller/InstanceInfo.js new file mode 100644 index 00000000..4098c30a --- /dev/null +++ b/test/controller/InstanceInfo.js @@ -0,0 +1,40 @@ +"use strict"; +const assert = require("assert").strict; + +const { InstanceInfo } = require("@clusterio/controller"); +const lib = require("@clusterio/lib"); + + +describe("controller/InstanceInfo", function() { + describe("class InstanceInfo", function() { + it("should round trip serialize", async function() { + const validate = lib.compile(InstanceInfo.jsonSchema); + function check(info) { + const json = JSON.parse(JSON.stringify(info)); + if (!validate(json)) { + throw validate.errors; + } + assert.deepEqual(InstanceInfo.fromJSON(json, info.config), info); + } + + if (!lib.InstanceConfig._finalized) { + lib.InstanceConfig.finalize(); + } + const config = new lib.InstanceConfig("controller"); + await config.init(); + + check(InstanceInfo.fromJSON({ config, status: "stopped" }, config)); + check(InstanceInfo.fromJSON({ config, status: "running" }, config)); + check(InstanceInfo.fromJSON({ config, status: "running", gamePort: 34197 }, config)); + check(InstanceInfo.fromJSON({ config, status: "running", updatedAt: Date.now() }, config)); + + // All at once + check(InstanceInfo.fromJSON({ + config, + status: "unknown", + gamePort: 10000, + updateAt: 20000, + }, config)); + }); + }); +}); diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js index b9931847..357baa09 100644 --- a/test/integration/clusterio.js +++ b/test/integration/clusterio.js @@ -315,16 +315,16 @@ describe("Integration of Clusterio", function() { describe("instance save create", function() { it("creates a save", async function() { slowTest(this); - getControl().saveListUpdates = []; + getControl().saveUpdates = []; await execCtl("instance save create test"); await checkInstanceStatus(44, "stopped"); }); }); - describe("saveListUpdateEventHandler()", function() { + describe("InstanceSaveDetailsUpdatesEvent", function() { it("should have triggered for the created save", function() { slowTest(this); - assert.equal(getControl().saveListUpdates.slice(-1)[0].saves[0].name, "world.zip"); + assert.equal(getControl().saveUpdates.slice(-1)[0].updates[0].name, "world.zip"); }); }); @@ -391,7 +391,7 @@ describe("Integration of Clusterio", function() { let savesDir = path.join("temp", "test", "instances", "test", "saves"); await fs.copy(path.join(savesDir, "world.zip"), path.join(savesDir, "_autosave1.zip")); await execCtl("instance start test"); - let saves = await getControl().sendTo({ instanceId: 44 }, new lib.InstanceListSavesRequest()); + let saves = await getControl().sendTo({ instanceId: 44 }, new lib.InstanceSaveDetailsListRequest()); let running = saves.find(s => s.loaded); assert(running.name !== "_autosave1.zip"); }); @@ -407,19 +407,19 @@ describe("Integration of Clusterio", function() { assert(log.some(info => /technobabble/.test(info.message)), "Command was not sent"); }); - it("should trigger saveListUpdate on save", async function() { + it("should trigger InstanceSaveDetailsUpdatesEvent on save", async function() { slowTest(this); - getControl().saveListUpdates = []; + getControl().saveUpdates = []; await execCtl("instance send-rcon test /server-save"); let received = false; for (let x = 0; x < 10; x++) { - if (getControl().saveListUpdates.length) { + if (getControl().saveUpdates.length) { received = true; break; } await wait(100); } - assert(received, "saveListUpdate not sent"); + assert(received, "InstanceSaveDetailsUpdatesEvent not sent"); }); }); @@ -827,10 +827,16 @@ describe("Integration of Clusterio", function() { describe("instance delete", function() { it("deletes the instance", async function() { slowTest(this); + getControl().saveUpdates = []; await execCtl("instance delete test"); assert(!await fs.exists(path.join(instancesDir, "test")), "Instance files was not deleted"); let instances = await getInstances(); assert(!instances.has(44), "instance was not deleted from controller"); + + const saveUpdates = getControl().saveUpdates; + assert.equal(saveUpdates.length, 1, "Expected one update sent on saves"); + assert(saveUpdates[0].updates.length > 0, "Expected saves update to contain at least one save updated"); + assert(saveUpdates[0].updates.every(update => update.isDeleted), "Expected all saves to be deleted"); }); }); @@ -885,6 +891,7 @@ describe("Integration of Clusterio", function() { reference.settings["runtime-global"].set("MyInt", { value: 1235 }); reference.settings["runtime-global"].set("MyDouble", { value: 12.25 }); reference.settings["runtime-per-user"].set("MyString", { value: "a-string" }); + reference.updatedAt = modPack.updatedAt; assert.deepEqual(modPack, reference); }); }); @@ -989,6 +996,7 @@ describe("Integration of Clusterio", function() { "dependencies:\n" + `size: ${stat.size}\n` + `sha1: ${hash}\n` + + `updatedAt: ${stat.mtimeMs}\n` + "isDeleted: false\n", ); }); diff --git a/test/integration/index.js b/test/integration/index.js index 8fd2585e..33db6e9c 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -26,7 +26,7 @@ class TestControl extends lib.Link { super(connector); this.hostUpdates = []; this.instanceUpdates = []; - this.saveListUpdates = []; + this.saveUpdates = []; this.modUpdates = []; this.modPackUpdates = []; this.userUpdates = []; @@ -36,56 +36,56 @@ class TestControl extends lib.Link { return; } this.send( - new lib.SubscriptionRequest(lib.HostUpdateEvent.name, true) + new lib.SubscriptionRequest(lib.HostUpdatesEvent.name, true) ).catch(err => logger.error(`Error setting host subscriptions:\n${err.stack}`)); this.send( - new lib.SubscriptionRequest(lib.InstanceDetailsUpdateEvent.name, true) + new lib.SubscriptionRequest(lib.InstanceDetailsUpdatesEvent.name, true) ).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`)); this.send( - new lib.SubscriptionRequest(lib.InstanceSaveListUpdateEvent.name, true) - ).catch(err => logger.error(`Error setting save list subscriptions:\n${err.stack}`)); + new lib.SubscriptionRequest(lib.InstanceSaveDetailsUpdatesEvent.name, true) + ).catch(err => logger.error(`Error setting save subscriptions:\n${err.stack}`)); this.send( - new lib.SubscriptionRequest(lib.ModUpdateEvent.name, true) + new lib.SubscriptionRequest(lib.ModUpdatesEvent.name, true) ).catch(err => logger.error(`Error setting mod subscriptions:\n${err.stack}`)); this.send( - new lib.SubscriptionRequest(lib.ModPackUpdateEvent.name, true) + new lib.SubscriptionRequest(lib.ModPackUpdatesEvent.name, true) ).catch(err => logger.error(`Error setting mod pack subscriptions:\n${err.stack}`)); this.send( - new lib.SubscriptionRequest(lib.UserUpdateEvent.name, true) + new lib.SubscriptionRequest(lib.UserUpdatesEvent.name, true) ).catch(err => logger.error(`Error setting user subscriptions:\n${err.stack}`)); }); this.handle(lib.AccountUpdateEvent); - this.handle(lib.HostUpdateEvent, this.handleHostUpdateEvent.bind(this)); - this.handle(lib.InstanceDetailsUpdateEvent, this.handleInstanceDetailsUpdateEvent.bind(this)); - this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this)); - this.handle(lib.ModUpdateEvent, this.handleModUpdateEvent.bind(this)); - this.handle(lib.ModPackUpdateEvent, this.handleModPackUpdateEvent.bind(this)); - this.handle(lib.UserUpdateEvent, this.handleUserUpdateEvent.bind(this)); + this.handle(lib.HostUpdatesEvent, this.handleHostUpdatesEvent.bind(this)); + this.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsUpdatesEvent.bind(this)); + this.handle(lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsUpdatesEvent.bind(this)); + this.handle(lib.ModUpdatesEvent, this.handleModUpdatesEvent.bind(this)); + this.handle(lib.ModPackUpdatesEvent, this.handleModPackUpdatesEvent.bind(this)); + this.handle(lib.UserUpdatesEvent, this.handleUserUpdatesEvent.bind(this)); } - async handleHostUpdateEvent(event) { - this.hostUpdates.push(event.update); + async handleHostUpdatesEvent(event) { + this.hostUpdates.push(...event.updates); } - async handleInstanceDetailsUpdateEvent(event) { - this.instanceUpdates.push(event.details); + async handleInstanceDetailsUpdatesEvent(event) { + this.instanceUpdates.push(...event.updates); } - async handleInstanceSaveListUpdateEvent(event) { - this.saveListUpdates.push(event); + async handleInstanceSaveDetailsUpdatesEvent(event) { + this.saveUpdates.push(event); } - async handleModUpdateEvent(event) { - this.modUpdates.push(event.mod); + async handleModUpdatesEvent(event) { + this.modUpdates.push(...event.updates); } - async handleModPackUpdateEvent(event) { - this.modPackUpdates.push(event.modPack); + async handleModPackUpdatesEvent(event) { + this.modPackUpdates.push(...event.updates); } - async handleUserUpdateEvent(event) { - this.userUpdates.push(event.user); + async handleUserUpdatesEvent(event) { + this.userUpdates.push(...event.updates); } } diff --git a/test/lib/data/ModInfo.js b/test/lib/data/ModInfo.js index 279f535f..8e0cb640 100644 --- a/test/lib/data/ModInfo.js +++ b/test/lib/data/ModInfo.js @@ -92,6 +92,7 @@ describe("lib/data/ModInfo", function() { dependencies: [], filename: "empty_mod_1.0.0.zip", size: stat.size, + updated_at: stat.mtimeMs, sha1: hash, }) ); diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index 5656717b..f345f906 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -19,20 +19,11 @@ describe("lib/subscriptions", function() { static src = ["controller", "control"]; static dst = ["controller", "control"]; static permission = null; - } - lib.Link.register(RegisteredEvent); - - class ChannelEvent { - static type = "event"; - static src = ["controller", "control"]; - static dst = ["controller", "control"]; - static permission = null; - - get subscriptionChannel() { - return "channelOne"; + constructor(updates = []) { + this.updates = updates; } } - lib.Link.register(ChannelEvent); + lib.Link.register(RegisteredEvent); class StringPermissionEvent { static type = "event"; @@ -58,46 +49,6 @@ describe("lib/subscriptions", function() { } } - describe("class SubscriptionResponse", function() { - it("should be round trip json serialisable without an event replay", function() { - const response = new lib.SubscriptionResponse(); - const json = JSON.stringify(response); - assert.equal(typeof json, "string"); - const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); - assert.deepEqual(reconstructed, response); - }); - - it("should be round trip json serialisable with an event replay", function() { - const response = new lib.SubscriptionResponse(new RegisteredEvent()); - const json = JSON.stringify(response); - assert.equal(typeof json, "string"); - const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); - assert.deepEqual(reconstructed, response); - }); - - it("should be round trip json serialisable with a null event replay", function() { - const response = new lib.SubscriptionResponse(null); - const json = JSON.stringify(response); - assert.equal(typeof json, "string"); - const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); - assert.deepEqual(reconstructed, response); - }); - - it("should be throw when given an unregistered event in a json", function() { - assert.throws( - () => lib.SubscriptionResponse.fromJSON([UnregisteredEvent.name, {}]), - new Error(`Unregistered Event class ${UnregisteredEvent.name}`) - ); - }); - - it("should be throw when given an unregistered event", function() { - assert.throws( - () => new lib.SubscriptionResponse(new UnregisteredEvent()), - new Error(`Unregistered Event class ${UnregisteredEvent.name}`) - ); - }); - }); - describe("class SubscriptionRequest", function() { describe("permission()", function() { function newSubscriptionRequestMessage(request) { @@ -130,14 +81,14 @@ describe("lib/subscriptions", function() { }); it("should be round trip json serialisable", function() { - const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, ["channelOne"], 123); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, 123); const json = JSON.stringify(request); assert.equal(typeof json, "string"); const reconstructed = lib.SubscriptionRequest.fromJSON(JSON.parse(json)); assert.deepEqual(reconstructed, request); }); - it("should be throw when given an unregistered event", function() { + it("should throw when given an unregistered event", function() { assert.throws( () => new lib.SubscriptionRequest(UnregisteredEvent.name, true), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) @@ -157,37 +108,24 @@ describe("lib/subscriptions", function() { request: new lib.SubscriptionRequest(RegisteredEvent.name, true), src: addr({ controlId: 1 }), dst: addr("controller"), - }, { - id: 2, - request: new lib.SubscriptionRequest(ChannelEvent.name, true), - src: addr({ controlId: 2 }), - dst: addr("controller"), - }, { - id: 3, - request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]), - src: addr({ controlId: 3 }), - dst: addr("controller"), - }, { - id: 4, - request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]), - src: addr({ controlId: 4 }), - dst: addr("controller"), }]; + function getLink(id) { + return mockController.wsServer.controlConnections.get(id); + } function onceConnectorSend(id) { - const link = mockController.wsServer.controlConnections.get(id); - return events.once(link.connector, "send"); + return events.once(getLink(id).connector, "send"); } function assertLastEvent(connectorId, Event) { - const link = mockController.wsServer.controlConnections.get(connectorId); + const link = getLink(connectorId); const lastMessage = link.connector.sentMessages.at(-1); const name = Event.plugin ? `${Event.plugin}:${Event.name}` : Event.name; assert.equal(lastMessage.name, name); } function assertNoEvent(connectorId) { - const link = mockController.wsServer.controlConnections.get(connectorId); + const link = getLink(connectorId); assert.equal(link.connector.sentMessages.length, 0); } @@ -201,10 +139,6 @@ describe("lib/subscriptions", function() { subscriptions = new lib.SubscriptionController(mockController); }); - it("should handle the SubscriptionRequest event", function() { - assert.equal(mockController.handles.has(lib.SubscriptionRequest), true); - }); - describe("handle()", function() { it("should accept registered events", function() { subscriptions.handle(RegisteredEvent); @@ -227,9 +161,10 @@ describe("lib/subscriptions", function() { describe("broadcast()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); - subscriptions.handle(ChannelEvent); for (let connectorData of connectorSetupDate) { - subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); + subscriptions.handleRequest( + getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst + ); } }); @@ -245,74 +180,63 @@ describe("lib/subscriptions", function() { new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`) ); }); - it("should notify all links who subscribed to all channels", async function() { + it("should notify all links who subscribed", async function() { subscriptions.broadcast(new RegisteredEvent()); await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]); assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All - assertNoEvent(2); // ChannelEvent: All - assertNoEvent(3); // ChannelEvent: channelOne - assertNoEvent(4); // ChannelEvent: channelTwo }); - it("should not notify a link who unsubscribed from all all channels", async function() { + it("should not notify a link who unsubscribed", async function() { const connectorData = connectorSetupDate[0]; const request = new lib.SubscriptionRequest(RegisteredEvent.name, false); - await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); - mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request + await subscriptions.handleRequest( + getLink(connectorData.id), request, connectorData.src, connectorData.dst, + ); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(1); assertNoEvent(0); assertLastEvent(1, RegisteredEvent); - assertNoEvent(2); - assertNoEvent(3); - assertNoEvent(4); - }); - it("should notify all links who subscribed the specific channel", async function() { - subscriptions.broadcast(new ChannelEvent()); - await Promise.all([onceConnectorSend(2), onceConnectorSend(3)]); - assertNoEvent(0); - assertNoEvent(1); - assertLastEvent(2, ChannelEvent); - assertLastEvent(3, ChannelEvent); - assertNoEvent(4); - }); - it("should not notify a link who unsubscribed from the specific channel", async function() { - const connectorData = connectorSetupDate[3]; - const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []); - await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); - mockController.wsServer.controlConnections.get(3).connector.sentMessages.pop(); // Response to request - subscriptions.broadcast(new ChannelEvent()); - await onceConnectorSend(2); - assertNoEvent(0); - assertNoEvent(1); - assertLastEvent(2, ChannelEvent); - assertNoEvent(3); - assertNoEvent(4); }); it("should not notify links which are closed or closing", async function() { - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); link.connector.closing = true; subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(1); assertNoEvent(0); assertLastEvent(1, RegisteredEvent); - assertNoEvent(2); - assertNoEvent(3); - assertNoEvent(4); }); }); - describe("_handleRequest()", function() { + describe("unsubscribe()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + for (let connectorData of connectorSetupDate) { + subscriptions.handleRequest( + getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst + ); + } + }); + + it("should remove an active subscription", async function() { + const link = getLink(0); + subscriptions.unsubscribe(link); + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(1); + assertNoEvent(0); + assertLastEvent(1, RegisteredEvent); + }); + }); + + describe("handleRequest()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); - subscriptions.handle(ChannelEvent); }); it("should not accept subscriptions to unregistered events", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = UnregisteredEvent.name; assert.rejects( - subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), + subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) ); }); @@ -320,83 +244,79 @@ describe("lib/subscriptions", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = StringPermissionEvent.name; assert.rejects( - subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), + subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")), new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`) ); }); - it("should accept a subscription to all channels", async function() { + it("should accept a subscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); }); - it("should accept a unsubscription from all channels", async function() { + it("should accept an unsubscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); link.connector.sentMessages.pop(); const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false); - await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); assertNoEvent(0); }); - it("should accept a subscription to a specific channel", async function() { - const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); - subscriptions.broadcast(new ChannelEvent()); - await onceConnectorSend(0); - assertLastEvent(0, ChannelEvent); - }); - it("should accept a unsubscription from a specific channel", async function() { - const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); - subscriptions.broadcast(new ChannelEvent()); - await onceConnectorSend(0); - assertLastEvent(0, ChannelEvent); - - const link = mockController.wsServer.controlConnections.get(0); - link.connector.sentMessages.pop(); - - const unsubRequest = new lib.SubscriptionRequest(ChannelEvent.name, false, []); - await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); - subscriptions.broadcast(new ChannelEvent()); - assertNoEvent(0); - }); - it("should accept a respond with an event replay when returned by the handler", async function() { - subscriptions.handle(StringPermissionEvent, undefined, async function() { + it("should accept a response with an event replay when returned by the handler", async function() { + subscriptions.handle(StringPermissionEvent, async function() { return new StringPermissionEvent(); }); const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); - subscriptions.broadcast(new StringPermissionEvent()); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); await onceConnectorSend(0); assertLastEvent(0, StringPermissionEvent); - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); const lastMessage = link.connector.sentMessages.pop(); const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent); const response = eventEntry.eventFromJSON(lastMessage.data); assert.deepEqual(response, new StringPermissionEvent()); }); + it("should accept a response with null when returned by the handler", async function() { + subscriptions.handle(StringPermissionEvent, async function() { + return null; + }); + + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); + await new Promise(resolve => setImmediate(resolve)); + assertNoEvent(0); + }); }); }); describe("class EventSubscriber", function() { - let channelEvent, mockControl, registeredEvent; + let mockControl, registeredEvent; beforeEach(function() { mockControl = new MockControl(new MockConnector( addr({ controlId: 0 }), addr("controller"), )); - registeredEvent = new lib.EventSubscriber(RegisteredEvent, undefined, mockControl); - channelEvent = new lib.EventSubscriber(ChannelEvent, undefined, mockControl); + registeredEvent = new lib.EventSubscriber(RegisteredEvent, mockControl); }); function assertLastRequest(request) { @@ -407,178 +327,135 @@ describe("lib/subscriptions", function() { assert.deepEqual(lastMessage.data, request); } + function assertNoMessageSent() { + assert.equal(mockControl.connector.sentMessages.length, 0, "messages were sent"); + } + describe("constructor()", function() { it("should not accept unregistered events", function() { assert.throws( - () => new lib.EventSubscriber(UnregisteredEvent), + () => new lib.EventSubscriber(UnregisteredEvent, mockControl), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) ); }); - it("should handle the provided event, if a control link is provided", function() { - assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); - assert.equal(mockControl._eventHandlers.has(ChannelEvent), true); - }); - it("should call and use the return of a pre-handler, if provided", function() { - let calledWith = null; - function prehandler() { return "PreHandlerReturn"; } - const eventSubscriber = new lib.EventSubscriber(StringPermissionEvent, prehandler, mockControl); - eventSubscriber.subscribe(function(value) { calledWith = value; }); - eventSubscriber._handle(new StringPermissionEvent()); - assert.equal(calledWith, "PreHandlerReturn"); - }); - }); - - describe("connectControl()", function() { it("should handle the provided event", function() { - const control = new MockControl(new MockConnector(addr({ controlId: 0 }), addr("controller"))); - registeredEvent.connectControl(control); - assert.equal(control._eventHandlers.has(RegisteredEvent), true); - }); - it("should do nothing if the control is already connected", function() { - assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); - registeredEvent.connectControl(mockControl); assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); }); }); describe("subscribe()", function() { - it("should allow subscriptions to an event", function() { + it("should allow subscriptions to an event", async function() { let calledWith = null; let calledWithTwo = null; - const event = new RegisteredEvent(); - registeredEvent.subscribe(function(value) { calledWith = value; }); - registeredEvent.subscribe(function(value) { calledWithTwo = value; }); - registeredEvent._handle(event); - assert.deepEqual(calledWith, event); - assert.deepEqual(calledWithTwo, event); + const item = { id: 1, updatedAt: 1, isDeleted: false }; + const event = new RegisteredEvent([item]); + registeredEvent.subscribe(function(updates, synced) { calledWith = [updates, synced]; }); + registeredEvent.subscribe(function(updates, synced) { calledWithTwo = [updates, synced]; }); + await registeredEvent._handle(event); + assert.deepEqual(calledWith, [[item], false]); + assert.deepEqual(calledWithTwo, [[item], false]); }); - }); - - describe("subscribeToChannel()", function() { - it("should allow subscriptions to a channel for an event", function() { - let calledWith = null; - let calledWithTwo = null; - const event = new ChannelEvent(); - channelEvent.subscribeToChannel("channelOne", function(value) { calledWith = value; }); - channelEvent.subscribeToChannel("channelTwo", function(value) { calledWithTwo = value; }); - channelEvent._handle(event); - assert.deepEqual(calledWith, event); - assert.deepEqual(calledWithTwo, null); - }); - }); - - describe("unsubscribe()", function() { - it("should allow unsubscribing from an event", function() { - let calledWith = null; + it("should allow unsubscribing from an event", async function() { + let called = false; let event = new RegisteredEvent(); - function callback(value) { calledWith = value; } - registeredEvent.subscribe(callback); - registeredEvent._handle(event); - assert.deepEqual(calledWith, event); + function callback() { called = true; } + const unsubscribe = registeredEvent.subscribe(callback); + await registeredEvent._handle(event); + assert.deepEqual(called, true); - calledWith = null; + called = false; event = new RegisteredEvent(); - registeredEvent.unsubscribe(callback); - registeredEvent._handle(event); - assert.equal(calledWith, null); + unsubscribe(); + // Wait for unsubscribe to happen. + await new Promise(resolve => setImmediate(resolve)); + await registeredEvent._handle(event); + assert.equal(called, false); }); - it("should throw an error if the handler was not subscribed", function() { - function callback(value) { calledWith = value; } - assert.rejects( - () => registeredEvent.unsubscribe(callback), - new Error("handler is not registered") - ); + it("should not throw error on incorrectly unsubscribing multiple times", async function() { + const unsubscribe = registeredEvent.subscribe(() => {}); + unsubscribe(); + unsubscribe(); + // Wait for unsubscribe to happen. + await new Promise(resolve => setImmediate(resolve)); }); }); - describe("unsubscribeFromChannel()", function() { - it("should allow unsubscribing from a channel for an event", function() { - let calledWith = null; - let event = new ChannelEvent(); - function callback(value) { calledWith = value; } - channelEvent.subscribeToChannel("channelOne", callback); - channelEvent._handle(event); - assert.deepEqual(calledWith, event); - - calledWith = null; - event = new ChannelEvent(); - channelEvent.unsubscribeFromChannel("channelOne", callback); - channelEvent._handle(event); - assert.equal(calledWith, null); - - let calledWithTwo = null; - function callbackTwo(value) { calledWithTwo = value; } - channelEvent.subscribeToChannel("channelOne", callback); - channelEvent.subscribeToChannel("channelOne", callbackTwo); - channelEvent._handle(event); - assert.equal(calledWith, event); - assert.equal(calledWithTwo, event); - - calledWith = null; - calledWithTwo = null; - event = new ChannelEvent(); - channelEvent.unsubscribeFromChannel("channelOne", callback); - channelEvent._handle(event); - assert.equal(calledWith, null); - assert.equal(calledWithTwo, event); - }); - it("should throw an error if the handler was not subscribed", function() { - function callback() {} - assert.rejects( - () => channelEvent.unsubscribeFromChannel("channelOne", callback), - new Error("handler is not registered") - ); - - let calledWith = null; - function callbackTwo(value) { calledWith = value; } - channelEvent.subscribeToChannel("channelOne", callbackTwo); - assert.rejects( - () => channelEvent.unsubscribeFromChannel("channelOne", callback), - new Error("handler is not registered") - ); - const event = new ChannelEvent(); - channelEvent._handle(event); - assert.equal(calledWith, event); + describe("getSnapshot()", function() { + it("should give an immutable copy of the tracked values", async function() { + const update1 = [{ id: 1, updatedAt: 1, isDeleted: false }]; + const update2 = [{ id: 1, updatedAt: 2, isDeleted: false }, { id: 2, updatedAt: 3, isDeleted: false }]; + const update3 = [{ id: 1, updatedAt: 4, isDeleted: true }]; + const snap0 = registeredEvent.getSnapshot(); + assert.deepEqual(snap0, [new Map(), false]); + await registeredEvent._handle(new RegisteredEvent(update1)); + const snap1 = registeredEvent.getSnapshot(); + assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]); + await registeredEvent._handle(new RegisteredEvent(update2)); + const snap2 = registeredEvent.getSnapshot(); + assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]); + await registeredEvent._handle(new RegisteredEvent(update3)); + const snap3 = registeredEvent.getSnapshot(); + assert.deepEqual(snap3, [new Map([[2, update2[1]]]), false]); + // validate previous snapshots were not modified + assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]); + assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]); + assert.deepEqual(snap0, [new Map(), false]); }); }); describe("_updateSubscription()", function() { - it("should correctly request a subscription for all channels", async function() { - const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true); + it("should do nothing if not connected", async function() { + registeredEvent.control.connector.connected = false; + registeredEvent.subscribe(() => true); + await new Promise(resolve => setImmediate(resolve)); + assertNoMessageSent(); + }); + it("should correctly request a subscription", async function() { + const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1); registeredEvent.subscribe(() => true); await events.once(mockControl.connector, "send"); assertLastRequest(expected); }); - it("should correctly request a subscription for some channels", async function() { - const expected = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); - channelEvent.subscribeToChannel("channelOne", () => true); - await events.once(mockControl.connector, "send"); - assertLastRequest(expected); - }); - it("should correctly request a subscription for no channels", async function() { - const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false); + it("should correctly request no subscriptions", async function() { + const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false, -1); function callback() {} - registeredEvent.subscribe(callback); + const unsubscribe = registeredEvent.subscribe(callback); await events.once(mockControl.connector, "send"); - registeredEvent.unsubscribe(callback); + unsubscribe(); await events.once(mockControl.connector, "send"); assertLastRequest(expected); }); - it("should call handlers when a replay event is returned", async function() { - let calledWith = null; - const event = new RegisteredEvent(); - const request = registeredEvent.subscribe(function(value) { calledWith = value; }); + it("should set synced and invoke callbacks when request is responded to", async function() { + let calledWith; + registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; }); + const [msg] = await events.once(mockControl.connector, "send"); + assert(!registeredEvent.synced, "synced was set before response"); + mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src)); + await new Promise(resolve => setImmediate(resolve)); + assert(calledWith !== undefined, "callback not called"); + assert.deepEqual(calledWith, [[], true]); + assert(registeredEvent.synced, "synced was not set"); + }); + it("should be called on connector connect event", async function() { + const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1); + registeredEvent._callbacks.push(() => {}); + registeredEvent.control.connector.emit("connect"); await events.once(mockControl.connector, "send"); + assertLastRequest(expected); + }); + it("should revert synced and invoke callbacks on connector close event", async function() { + let calledWith; + registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; }); + const [msg] = await events.once(mockControl.connector, "send"); + mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src)); + await new Promise(resolve => setImmediate(resolve)); - mockControl.connector.emit("message", new lib.MessageResponse( - 0, - mockControl.connector.dst, - new lib.Address(lib.Address.control, 0, mockControl._nextRequestId - 1), - new lib.SubscriptionResponse(event).toJSON() - )); - - await request; - assert.deepEqual(calledWith, event); + assert(registeredEvent.synced, "synced was not set before close"); + calledWith = undefined; + registeredEvent.control.connector.emit("close"); + assert(calledWith !== undefined, "callback not called"); + assert.deepEqual(calledWith, [[], false]); + assert(!registeredEvent.synced, "synced was not reset"); }); }); });