realtimechannel.ts

import { actions, channelModes } from '../types/protocolmessagecommon';
import ProtocolMessage, { fromValues as protocolMessageFromValues } from '../types/protocolmessage';
import EventEmitter from '../util/eventemitter';
import * as Utils from '../util/utils';
import Logger from '../util/logger';
import { EncodingDecodingContext, CipherOptions, populateFieldsFromParent } from '../types/basemessage';
import Message, { getMessagesSize, encodeArray as encodeMessagesArray } from '../types/message';
import ChannelStateChange from './channelstatechange';
import ErrorInfo, { PartialErrorInfo } from '../types/errorinfo';
import * as API from '../../../../ably';
import ConnectionManager from '../transport/connectionmanager';
import { StandardCallback } from '../../types/utils';
import BaseRealtime from './baserealtime';
import { ChannelOptions } from '../../types/channel';
import { normaliseChannelOptions } from '../util/defaults';
import { PaginatedResult } from './paginatedresource';
import type { PushChannel } from 'plugins/push';
import type { WirePresenceMessage } from '../types/presencemessage';
import type { RealtimeObject, WireObjectMessage } from 'plugins/liveobjects';
import type RealtimePresence from './realtimepresence';
import type RealtimeAnnotations from './realtimeannotations';

interface RealtimeHistoryParams {
  start?: number;
  end?: number;
  direction?: string;
  limit?: number;
  untilAttach?: boolean;
  from_serial?: string;
}

function validateChannelOptions(options?: API.ChannelOptions) {
  if (options && 'params' in options && !Utils.isObject(options.params)) {
    return new ErrorInfo('options.params must be an object', 40000, 400);
  }
  if (options && 'modes' in options) {
    if (!Array.isArray(options.modes)) {
      return new ErrorInfo('options.modes must be an array', 40000, 400);
    }
    for (let i = 0; i < options.modes.length; i++) {
      const currentMode = options.modes[i];
      if (
        !currentMode ||
        typeof currentMode !== 'string' ||
        !channelModes.includes(String.prototype.toUpperCase.call(currentMode))
      ) {
        return new ErrorInfo('Invalid channel mode: ' + currentMode, 40000, 400);
      }
    }
  }
}

class RealtimeChannel extends EventEmitter {
  name: string;
  channelOptions: ChannelOptions;
  client: BaseRealtime;
  private _presence: RealtimePresence | null;
  private _annotations: RealtimeAnnotations | null = null;
  get presence(): RealtimePresence {
    if (!this._presence) {
      Utils.throwMissingPluginError('RealtimePresence');
    }
    return this._presence;
  }
  get annotations(): RealtimeAnnotations {
    if (!this._annotations) {
      Utils.throwMissingPluginError('Annotations');
    }
    return this._annotations;
  }
  connectionManager: ConnectionManager;
  state: API.ChannelState;
  subscriptions: EventEmitter;
  filteredSubscriptions?: Map<API.messageCallback<Message>, Map<API.MessageFilter, API.messageCallback<Message>[]>>;
  syncChannelSerial?: string | null;
  properties: {
    attachSerial: string | null | undefined;
    channelSerial: string | null | undefined;
  };
  errorReason: ErrorInfo | string | null;
  _mode = 0;
  _attachResume: boolean;
  _decodingContext: EncodingDecodingContext;
  _lastPayload: {
    messageId?: string | null;
    protocolMessageChannelSerial?: string | null;
    decodeFailureRecoveryInProgress: null | boolean;
  };
  /**
   * Emits an 'attached' event (with no payload) whenever an ATTACHED protocol
   * message is received from the server. Used by setOptions (RTL16a) to know
   * when the server has confirmed new channel options.
   */
  _attachedReceived: EventEmitter;
  /**
   * Internal event emitter for channel state changes, not affected by public
   * off() calls. Exists to satisfy RTC10: the client library should never
   * register internal listeners with the public EventEmitter in such a way
   * that a user calling off() would result in the library not working as
   * expected.
   */
  internalStateChanges: EventEmitter;
  params?: Record<string, any>;
  modes: API.ChannelMode[] | undefined;
  stateTimer?: number | NodeJS.Timeout | null;
  retryTimer?: number | NodeJS.Timeout | null;
  retryCount: number = 0;
  _push?: PushChannel;
  _object?: RealtimeObject;

  constructor(client: BaseRealtime, name: string, options?: API.ChannelOptions) {
    super(client.logger);
    Logger.logAction(this.logger, Logger.LOG_MINOR, 'RealtimeChannel()', 'started; name = ' + name);
    this.name = name;
    this.channelOptions = normaliseChannelOptions(client._Crypto ?? null, this.logger, options);
    this.client = client;
    this._presence = client._RealtimePresence ? new client._RealtimePresence.RealtimePresence(this) : null;
    if (client._Annotations) {
      this._annotations = new client._Annotations.RealtimeAnnotations(this);
    }
    this.connectionManager = client.connection.connectionManager;
    this.state = 'initialized';
    this.subscriptions = new EventEmitter(this.logger);
    this.syncChannelSerial = undefined;
    this.properties = {
      attachSerial: undefined,
      channelSerial: undefined,
    };
    this.setOptions(options);
    this.errorReason = null;
    this._attachResume = false;
    this._decodingContext = {
      channelOptions: this.channelOptions,
      plugins: client.options.plugins || {},
      baseEncodedPreviousPayload: undefined,
    };
    this._lastPayload = {
      messageId: null,
      protocolMessageChannelSerial: null,
      decodeFailureRecoveryInProgress: null,
    };
    this._attachedReceived = new EventEmitter(this.logger);
    this.internalStateChanges = new EventEmitter(this.logger);

    if (client.options.plugins?.Push) {
      this._push = new client.options.plugins.Push.PushChannel(this);
    }

    if (client.options.plugins?.LiveObjects) {
      this._object = new client.options.plugins.LiveObjects.RealtimeObject(this);
    }
  }

  get push() {
    if (!this._push) {
      Utils.throwMissingPluginError('Push');
    }
    return this._push;
  }

  /** @spec RTL27 */
  get object() {
    if (!this._object) {
      Utils.throwMissingPluginError('LiveObjects'); // RTL27b
    }
    return this._object; // RTL27a
  }

  // Override of EventEmitter method
  emit(event: string, ...args: unknown[]) {
    super.emit(event, ...args);
    this.internalStateChanges.emit(event, ...args);
  }

  invalidStateError(): ErrorInfo {
    return new ErrorInfo(
      'Channel operation failed as channel state is ' + this.state,
      90001,
      400,
      this.errorReason || undefined,
    );
  }

  static processListenerArgs(args: unknown[]): any[] {
    /* [event], listener */
    args = Array.prototype.slice.call(args);
    if (typeof args[0] === 'function') {
      args.unshift(null);
    }
    return args;
  }

  async setOptions(options?: API.ChannelOptions): Promise<void> {
    const previousChannelOptions = this.channelOptions;
    const err = validateChannelOptions(options);
    if (err) {
      throw err;
    }
    this.channelOptions = normaliseChannelOptions(this.client._Crypto ?? null, this.logger, options);
    if (this._decodingContext) this._decodingContext.channelOptions = this.channelOptions;
    if (this._shouldReattachToSetOptions(options, previousChannelOptions)) {
      /* This does not just do _attach(true, null, callback) because that would put us
       * into the 'attaching' state until we receive the new attached, which is
       * conceptually incorrect: we are still attached, we just have a pending request to
       * change some channel params. Per RTL17 going into the attaching state would mean
       * rejecting messages until we have confirmation that the options have changed,
       * which would unnecessarily lose message continuity. */
      this.attachImpl();
      return new Promise<void>((resolve, reject) => {
        const cleanup = () => {
          this._attachedReceived.off(onAttached);
          this.internalStateChanges.off(onFailure);
        };
        const onAttached = () => {
          cleanup();
          resolve();
        };
        const onFailure = (stateChange: ChannelStateChange) => {
          cleanup();
          reject(stateChange.reason);
        };
        this._attachedReceived.once('attached', onAttached);
        this.internalStateChanges.once(['detached', 'failed'], onFailure);
      });
    }
  }

  _shouldReattachToSetOptions(options: API.ChannelOptions | undefined, prevOptions: API.ChannelOptions) {
    if (!(this.state === 'attached' || this.state === 'attaching')) {
      return false;
    }
    if (options?.params) {
      // Don't check against the `agent` param - it isn't returned in the ATTACHED message
      const requestedParams = omitAgent(options.params);
      const existingParams = omitAgent(prevOptions.params);

      if (Object.keys(requestedParams).length !== Object.keys(existingParams).length) {
        return true;
      }

      if (!Utils.shallowEquals(existingParams, requestedParams)) {
        return true;
      }
    }
    if (options?.modes) {
      if (!prevOptions.modes || !Utils.arrEquals(options.modes, prevOptions.modes)) {
        return true;
      }
    }
    return false;
  }

  async publish(...args: any[]): Promise<API.PublishResult> {
    const first = args[0],
      second = args[1];
    let messages: Message[];
    let params: Record<string, any> | undefined;

    if (typeof first === 'string' || first === null || first === undefined) {
      messages = [Message.fromValues({ name: first, data: second })];
      params = args[2];
    } else if (Utils.isObject(first)) {
      messages = [Message.fromValues(first)];
      params = args[1];
    } else if (Array.isArray(first)) {
      messages = Message.fromValuesArray(first);
      params = args[1];
    } else {
      throw new ErrorInfo(
        'The single-argument form of publish() expects a message object or an array of message objects',
        40013,
        400,
      );
    }
    const maxMessageSize = this.client.options.maxMessageSize;
    // TODO get rid of CipherOptions type assertion, indicates channeloptions types are broken
    const wireMessages = await encodeMessagesArray(messages, this.channelOptions as CipherOptions);
    /* RSL1i */
    const size = getMessagesSize(wireMessages);
    if (size > maxMessageSize) {
      throw new ErrorInfo(
        `Maximum size of messages that can be published at once exceeded (was ${size} bytes; limit is ${maxMessageSize} bytes)`,
        40009,
        400,
      );
    }

    this.throwIfUnpublishableState();

    Logger.logAction(
      this.logger,
      Logger.LOG_MICRO,
      'RealtimeChannel.publish()',
      'sending message; channel state is ' + this.state + ', message count = ' + wireMessages.length,
    );

    const pm = protocolMessageFromValues({
      action: actions.MESSAGE,
      channel: this.name,
      messages: wireMessages,
      params: params ? Utils.stringifyValues(params) : undefined,
    });
    return this.sendAndAwaitAck(pm);
  }

  throwIfUnpublishableState(): void {
    if (!this.connectionManager.activeState()) {
      throw this.connectionManager.getError();
    }
    if (this.state === 'failed' || this.state === 'suspended') {
      throw this.invalidStateError();
    }
  }

  onEvent(messages: Array<any>): void {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.onEvent()', 'received message');
    const subscriptions = this.subscriptions;
    for (let i = 0; i < messages.length; i++) {
      const message = messages[i];
      subscriptions.emit(message.name, message);
    }
  }

  async attach(): Promise<ChannelStateChange | null> {
    if (this.state === 'attached') {
      return null;
    }

    return new Promise((resolve, reject) => {
      this._attach(false, null, (err, result) => (err ? reject(err) : resolve(result!)));
    });
  }

  _attach(
    forceReattach: boolean,
    attachReason: ErrorInfo | null,
    callback?: StandardCallback<ChannelStateChange>,
  ): void {
    if (!callback) {
      callback = (err?: ErrorInfo | null) => {
        if (err) {
          Logger.logAction(
            this.logger,
            Logger.LOG_ERROR,
            'RealtimeChannel._attach()',
            'Channel attach failed: ' + err.toString(),
          );
        }
      };
    }

    const connectionManager = this.connectionManager;
    if (!connectionManager.activeState()) {
      callback(connectionManager.getError());
      return;
    }

    if (this.state !== 'attaching' || forceReattach) {
      this.requestState('attaching', attachReason);
    }

    this.internalStateChanges.once(function (this: { event: string }, stateChange: ChannelStateChange) {
      switch (this.event) {
        case 'attached':
          callback?.(null, stateChange);
          break;
        case 'detached':
        case 'suspended':
        case 'failed':
          callback?.(
            stateChange.reason ||
              connectionManager.getError() ||
              new ErrorInfo('Unable to attach; reason unknown; state = ' + this.event, 90000, 500),
          );
          break;
        case 'detaching':
          callback?.(new ErrorInfo('Attach request superseded by a subsequent detach request', 90000, 409));
          break;
      }
    });
  }

  attachImpl(): void {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.attachImpl()', 'sending ATTACH message');
    const attachMsg = protocolMessageFromValues({
      action: actions.ATTACH,
      channel: this.name,
      params: this.channelOptions.params,
      // RTL4c1: Includes the channel serial to resume from a previous message
      // or attachment.
      channelSerial: this.properties.channelSerial,
    });
    if (this.channelOptions.modes) {
      attachMsg.encodeModesToFlags(Utils.allToUpperCase(this.channelOptions.modes) as API.ChannelMode[]);
    }
    if (this._attachResume) {
      attachMsg.setFlag('ATTACH_RESUME');
    }
    if (this._lastPayload.decodeFailureRecoveryInProgress) {
      attachMsg.channelSerial = this._lastPayload.protocolMessageChannelSerial;
    }
    this.send(attachMsg);
  }

  async detach(): Promise<void> {
    const connectionManager = this.connectionManager;
    switch (this.state) {
      // RTL5j
      case 'suspended':
        this.notifyState('detached');
        return;
      case 'detached':
        return;
      // RTL5b
      case 'failed':
        throw new ErrorInfo('Unable to detach; channel state = failed', 90001, 400);
      default:
        // RTL5l: if connection is not connected, immediately transition to detached
        if (connectionManager.state.state !== 'connected') {
          this.notifyState('detached');
          return;
        }
        this.requestState('detaching');
      // eslint-disable-next-line no-fallthrough
      case 'detaching':
        return new Promise((resolve, reject) => {
          this.internalStateChanges.once(function (this: { event: string }, stateChange: ChannelStateChange) {
            switch (this.event) {
              case 'detached':
                resolve();
                break;
              case 'attached':
              case 'suspended':
              case 'failed':
                reject(
                  stateChange.reason ||
                    connectionManager.getError() ||
                    new ErrorInfo('Unable to detach; reason unknown; state = ' + this.event, 90000, 500),
                );
                break;
              case 'attaching':
                reject(new ErrorInfo('Detach request superseded by a subsequent attach request', 90000, 409));
                break;
            }
          });
        });
    }
  }

  detachImpl(): void {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.detach()', 'sending DETACH message');
    const msg = protocolMessageFromValues({ action: actions.DETACH, channel: this.name });
    this.send(msg);
  }

  async subscribe(...args: unknown[] /* [event], listener */): Promise<ChannelStateChange | null> {
    const [event, listener] = RealtimeChannel.processListenerArgs(args);

    if (this.state === 'failed') {
      throw ErrorInfo.fromValues(this.invalidStateError());
    }

    // Filtered
    if (event && typeof event === 'object' && !Array.isArray(event)) {
      this.client._FilteredSubscriptions.subscribeFilter(this, event, listener);
    } else {
      this.subscriptions.on(event, listener);
    }

    // (RTL7g)
    if (this.channelOptions.attachOnSubscribe !== false) {
      return this.attach();
    } else {
      return null;
    }
  }

  unsubscribe(...args: unknown[] /* [event], listener */): void {
    const [event, listener] = RealtimeChannel.processListenerArgs(args);

    // If we either have a filtered listener, a filter or both we need to do additional processing to find the original function(s)
    if ((typeof event === 'object' && !listener) || this.filteredSubscriptions?.has(listener)) {
      this.client._FilteredSubscriptions
        .getAndDeleteFilteredSubscriptions(this, event, listener)
        .forEach((l) => this.subscriptions.off(l));
      return;
    }

    this.subscriptions.off(event, listener);
  }

  sync(): void {
    /* check preconditions */
    switch (this.state) {
      case 'initialized':
      case 'detaching':
      case 'detached':
        throw new PartialErrorInfo('Unable to sync to channel; not attached', 40000);
      default:
    }
    const connectionManager = this.connectionManager;
    if (!connectionManager.activeState()) {
      throw connectionManager.getError();
    }

    /* send sync request */
    const syncMessage = protocolMessageFromValues({ action: actions.SYNC, channel: this.name });
    if (this.syncChannelSerial) {
      syncMessage.channelSerial = this.syncChannelSerial;
    }
    connectionManager.send(syncMessage);
  }

  send(msg: ProtocolMessage): void {
    this.connectionManager.send(msg);
  }

  async sendAndAwaitAck(msg: ProtocolMessage): Promise<API.PublishResult> {
    return new Promise((resolve, reject) => {
      this.connectionManager.send(msg, this.client.options.queueMessages, (err, publishResponse) => {
        if (err) {
          reject(err);
        } else {
          resolve(publishResponse!);
        }
      });
    });
  }

  async sendPresence(presence: WirePresenceMessage[]): Promise<void> {
    const msg = protocolMessageFromValues({
      action: actions.PRESENCE,
      channel: this.name,
      presence: presence,
    });
    await this.sendAndAwaitAck(msg);
  }

  async sendState(objectMessages: WireObjectMessage[]): Promise<API.PublishResult> {
    const msg = protocolMessageFromValues({
      action: actions.OBJECT,
      channel: this.name,
      state: objectMessages,
    });
    return this.sendAndAwaitAck(msg);
  }

  // Access to this method is synchronised by ConnectionManager#processChannelMessage, in order to synchronise access to the state stored in _decodingContext.
  async processMessage(message: ProtocolMessage): Promise<void> {
    if (
      message.action === actions.ATTACHED ||
      message.action === actions.MESSAGE ||
      message.action === actions.PRESENCE ||
      message.action === actions.OBJECT ||
      message.action === actions.ANNOTATION
    ) {
      // RTL15b
      this.setChannelSerial(message.channelSerial);
    }

    let syncChannelSerial,
      isSync = false;
    switch (message.action) {
      case actions.ATTACHED: {
        this.properties.attachSerial = message.channelSerial;
        this._mode = message.getMode();
        this.params = (message as any).params || {};
        const modesFromFlags = message.decodeModesFromFlags();
        this.modes = (modesFromFlags && (Utils.allToLowerCase(modesFromFlags) as API.ChannelMode[])) || undefined;
        const resumed = message.hasFlag('RESUMED');
        const hasPresence = message.hasFlag('HAS_PRESENCE');
        const hasBacklog = message.hasFlag('HAS_BACKLOG');
        const hasObjects = message.hasFlag('HAS_OBJECTS');
        this._attachedReceived.emit('attached');
        if (this.state === 'attached') {
          if (!resumed) {
            // we have lost continuity.
            // the presence set needs to be re-synced
            if (this._presence) {
              this._presence.onAttached(hasPresence);
            }
            // the Objects tree needs to be re-synced
            if (this._object) {
              this._object.onAttached(hasObjects);
            }
          }
          const change = new ChannelStateChange(this.state, this.state, resumed, hasBacklog, message.error);
          if (!resumed || this.channelOptions.updateOnAttached) {
            this.emit('update', change);
          }
        } else if (this.state === 'detaching') {
          /* RTL5i: re-send DETACH and remain in the 'detaching' state */
          this.checkPendingState();
        } else {
          this.notifyState('attached', message.error, resumed, hasPresence, hasBacklog, hasObjects);
        }
        break;
      }

      case actions.DETACHED: {
        const detachErr = message.error
          ? ErrorInfo.fromValues(message.error)
          : new ErrorInfo('Channel detached', 90001, 404);
        if (this.state === 'detaching') {
          this.notifyState('detached', detachErr);
        } else if (this.state === 'attaching') {
          /* Only retry immediately if we were previously attached. If we were
           * attaching, go into suspended, fail messages, and wait a few seconds
           * before retrying */
          this.notifyState('suspended', detachErr);
        } else if (this.state === 'attached' || this.state === 'suspended') {
          // RTL13a
          this.requestState('attaching', detachErr);
        }
        // else no action (detached in initialized, detached, or failed state is a noop)
        break;
      }

      case actions.SYNC:
        /* syncs can have channelSerials, but might not if the sync is one page long */
        isSync = true;
        syncChannelSerial = this.syncChannelSerial = message.channelSerial;
        /* syncs can happen on channels with no presence data as part of connection
         * resuming, in which case protocol message has no presence property */
        if (!message.presence) break;
      // eslint-disable-next-line no-fallthrough
      case actions.PRESENCE: {
        if (!message.presence) {
          break;
        }

        populateFieldsFromParent(message);
        const options = this.channelOptions;
        if (this._presence) {
          const presenceMessages = await Promise.all(
            message.presence.map((wpm) => {
              return wpm.decode(options, this.logger);
            }),
          );

          this._presence.setPresence(presenceMessages, isSync, syncChannelSerial as any);
        }
        break;
      }

      // RTL1
      // OBJECT and OBJECT_SYNC message processing share most of the logic, so group them together
      case actions.OBJECT:
      case actions.OBJECT_SYNC: {
        if (!this._object || !message.state) {
          return;
        }

        populateFieldsFromParent(message);
        // need to use the active protocol format instead of just client's useBinaryProtocol option,
        // as comet transport does not support msgpack and will default to json without changing useBinaryProtocol.
        // message processing is done in the same event loop tick up until this point,
        // so we can reliably expect an active protocol to exist and be the one that received the object message.
        const format = this.client.connection.connectionManager.getActiveTransportFormat()!;
        const objectMessages = message.state.map((om) => om.decode(this.client, format));

        if (message.action === actions.OBJECT) {
          this._object.handleObjectMessages(objectMessages);
        } else {
          this._object.handleObjectSyncMessages(objectMessages, message.channelSerial);
        }

        break;
      }

      case actions.MESSAGE: {
        //RTL17
        if (this.state !== 'attached') {
          Logger.logAction(
            this.logger,
            Logger.LOG_MAJOR,
            'RealtimeChannel.processMessage()',
            'Message "' +
              message.id +
              '" skipped as this channel "' +
              this.name +
              '" state is not "attached" (state is "' +
              this.state +
              '").',
          );
          return;
        }

        populateFieldsFromParent(message);

        const encoded = message.messages!,
          firstMessage = encoded[0],
          lastMessage = encoded[encoded.length - 1];

        if (
          firstMessage.extras &&
          firstMessage.extras.delta &&
          firstMessage.extras.delta.from !== this._lastPayload.messageId
        ) {
          const msg =
            'Delta message decode failure - previous message not available for message "' +
            message.id +
            '" on this channel "' +
            this.name +
            '".';
          Logger.logAction(this.logger, Logger.LOG_ERROR, 'RealtimeChannel.processMessage()', msg);
          this._startDecodeFailureRecovery(new ErrorInfo(msg, 40018, 400));
          break;
        }

        let messages: Message[] = [];
        for (let i = 0; i < encoded.length; i++) {
          const { decoded, err } = await encoded[i].decodeWithErr(this._decodingContext, this.logger);
          messages[i] = decoded;

          if (err) {
            switch (err.code) {
              case 40018:
                /* decode failure */
                this._startDecodeFailureRecovery(err);
                return;

              case 40019: /* No vcdiff plugin passed in - no point recovering, give up */
              case 40021:
                /* Browser does not support deltas, similarly no point recovering */
                this.notifyState('failed', err);
                return;

              default:
              // do nothing, continue decoding
            }
          }
        }

        this._lastPayload.messageId = lastMessage.id;
        this._lastPayload.protocolMessageChannelSerial = message.channelSerial;
        this.onEvent(messages);
        break;
      }

      case actions.ANNOTATION: {
        populateFieldsFromParent(message);
        const options = this.channelOptions;
        if (this._annotations) {
          const annotations = await Promise.all(
            (message.annotations || []).map((wpm) => {
              return wpm.decode(options, this.logger);
            }),
          );

          this._annotations._processIncoming(annotations);
        }
        break;
      }

      case actions.ERROR: {
        /* there was a channel-specific error */
        const err = message.error as ErrorInfo;
        if (err && err.code == 80016) {
          /* attach/detach operation attempted on superseded transport handle */
          this.checkPendingState();
        } else {
          this.notifyState('failed', ErrorInfo.fromValues(err));
        }
        break;
      }

      default:
        // RSF1, should handle unrecognized message actions gracefully and don't abort the realtime connection to ensure forward compatibility
        Logger.logAction(
          this.logger,
          Logger.LOG_MAJOR,
          'RealtimeChannel.processMessage()',
          'Protocol error: unrecognised message action (' + message.action + ')',
        );
    }
  }

  _startDecodeFailureRecovery(reason: ErrorInfo): void {
    if (!this._lastPayload.decodeFailureRecoveryInProgress) {
      Logger.logAction(
        this.logger,
        Logger.LOG_MAJOR,
        'RealtimeChannel.processMessage()',
        'Starting decode failure recovery process.',
      );
      this._lastPayload.decodeFailureRecoveryInProgress = true;
      this._attach(true, reason, () => {
        this._lastPayload.decodeFailureRecoveryInProgress = false;
      });
    }
  }

  onAttached(): void {
    Logger.logAction(
      this.logger,
      Logger.LOG_MINOR,
      'RealtimeChannel.onAttached',
      'activating channel; name = ' + this.name,
    );
  }

  notifyState(
    state: API.ChannelState,
    reason?: ErrorInfo | null,
    resumed?: boolean,
    hasPresence?: boolean,
    hasBacklog?: boolean,
    hasObjects?: boolean,
  ): void {
    Logger.logAction(
      this.logger,
      Logger.LOG_MICRO,
      'RealtimeChannel.notifyState',
      'name = ' + this.name + ', current state = ' + this.state + ', notifying state ' + state,
    );
    this.clearStateTimer();

    // RTP5a1
    if (['detached', 'suspended', 'failed'].includes(state)) {
      this.properties.channelSerial = null;
    }

    if (state === this.state) {
      return;
    }
    if (this._presence) {
      this._presence.actOnChannelState(state, hasPresence, reason);
    }
    if (this._object) {
      this._object.actOnChannelState(state, hasObjects);
    }
    if (state === 'suspended' && this.connectionManager.state.sendEvents) {
      this.startRetryTimer();
    } else {
      this.cancelRetryTimer();
    }
    if (reason) {
      this.errorReason = reason;
    }
    const change = new ChannelStateChange(this.state, state, resumed, hasBacklog, reason);
    const action = 'Channel state for channel "' + this.name + '"';
    const message = state + (reason ? '; reason: ' + reason : '');
    if (state === 'failed') {
      Logger.logAction(this.logger, Logger.LOG_ERROR, action, message);
    } else {
      Logger.logAction(this.logger, Logger.LOG_MAJOR, action, message);
    }

    if (state !== 'attaching' && state !== 'suspended') {
      this.retryCount = 0;
    }

    /* Note: we don't set inProgress for pending states until the request is actually in progress */
    if (state === 'attached') {
      this.onAttached();
    }

    if (state === 'attached') {
      this._attachResume = true;
    } else if (state === 'detaching' || state === 'failed') {
      this._attachResume = false;
    }

    this.state = state;
    this.emit(state, change);
  }

  requestState(state: API.ChannelState, reason?: ErrorInfo | null): void {
    Logger.logAction(
      this.logger,
      Logger.LOG_MINOR,
      'RealtimeChannel.requestState',
      'name = ' + this.name + ', state = ' + state,
    );
    this.notifyState(state, reason);
    /* send the event and await response */
    this.checkPendingState();
  }

  checkPendingState(): void {
    /* if can't send events, do nothing */
    const cmState = this.connectionManager.state;
    if (!cmState.sendEvents) {
      Logger.logAction(
        this.logger,
        Logger.LOG_MINOR,
        'RealtimeChannel.checkPendingState',
        'sendEvents is false; state is ' + this.connectionManager.state.state,
      );
      return;
    }

    Logger.logAction(
      this.logger,
      Logger.LOG_MINOR,
      'RealtimeChannel.checkPendingState',
      'name = ' + this.name + ', state = ' + this.state,
    );
    /* Only start the state timer running when actually sending the event */
    switch (this.state) {
      case 'attaching':
        this.startStateTimerIfNotRunning();
        this.attachImpl();
        break;
      case 'detaching':
        this.startStateTimerIfNotRunning();
        this.detachImpl();
        break;
      case 'attached':
        /* resume any sync operation that was in progress */
        this.sync();
        break;
      default:
        break;
    }
  }

  timeoutPendingState(): void {
    switch (this.state) {
      case 'attaching': {
        const err = new ErrorInfo('Channel attach timed out', 90007, 408);
        this.notifyState('suspended', err);
        break;
      }
      case 'detaching': {
        const err = new ErrorInfo('Channel detach timed out', 90007, 408);
        this.notifyState('attached', err);
        break;
      }
      default:
        this.checkPendingState();
        break;
    }
  }

  startStateTimerIfNotRunning(): void {
    if (!this.stateTimer) {
      this.stateTimer = setTimeout(() => {
        Logger.logAction(this.logger, Logger.LOG_MINOR, 'RealtimeChannel.startStateTimerIfNotRunning', 'timer expired');
        this.stateTimer = null;
        this.timeoutPendingState();
      }, this.client.options.timeouts.realtimeRequestTimeout);
    }
  }

  clearStateTimer(): void {
    const stateTimer = this.stateTimer;
    if (stateTimer) {
      clearTimeout(stateTimer);
      this.stateTimer = null;
    }
  }

  startRetryTimer(): void {
    if (this.retryTimer) return;

    this.retryCount++;
    const retryDelay = Utils.getRetryTime(this.client.options.timeouts.channelRetryTimeout, this.retryCount);

    this.retryTimer = setTimeout(() => {
      /* If connection is not connected, just leave in suspended, a reattach
       * will be triggered once it connects again */
      if (this.state === 'suspended' && this.connectionManager.state.sendEvents) {
        this.retryTimer = null;
        Logger.logAction(
          this.logger,
          Logger.LOG_MINOR,
          'RealtimeChannel retry timer expired',
          'attempting a new attach',
        );
        this.requestState('attaching');
      }
    }, retryDelay);
  }

  cancelRetryTimer(): void {
    if (this.retryTimer) {
      clearTimeout(this.retryTimer as NodeJS.Timeout);
      this.retryTimer = null;
    }
  }

  history = async function (
    this: RealtimeChannel,
    params: RealtimeHistoryParams | null,
  ): Promise<PaginatedResult<Message>> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.history()', 'channel = ' + this.name);

    // We fetch this first so that any plugin-not-provided error takes priority over other errors
    const restMixin = this.client.rest.channelMixin;

    if (params && params.untilAttach) {
      if (this.state !== 'attached') {
        throw new ErrorInfo('option untilAttach requires the channel to be attached', 40000, 400);
      }
      if (!this.properties.attachSerial) {
        throw new ErrorInfo(
          'untilAttach was specified and channel is attached, but attachSerial is not defined',
          40000,
          400,
        );
      }
      delete params.untilAttach;
      params.from_serial = this.properties.attachSerial;
    }

    return restMixin.history(this, params);
  } as any;

  whenState = ((state: string) => {
    return EventEmitter.prototype.whenState.call(this, state, this.state);
  }) as any;

  /* @returns null (if can safely be released) | ErrorInfo (if cannot) */
  getReleaseErr(): ErrorInfo | null {
    const s = this.state;
    if (s === 'initialized' || s === 'detached' || s === 'failed') {
      return null;
    }
    return new ErrorInfo(
      'Can only release a channel in a state where there is no possibility of further updates from the server being received (initialized, detached, or failed); was ' +
        s,
      90001,
      400,
    );
  }

  setChannelSerial(channelSerial?: string | null): void {
    Logger.logAction(
      this.logger,
      Logger.LOG_MICRO,
      'RealtimeChannel.setChannelSerial()',
      'Updating channel serial; serial = ' + channelSerial + '; previous = ' + this.properties.channelSerial,
    );

    // RTP17h: Only update the channel serial if its present (it won't always
    // be set).
    if (channelSerial) {
      this.properties.channelSerial = channelSerial;
    }
  }

  async status(): Promise<API.ChannelDetails> {
    return this.client.rest.channelMixin.status(this);
  }

  async getMessage(serialOrMessage: string | Message): Promise<Message> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.getMessage()', 'channel = ' + this.name);
    const restMixin = this.client.rest.channelMixin;
    return restMixin.getMessage(this, serialOrMessage);
  }

  async updateMessage(
    message: Message,
    operation?: API.MessageOperation,
    params?: Record<string, any>,
  ): Promise<API.UpdateDeleteResult> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.updateMessage()', 'channel = ' + this.name);
    return this.sendUpdate(message, 'message.update', operation, params);
  }

  async deleteMessage(
    message: Message,
    operation?: API.MessageOperation,
    params?: Record<string, any>,
  ): Promise<API.UpdateDeleteResult> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.deleteMessage()', 'channel = ' + this.name);
    return this.sendUpdate(message, 'message.delete', operation, params);
  }

  async appendMessage(
    message: Message,
    operation?: API.MessageOperation,
    params?: Record<string, any>,
  ): Promise<API.UpdateDeleteResult> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.appendMessage()', 'channel = ' + this.name);
    return this.sendUpdate(message, 'message.append', operation, params);
  }

  private async sendUpdate(
    message: Message,
    action: 'message.update' | 'message.delete' | 'message.append',
    operation?: API.MessageOperation,
    params?: Record<string, any>,
  ): Promise<API.UpdateDeleteResult> {
    if (!message.serial) {
      throw new ErrorInfo(
        'This message lacks a serial and cannot be updated. Make sure you have enabled "Message annotations, updates, and deletes" in channel settings on your dashboard.',
        40003,
        400,
      );
    }

    this.throwIfUnpublishableState();

    const updateDeleteMsg = Message.fromValues({
      ...message,
      action: action,
      version: operation,
    });

    const wireMessage = await updateDeleteMsg.encode(this.channelOptions as CipherOptions);
    const pm = protocolMessageFromValues({
      action: actions.MESSAGE,
      channel: this.name,
      messages: [wireMessage],
      params: params ? Utils.stringifyValues(params) : undefined,
    });
    const publishResponse = await this.sendAndAwaitAck(pm);
    return { versionSerial: publishResponse.serials[0] ?? null };
  }

  async getMessageVersions(
    serialOrMessage: string | Message,
    params?: Record<string, any>,
  ): Promise<PaginatedResult<Message>> {
    Logger.logAction(this.logger, Logger.LOG_MICRO, 'RealtimeChannel.getMessageVersions()', 'channel = ' + this.name);
    const restMixin = this.client.rest.channelMixin;
    return restMixin.getMessageVersions(this, serialOrMessage, params);
  }

  /**
   * Ensures the channel is attached, attaching if necessary.
   *
   * This method is intended for use by features like Presence or Objects that need to
   * implicitly attach the channel when an operation is called (e.g., `presence.get()` per RTP11b,
   * or `objects.get()`). This guarantees that the corresponding sync sequence will start and
   * that the operation will resolve for callers even if they did not explicitly attach beforehand.
   */
  async ensureAttached(): Promise<void> {
    switch (this.state) {
      case 'attached':
      case 'suspended':
        break;
      case 'initialized':
      case 'detached':
      case 'detaching':
      case 'attaching':
        await this.attach();
        break;
      case 'failed':
      default:
        throw ErrorInfo.fromValues(this.invalidStateError());
    }
  }
}

function omitAgent(channelParams?: API.ChannelParams) {
  const { agent: _, ...paramsWithoutAgent } = channelParams || {};
  return paramsWithoutAgent;
}

export default RealtimeChannel;