diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts new file mode 100644 index 00000000..a8be14b6 --- /dev/null +++ b/packages/lib/src/subscriptions.ts @@ -0,0 +1,273 @@ +import { Type, Static } from "@sinclair/typebox"; +import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector } from "./link"; +import { Address, MessageRequest } from "./data"; +import { User } from "./users"; + +export type SubscriptionChannelCategoriser = (event: Event) => string | number; +export type SubscriptionRequestHandler = RequestHandler | null>; +export type EventSubscriberCallback = (event: Event) => Promise; + +/** + * Response received by the subscriber after a request + * It can contain an eventReplay value if the event sender implements a subscription handler + */ +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]; + } else { + 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(); + } + } +} + +/** + * A subscription request sent by a subscriber, this updates what events the subscriber will be sent + * allChannels: false, channels: [] will unsubscribe the subscriber from being sent any events + */ +export class SubscriptionRequest { + declare ["constructor"]: typeof SubscriptionRequest; + static type = "request" as const; + static src = ["control", "instance"] as const; + static dst = "controller" as const; + static plugin = "exp_commands" as const; + static Response = SubscriptionResponse; + static permission(user: User, message: MessageRequest) { + if (typeof message.data === "object" && message.data !== null) { + const data = message.data as Static; + const entry = Link._eventsByName.get(data[0]); + if (entry && entry.Event.permission) { + if (typeof entry.Event.permission === "string") { + user.checkPermission(entry.Event.permission); + } else { + entry.Event.permission(user, message); + } + } + } + } + + constructor( + public eventName: string, + public allChannels: boolean, + public channels: Array = [], + public lastRequestTime: number = 0, + ) { + } + + 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]; + } + + static fromJSON(json: Static): SubscriptionRequest { + return new this(...json); + } +} + +/** + * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers + * After creation, no other handler can be registered for SubscriptionRequest + */ +export class SubscriptionController { + _events = new Map, + channelCategoriser?: SubscriptionChannelCategoriser, + subscriptions: Map }>, + }>(); + + constructor( + private controller: any, // Controller | Link + ) { + this.controller.handle(SubscriptionRequest, this._handleEvent.bind(this)); + } + + handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser): void; + handle( + Event: EventClass, + subscriptionUpdate?: SubscriptionRequestHandler, + channelCategoriser?: SubscriptionChannelCategoriser, + ) { + const entry = Link._eventsByClass.get(Event); + if (!entry) { + throw new Error(`Unregistered Event class ${Event.name}`); + } + if (this._events.has(entry.name)) { + throw new Error(`Event ${entry.name} is already registered`); + } + this._events.set(entry.name, { + subscriptionUpdate: subscriptionUpdate, + channelCategoriser: channelCategoriser, + subscriptions: new Map(), + }); + } + + broadcast(event: Event): void; + broadcast(event: Event) { + const entry = Link._eventsByClass.get(event.constructor); + if (!entry) { + throw new Error(`Unregistered Event class ${Event.name}`); + } + const eventData = this._events.get(entry.name); + if (!eventData) { + throw new Error(`Event ${entry.name} is not a registered as subscribable`); + } + const channel = eventData.channelCategoriser ? eventData.channelCategoriser(event) : null; + for (let [link, subscription] of eventData.subscriptions.entries()) { + if (subscription.all || (channel && subscription.channels.includes(channel))) { + link.send(event); + } + } + } + + async _handleEvent(event: SubscriptionRequest, src: Address, dst: Address) { + if (!Link._eventsByName.has(event.eventName)) { + throw new Error(`Event ${event.eventName} is not a registered event`); + } + const eventData = this._events.get(event.eventName); + 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) { + eventData.subscriptions.delete(link); + return new SubscriptionResponse(eventReplay); + } else { + eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); + return new SubscriptionResponse(eventReplay); + } + } +} + +/** + * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers + * After creation, no other handler can be registered for SubscriptionRequest + */ +export class EventSubscriber { + _callbacks = new Array>() + _channelCallbacks = new Map>>() + lastResponse?: Event = undefined; + lastResponseTime = 0; + + constructor( + private event: EventClass, + private channelCategoriser?: SubscriptionChannelCategoriser, + private control?: Link + ) { + const entry = Link._eventsByClass.get(this.event); + if (!entry) { + throw new Error(`Unregistered Event class ${this.event.name}`); + } + if (this.control) { + this.control.handle(this.event, this._handle.bind(this)); + } + } + + async _handle(response: Event) { + this.lastResponse = response; + this.lastResponseTime = Date.now(); + for (let callback of this._callbacks) { + callback(response); + } + } + + connectControl(control: Link) { + if (this.control === control) return; + this.control = control; + this.control.handle(this.event, this._handle.bind(this)); + } + + subscribeAll(handler: EventSubscriberCallback) { + this._callbacks.push(handler); + this._updateSubscription(); + } + + subscribe(channel: string | number, handler: EventSubscriberCallback) { + if (!this._channelCallbacks.get(channel)) { + this._channelCallbacks.set(channel, []); + } + this._channelCallbacks.get(channel)!.push(handler); + this._updateSubscription(); + } + + unsubscribeAll(handler: EventSubscriberCallback) { + let index = this._callbacks.lastIndexOf(handler); + if (index === -1) { + throw new Error("handler is not registered"); + } + + this._callbacks.splice(index, 1); + this._updateSubscription(); + } + + unsubscribe(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); + } + this._updateSubscription(); + } + + async _updateSubscription() { + if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return; + 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) { + this._handle(response.eventReplay); + } + } +} \ No newline at end of file