feat(plugin-gateway): add RegisterAsDispatchListener

.changeset/broker-mark-forwarded-dispatches.md

@@ -0,0 +1,5 @@
+---
+"@wolfstar/plugin-broker": minor
+---
+
+`forwardGatewayDispatches` marks a `GatewayClient` as forwarding (`forwardsDispatches`) until its stop function is called, through the new optional `GatewayDispatchEmitterLike#markDispatchesForwarded`, so listeners registered with `RegisterAsDispatchListener` from `@wolfstar/plugin-gateway` run on the workers only. Emitters without it forward as before.

.changeset/gateway-register-as-dispatch-listener.md

@@ -0,0 +1,5 @@
+---
+"@wolfstar/plugin-gateway": minor
+---
+
+Add `RegisterAsDispatchListener`, a `RegisterAsGatewayListener` whose listener runs on one process only: the workers replaying the dispatches while the gateway client forwards them, the gateway client otherwise, so a bot loading the same pieces on both no longer handles each dispatch twice. `GatewayClient` gains `forwardsDispatches` and `markDispatchesForwarded()` for `forwardGatewayDispatches` to set, and `DispatchEventName`, `GatewayOnlyEventName` and `gatewayOnlyEvents` tell the events a worker receives from the ones only the gateway client emits, which the decorator rejects.

packages/plugin-broker/README.md

@@ -232,6 +232,23 @@ entries and run them concurrently, so two events of the same guild can be replay
 across workers, and the relations a listener resolves from the cache can differ from the gateway
 process's. The previous state shipped with each entry is a snapshot, so `old` stays right.
 
+A bot that loads the same pieces on the gateway process and on the workers would handle each dispatch
+twice. `forwardGatewayDispatches` marks a `GatewayClient` as forwarding (`forwardsDispatches`) until
+its stop function is called, and `RegisterAsDispatchListener` from
+[`@wolfstar/plugin-gateway`](https://www.npmjs.com/package/@wolfstar/plugin-gateway) uses it to run
+a listener on the workers only, and on the gateway client when nothing is forwarded, with no
+environment variable to read:
+
+```ts
+@RegisterAsDispatchListener("messageCreate")
+export class ModerationListener extends EventGatewayListener<"messageCreate"> {}
+```
+
+It rejects the events a worker never receives (`clientReady`, `shardReady`, …), and a `guildDelete`
+emitted by `READY` is not forwarded
+([#243](https://github.com/wolfstar-project/plugins/issues/243)), so such a listener on the workers
+misses it.
+
 ### `BrokerListener` piece
 
 Like `EventGatewayListener` in `@wolfstar/plugin-gateway`, a `BrokerListener` piece typed by event

packages/plugin-broker/src/lib/gateway.ts

@@ -37,6 +37,12 @@ export interface GatewayDispatchEmitterLike {
    * {@link GatewayReplayTargetLike.reviveDispatchState}. An emitter without it forwards dispatches without state.
    */
   serializeDispatchState?(type: string, state: unknown): unknown;
+  /**
+   * Marks the emitter as forwarding its dispatches, until the returned function is called, so that listeners meant to
+   * run on one process only (`RegisterAsDispatchListener`) stay idle on it. An emitter without it forwards without
+   * telling.
+   */
+  markDispatchesForwarded?(): () => void;
   on(event: "dispatch", listener: GatewayDispatchListener): unknown;
   off(event: "dispatch", listener: GatewayDispatchListener): unknown;
 }
@@ -63,7 +69,8 @@ export interface ForwardGatewayDispatchesOptions {
  * sharing that cache (a `@wolfstar/plugin-cache` Redis cache) see the state the dispatch left when they receive it.
  * Publishing does not hold up the gateway, and the entries are added in the order the dispatches were processed.
  * For a `GatewayClient` the entry also carries the dispatch's previous state and shard, which
- * `replayGatewayDispatches` uses on workers.
+ * `replayGatewayDispatches` uses on workers, and the client is marked as forwarding until forwarding stops, so the
+ * listeners registered with `RegisterAsDispatchListener` leave the dispatches to the workers.
  *
  * @example
  * ```typescript
@@ -125,7 +132,11 @@ export function forwardGatewayDispatches(
   };
 
   client.on("dispatch", listener);
-  return () => void client.off("dispatch", listener);
+  const unmark = client.markDispatchesForwarded?.();
+  return () => {
+    client.off("dispatch", listener);
+    unmark?.();
+  };
 }
 
 /**

packages/plugin-broker/tests/gateway-replay.test.ts

@@ -1,5 +1,11 @@
 import { createInMemoryCache } from "@wolfstar/plugin-cache";
-import { GatewayClient } from "@wolfstar/plugin-gateway";
+import { container } from "@wolfstar/http-framework";
+import {
+  EventGatewayListener,
+  GatewayClient,
+  RegisterAsDispatchListener,
+  type Message,
+} from "@wolfstar/plugin-gateway";
 import { afterEach, describe, expect, test, vi } from "vitest";
 import {
   BrokerConsumer,
@@ -299,4 +305,46 @@ describe("replayGatewayDispatches", () => {
 
     expect(replayDispatch).not.toHaveBeenCalled();
   });
+
+  test("GIVEN a gateway process and a worker loading the same dispatch listener THEN only the worker runs it", async () => {
+    const { broker, consumer } = setup();
+    const received: string[] = [];
+
+    @RegisterAsDispatchListener("messageCreate")
+    class LogMessagesListener extends EventGatewayListener<"messageCreate"> {
+      public override run(_message: Message) {
+        received.push(this.name);
+      }
+    }
+    const store = container.stores.get("listeners");
+    const context = (name: string): EventGatewayListener.LoaderContext => ({
+      name,
+      path: `/virtual/${name}.js`,
+      root: "/virtual",
+      store,
+    });
+
+    const { producer, worker } = clients();
+    try {
+      // A listener binds to `container.client`, so point it at each client in turn.
+      container.client = producer;
+      await store.insert(new LogMessagesListener(context("gateway"), {} as never));
+      container.client = worker;
+      await store.insert(new LogMessagesListener(context("worker"), {} as never));
+
+      const stopForwarding = forwardGatewayDispatches(producer, broker);
+      replayGatewayDispatches(consumer, worker);
+      await consumer.start();
+
+      await feed(producer, "MESSAGE_CREATE", message());
+      await vi.waitFor(() => expect(received).toEqual(["worker"]));
+
+      // Once forwarding stops, the gateway process handles its dispatches itself again.
+      stopForwarding();
+      await feed(producer, "MESSAGE_CREATE", message({ id: "1200000000000000001" }));
+      expect(received).toEqual(["worker", "gateway"]);
+    } finally {
+      await store.unloadAll();
+    }
+  });
 });

packages/plugin-broker/tests/gateway.test.ts

@@ -86,6 +86,19 @@ describe("forwardGatewayDispatches", () => {
     expect(client.listenerCount("dispatch")).toBe(0);
   });
 
+  test("GIVEN an emitter that can be marked THEN it is marked until forwarding stops", () => {
+    const unmark = vi.fn();
+    const markDispatchesForwarded = vi.fn(() => unmark);
+    const client = Object.assign(gateway(), { markDispatchesForwarded });
+
+    const stop = forwardGatewayDispatches(client, { publish: vi.fn() });
+    expect(markDispatchesForwarded).toHaveBeenCalledOnce();
+    expect(unmark).not.toHaveBeenCalled();
+
+    stop();
+    expect(unmark).toHaveBeenCalledOnce();
+  });
+
   test("GIVEN an emitter that serializes state THEN the serialized state is published with the dispatch", () => {
     const publish = vi.fn<Broker["publish"]>().mockResolvedValue("1-0");
     const serializeDispatchState = vi.fn((_type: string, state: unknown) => ({ raw: state }));

packages/plugin-gateway/README.md

@@ -283,6 +283,39 @@ export class LogMessagesListener extends EventGatewayListener<"messageCreate"> {
 
 With `once: true`, the listener unloads itself after its first run.
 
+With [`@wolfstar/plugin-broker`](https://www.npmjs.com/package/@wolfstar/plugin-broker), a
+dispatch is handled by the gateway process and, replayed, by every worker, so pieces loaded on both
+would run twice. `RegisterAsDispatchListener` takes the same arguments and runs the listener on
+one process only: the workers while the gateway client forwards its dispatches
+(`forwardGatewayDispatches` marks it as `client.forwardsDispatches`), the gateway client itself
+otherwise, so it also works on a bot that does not use the broker:
+
+```ts
+import {
+  EventGatewayListener,
+  RegisterAsDispatchListener,
+  type Message,
+} from "@wolfstar/plugin-gateway";
+
+@RegisterAsDispatchListener("messageCreate")
+export class LogMessagesListener extends EventGatewayListener<"messageCreate"> {
+  public override run(message: Message) {
+    console.log(`${message.author.username}: ${message.content}`);
+  }
+}
+```
+
+The events only the gateway client emits (`clientReady`, `shardReady`, `shardResume`,
+`shardClose`, `shardError`, `dispatch`, `cacheError`, `cacheSweep`) never reach a worker: the
+types reject them and the decorator throws, so keep `RegisterAsGatewayListener` for those. The
+decision is made on every emit, so the order forwarding starts and the pieces load in does not
+matter. Two limits come with it: a `guildDelete` that `READY` emits for the guilds the client no
+longer lists is not forwarded
+([#243](https://github.com/wolfstar-project/plugins/issues/243)), so a listener on the workers
+misses it, and with `events` set on `forwardGatewayDispatches`, a listener of an event built from
+a dispatch left out runs nowhere. Workers receive dispatches at least once, so keep the listener
+idempotent.
+
 Any piece (listener, command, interaction handler) reaches the client through
 `this.container.gatewayClient`, typed as `GatewayClient`, so its managers need no cast:
 

packages/plugin-gateway/src/GatewayClient.ts

@@ -458,6 +458,9 @@ export class GatewayClient extends Client {
   // guilds proceed concurrently.
   readonly #queue = new DispatchQueue();
 
+  // How many forwarders of this client's dispatches are running, see `markDispatchesForwarded`.
+  #dispatchForwarders = 0;
+
   // One cache per entity, shared by every manager of that entity.
   readonly #caches = new Map<CacheEntityName, Cache<any>>();
 
@@ -940,6 +943,31 @@ export class GatewayClient extends Client {
     return emitAndWait(this, event, args);
   }
 
+  /**
+   * Whether the dispatches of this client are forwarded to workers, which replay them on their own listeners, see
+   * `forwardGatewayDispatches` from `@wolfstar/plugin-broker`. Listeners registered with
+   * {@link RegisterAsDispatchListener} stay idle on a client that does.
+   */
+  public get forwardsDispatches(): boolean {
+    return this.#dispatchForwarders > 0;
+  }
+
+  /**
+   * Marks this client as forwarding its dispatches, until the returned function is called. Called by
+   * `forwardGatewayDispatches`, it counts, so the client keeps forwarding while any forwarder is left.
+   *
+   * @returns A function to undo the mark, which does nothing when called again.
+   */
+  public markDispatchesForwarded(): () => void {
+    this.#dispatchForwarders++;
+    let marked = true;
+    return () => {
+      if (!marked) return;
+      marked = false;
+      this.#dispatchForwarders--;
+    };
+  }
+
   /**
    * The dispatch types {@link GatewayClient.replayDispatch} turns into events.
    */

packages/plugin-gateway/src/util/decorators.ts

@@ -1,4 +1,4 @@
-import type { GatewayEventName } from "./events.js";
+import { gatewayOnlyEvents, type DispatchEventName, type GatewayEventName } from "./events.js";
 import type { EventGatewayListener } from "../EventGatewayListener.js";
 
 type ListenerConstructor = new (...args: any[]) => EventGatewayListener<any>;
@@ -40,3 +40,72 @@ export function RegisterAsGatewayListener<Event extends GatewayEventName>(
     } as unknown as T;
   };
 }
+
+/**
+ * Like {@link RegisterAsGatewayListener}, for an event that comes from a dispatch, but the listener only runs on one
+ * process: the workers replaying the dispatches while the gateway client forwards them (see `forwardGatewayDispatches`
+ * from `@wolfstar/plugin-broker`), the gateway client itself otherwise. A bot loading the same pieces on both
+ * processes therefore handles each dispatch once, without reading its environment to tell which process it is on.
+ *
+ * @remarks
+ * The decision is made on every emit, so it does not matter whether the forwarding starts before or after the
+ * listener is loaded, and stopping it brings the listener back. A `once` listener is not spent while it idles.
+ *
+ * The events only the gateway client emits (`clientReady`, `shardReady`, `shardResume`, `shardClose`, `shardError`,
+ * `dispatch`, `cacheError`, `cacheSweep`) are never replayed on a worker: they are rejected by the types, and the
+ * decorator throws for them. Use {@link RegisterAsGatewayListener} for those.
+ *
+ * Only the dispatches that are forwarded reach the workers: with `forwardGatewayDispatches`'s `events` set, a listener
+ * of an event built from another dispatch runs nowhere. A `guildDelete` that `READY` emits for the guilds it no longer
+ * lists is not forwarded either (wolfstar-project/plugins#243), so a listener on the workers misses it. Delivery to the
+ * workers is at-least-once, so the listener should be idempotent.
+ *
+ * @example
+ * ```typescript
+ * import { EventGatewayListener, RegisterAsDispatchListener, type Message } from '@wolfstar/plugin-gateway';
+ *
+ * @RegisterAsDispatchListener('messageCreate')
+ * export class LogMessagesListener extends EventGatewayListener<'messageCreate'> {
+ *   public override run(message: Message) {
+ *     console.log(`${message.author.username}: ${message.content}`);
+ *   }
+ * }
+ * ```
+ *
+ * @param event The dispatch event to listen to.
+ * @param options The additional listener options.
+ * @throws {TypeError} When the event is one only the gateway client emits.
+ */
+export function RegisterAsDispatchListener<Event extends DispatchEventName>(
+  event: Event,
+  options: Omit<EventGatewayListener.Options<Event>, "event"> = {},
+) {
+  if (gatewayOnlyEvents.has(event)) {
+    throw new TypeError(
+      `The "${event}" event is only emitted by the gateway client, and never reaches a worker: use RegisterAsGatewayListener.`,
+    );
+  }
+
+  const register = RegisterAsGatewayListener(event, options);
+  return function decorate<T extends ListenerConstructor>(target: T): T {
+    const base = register(target) as unknown as new (...args: any[]) => DispatchListenerInternals;
+    return class extends base {
+      public constructor(...args: any[]) {
+        super(...args);
+
+        // The outermost wrapper, so a `once` listener is only unloaded by a run that happens.
+        // oxlint-disable-next-line no-underscore-dangle
+        const listener = this._listener;
+        // oxlint-disable-next-line no-underscore-dangle
+        this._listener = (...listenerArgs: readonly unknown[]) =>
+          this.emitter.forwardsDispatches ? undefined : listener(...listenerArgs);
+      }
+    } as unknown as T;
+  };
+}
+
+interface DispatchListenerInternals {
+  // Anything but a `GatewayClient` has no flag, and so never forwards.
+  emitter: { forwardsDispatches?: boolean };
+  _listener: (...args: readonly unknown[]) => unknown;
+}

packages/plugin-gateway/src/util/events.ts

@@ -398,6 +398,39 @@ export interface GatewayEventMap {
  */
 export type GatewayEventName = keyof GatewayEventMap;
 
+/**
+ * The events only the process that connects to Discord emits: a worker replaying forwarded dispatches never sees
+ * them. Every other event comes from a dispatch, `raw` included, which {@link GatewayClient.replayDispatch} emits too.
+ */
+export type GatewayOnlyEventName =
+  | "dispatch"
+  | "shardReady"
+  | "clientReady"
+  | "shardResume"
+  | "shardClose"
+  | "shardError"
+  | "cacheError"
+  | "cacheSweep";
+
+/**
+ * The name of any event that comes from a dispatch, and so reaches a worker replaying the forwarded ones.
+ */
+export type DispatchEventName = Exclude<GatewayEventName, GatewayOnlyEventName>;
+
+/**
+ * The {@link GatewayOnlyEventName} events, for telling them from {@link DispatchEventName} ones at runtime.
+ */
+export const gatewayOnlyEvents: ReadonlySet<string> = new Set<GatewayOnlyEventName>([
+  "dispatch",
+  "shardReady",
+  "clientReady",
+  "shardResume",
+  "shardClose",
+  "shardError",
+  "cacheError",
+  "cacheSweep",
+]);
+
 /**
  * The name of every event a {@link GatewayClient} emits, mirroring the keys of {@link GatewayEventMap}, like
  * `@wolfstar/http-framework`'s own `Events`.