Class RedisZodSubscriber

apps/webapp/app/v3/utils/zodPubSub.server.ts:23–97

}

class RedisZodSubscriber<TMessageCatalog extends ZodMessageCatalogSchema>
  implements ZodSubscriber<TMessageCatalog>
{
  private _subscriber: Redis;
  private _listeners: Map<string, (payload: unknown) => Promise<void>> = new Map();
  private _messageHandler: ZodMessageHandler<TMessageCatalog>;

  public onUnsubscribed: Evt<{
    pattern: string;
  }> = new Evt();

  constructor(
    private readonly _pattern: string,
    private readonly _options: ZodPubSubOptions<TMessageCatalog>,
    private readonly _logger: Logger
  ) {
    this._subscriber = new Redis(_options.redis);
    this._messageHandler = new ZodMessageHandler({
      schema: _options.schema,
    });
  }

  async initialize() {
    await this._subscriber.psubscribe(this._pattern);
    this._subscriber.on("pmessage", this.#onMessage.bind(this));
  }

  public on<K extends keyof TMessageCatalog>(
    eventName: K,
    listener: (payload: z.infer<TMessageCatalog[K]>) => Promise<void>
  ): void {
    this._listeners.set(eventName as string, listener);
  }

  public async stopListening(): Promise<void> {
    this._listeners.clear();
    await this._subscriber.punsubscribe();

    this.onUnsubscribed.post({ pattern: this._pattern });

    this._subscriber.quit();
  }

  async #onMessage(pattern: string, channel: string, serializedMessage: string) {
    if (pattern !== this._pattern) {
      return;
    }

    const parsedMessage = safeJsonParse(serializedMessage);

    if (!parsedMessage) {
      return;
    }

    const message = this._messageHandler.parseMessage(parsedMessage);

    if (typeof message.type !== "string") {
      return;