packages/js/src/ws/socket.ts - next - githubmirror/novu

packages/js/src/ws/socket.ts
245 строк7 KB
Adam Chmara
feat(js,react,api): wire agent-chat live WS, status, fetchMore, approvals fixes NV-8445 (#12292)
10 авг 2026, 12:55
10 авг 2026, 12:5505d4c47
import type { AgentEventEnvelope } from '@novu/agent-event-protocol';
import io, { Socket as SocketIO } from 'socket.io-client';
import { InboxService } from '../api';
import { BaseModule } from '../base-module';
import {
  AgentChatAgentEvent,
  NotificationReceivedEvent,
  NotificationUnreadEvent,
  NotificationUnseenEvent,
  NovuEventEmitter,
  SocketEventNames,
} from '../event-emitter';
import { Notification } from '../notifications';
import {
  ActionTypeEnum,
  InboxNotification,
  NotificationActionStatus,
  Result,
  Session,
  Subscriber,
  TODO,
  WebSocketEvent,
} from '../types';
import { NovuError } from '../utils/errors';
import { sanitizeInAppRedirect } from '../utils/in-app-redirect-url';
import type { BaseSocketInterface } from './base-socket';

const PRODUCTION_SOCKET_URL = 'https://ws.novu.co';
const NOTIFICATION_RECEIVED: NotificationReceivedEvent = 'notifications.notification_received';
const UNSEEN_COUNT_CHANGED: NotificationUnseenEvent = 'notifications.unseen_count_changed';
const UNREAD_COUNT_CHANGED: NotificationUnreadEvent = 'notifications.unread_count_changed';
const AGENT_CHAT_AGENT_EVENT: AgentChatAgentEvent = 'agent_chat.agent_event';

const mapToNotification = ({
  _id,
  transactionId,
  content,
  read,
  seen,
  archived,
  snoozedUntil,
  deliveredAt,
  createdAt,
  lastReadDate,
  firstSeenDate,
  archivedAt,
  channel,
  subscriber,
  subject,
  avatar,
  cta,
  tags,
  data,
  workflow,
  severity,
}: TODO): InboxNotification => {
  const to: Subscriber = {
    id: subscriber?._id,
    subscriberId: subscriber?.subscriberId,
    firstName: subscriber?.firstName,
    lastName: subscriber?.lastName,
    avatar: subscriber?.avatar,
    locale: subscriber?.locale,
    data: subscriber?.data,
    timezone: subscriber?.timezone,
    email: subscriber?.email,
    phone: subscriber?.phone,
  };
  const primaryCta = cta.action?.buttons?.find((button: any) => button.type === ActionTypeEnum.PRIMARY);
  const secondaryCta = cta.action?.buttons?.find((button: any) => button.type === ActionTypeEnum.SECONDARY);
  const actionType = cta.action?.result?.type;
  const actionStatus = cta.action?.status;

  return {
    id: _id,
    transactionId,
    subject,
    body: content as string,
    to,
    isRead: read,
    isSeen: seen,
    isArchived: archived,
    isSnoozed: !!snoozedUntil,
    ...(deliveredAt && {
      deliveredAt,
    }),
    ...(snoozedUntil && {
      snoozedUntil,
    }),
    createdAt,
    readAt: lastReadDate,
    firstSeenAt: firstSeenDate,
    archivedAt,
    avatar,
    primaryAction: primaryCta && {
      label: primaryCta.content,
      isCompleted: actionType === ActionTypeEnum.PRIMARY && actionStatus === NotificationActionStatus.DONE,
      redirect: sanitizeInAppRedirect(primaryCta.url, primaryCta.target),
    },
    secondaryAction: secondaryCta && {
      label: secondaryCta.content,
      isCompleted: actionType === ActionTypeEnum.SECONDARY && actionStatus === NotificationActionStatus.DONE,
      redirect: sanitizeInAppRedirect(secondaryCta.url, secondaryCta.target),
    },
    channelType: channel,
    tags,
    redirect: sanitizeInAppRedirect(cta.data?.url, cta.data?.target),
    data,
    workflow,
    severity,
  };
};

export class Socket extends BaseModule implements BaseSocketInterface {
  #token: string;
  #emitter: NovuEventEmitter;
  #socketIo: SocketIO | undefined;
  #socketUrl: string;
  #socketOptions?: Record<string, unknown>;

  constructor({
    socketUrl,
    socketOptions,
    inboxServiceInstance,
    eventEmitterInstance,
  }: {
    socketUrl?: string;
    socketOptions?: Record<string, unknown>;
    inboxServiceInstance: InboxService;
    eventEmitterInstance: NovuEventEmitter;
  }) {
    super({
      eventEmitterInstance,
      inboxServiceInstance,
    });
    this.#emitter = eventEmitterInstance;
    this.#socketUrl = socketUrl ?? PRODUCTION_SOCKET_URL;
    this.#socketOptions = socketOptions;
  }

  protected onSessionSuccess({ token }: Session): void {
    this.#token = token;
  }

  #notificationReceived = ({ message }: { message: TODO }) => {
    this.#emitter.emit(NOTIFICATION_RECEIVED, {
      result: new Notification(mapToNotification(message), this.#emitter, this._inboxService),
    });
  };

  #unseenCountChanged = ({ unseenCount }: { unseenCount: number }) => {
    this.#emitter.emit(UNSEEN_COUNT_CHANGED, {
      result: unseenCount,
    });
  };

  #unreadCountChanged = ({ counts }: { counts: { total: number; severity: Record<string, number> } }) => {
    this.#emitter.emit(UNREAD_COUNT_CHANGED, {
      result: counts,
    });
  };

  #agentEvent = (envelope: AgentEventEnvelope) => {
    this.#emitter.emit(AGENT_CHAT_AGENT_EVENT, {
      result: envelope,
    });
  };

  async #initializeSocket(): Promise<void> {
    if (this.#socketIo) {
      return;
    }

    const args = { socketUrl: this.#socketUrl };
    this.#emitter.emit('socket.connect.pending', { args });

    this.#socketIo = io(this.#socketUrl, {
      reconnectionDelayMax: 10000,
      transports: ['websocket'],
      query: {
        token: `${this.#token}`,
      },
      ...(this.#socketOptions ?? {}),
    });

    this.#socketIo.on('connect', () => {
      this.#emitter.emit('socket.connect.resolved', { args });
    });

    this.#socketIo.on('connect_error', (error) => {
      this.#emitter.emit('socket.connect.resolved', { args, error });
    });

    this.#socketIo?.on(WebSocketEvent.RECEIVED, this.#notificationReceived);
    this.#socketIo?.on(WebSocketEvent.UNSEEN, this.#unseenCountChanged);
    this.#socketIo?.on(WebSocketEvent.UNREAD, this.#unreadCountChanged);
    this.#socketIo?.on(WebSocketEvent.AGENT_EVENT, this.#agentEvent);
  }

  async #handleConnectSocket(): Result<void> {
    try {
      await this.#initializeSocket();

      return {};
    } catch (error) {
      return { error: new NovuError('Failed to initialize the socket', error) };
    }
  }

  async #handleDisconnectSocket(): Result<void> {
    try {
      this.#socketIo?.disconnect();
      this.#socketIo = undefined;

      return {};
    } catch (error) {
      return { error: new NovuError('Failed to disconnect from the socket', error) };
    }
  }

  isSocketEvent(eventName: string): eventName is SocketEventNames {
    return (
      eventName === NOTIFICATION_RECEIVED ||
      eventName === UNSEEN_COUNT_CHANGED ||
      eventName === UNREAD_COUNT_CHANGED ||
      eventName === AGENT_CHAT_AGENT_EVENT
    );
  }

  async connect(): Result<void> {
    if (this.#token) {
      return this.#handleConnectSocket();
    }

    return this.callWithSession(this.#handleConnectSocket.bind(this));
  }

  async disconnect(): Result<void> {
    if (this.#socketIo) {
      return this.#handleDisconnectSocket();
    }

    return this.callWithSession(this.#handleDisconnectSocket.bind(this));
  }
}