refactor(plugin-notification): in app message ws (#7302)

* refactor(plugin-notification-in-app-message): change to ws instead of sse

* revert(plugin-notification-in-app-message): use legacy ws api

* test(plugin-notification-in-app-message): fix test cases

packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/components/Inbox.tsx

@@ -16,11 +16,10 @@
  * For more information, please rwefer to: https://www.nocobase.com/agreement.
  */
 
-import { reaction } from '@formily/reactive';
 import { observer } from '@formily/reactive-react';
-import { Icon, useCurrentUserContext, useMobileLayout } from '@nocobase/client';
+import { Icon, useApp, useCurrentUserContext, useMobileLayout } from '@nocobase/client';
 import { MobilePopup } from '@nocobase/plugin-mobile/client';
-import { Badge, Button, ConfigProvider, Drawer, notification, theme, Tooltip } from 'antd';
+import { Badge, Button, ConfigProvider, Drawer, theme, Tooltip } from 'antd';
 import { createStyles } from 'antd-style';
 import React, { FC, useCallback, useEffect, useState } from 'react';
 import { useLocalTranslation } from '../../locale';
@@ -28,10 +27,7 @@ import { Channel } from '../../types';
 import {
   fetchChannels,
   inboxVisible,
-  liveSSEObs,
   messageMapObs,
-  selectedChannelNameObs,
-  startMsgSSEStreamWithRetry,
   unreadMsgsCountObs,
   updateUnreadMsgsCount,
   userIdObs,
@@ -89,11 +85,17 @@ const InboxPopup: FC<{ title: string; visible: boolean; onClose: () => void }> =
 };
 
 const InnerInbox = (props) => {
+  const app = useApp();
   const { t } = useLocalTranslation();
   const { styles } = useStyles();
   const ctx = useCurrentUserContext();
   const currUserId = ctx.data?.data?.id;
-  const { token } = theme.useToken();
+
+  const onMessageUpdated = useCallback(({ detail }: CustomEvent) => {
+    messageMapObs.value[detail.id] = detail;
+    fetchChannels({ filter: { name: detail.channelName } });
+    updateUnreadMsgsCount();
+  }, []);
 
   useEffect(() => {
     updateUnreadMsgsCount();
@@ -108,64 +110,13 @@ const InnerInbox = (props) => {
   }, []);
 
   useEffect(() => {
-    const disposes: Array<() => void> = [];
-    disposes.push(startMsgSSEStreamWithRetry());
-    const disposeAll = () => {
-      while (disposes.length > 0) {
-        const dispose = disposes.pop();
-        dispose && dispose();
-      }
-    };
+    app.eventBus.addEventListener('ws:message:in-app-message:updated', onMessageUpdated);
 
-    const onVisibilityChange = () => {
-      if (document.visibilityState === 'visible') {
-        disposes.push(startMsgSSEStreamWithRetry());
-      } else {
-        disposeAll();
-      }
-    };
-
-    document.addEventListener('visibilitychange', onVisibilityChange);
     return () => {
-      disposeAll();
-      document.removeEventListener('visibilitychange', onVisibilityChange);
+      app.eventBus.removeEventListener('ws:message:in-app-message:updated', onMessageUpdated);
     };
-  }, []);
-  useEffect(() => {
-    const dispose = reaction(
-      () => liveSSEObs.value,
-      (sseData) => {
-        if (!sseData) return;
+  }, [app.eventBus, onMessageUpdated]);
 
-        if (['message:created', 'message:updated'].includes(sseData.type)) {
-          const { data } = sseData;
-          messageMapObs.value[data.id] = data;
-          if (sseData.type === 'message:created') {
-            notification.info({
-              message: (
-                <div
-                  style={{
-                    textOverflow: 'ellipsis',
-                    overflow: 'hidden',
-                    whiteSpace: 'nowrap',
-                  }}
-                >
-                  {data.title}
-                </div>
-              ),
-              description: data.content.slice(0, 100) + (data.content.length > 100 ? '...' : ''),
-              onClick: () => {
-                inboxVisible.value = true;
-                selectedChannelNameObs.value = data.channelName;
-                notification.destroy();
-              },
-            });
-          }
-        }
-      },
-    );
-    return dispose;
-  }, []);
   return (
     <ConfigProvider
       theme={{

packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/components/mobile/MobileTabBarMessageItem.tsx

@@ -9,9 +9,9 @@
 
 import { observer } from '@formily/reactive-react';
 import { MobileTabBarItem } from '@nocobase/plugin-mobile/client';
-import React, { useEffect } from 'react';
+import React from 'react';
 import { useLocation, useNavigate } from 'react-router-dom';
-import { startMsgSSEStreamWithRetry, unreadMsgsCountObs, updateUnreadMsgsCount } from '../../observables';
+import { unreadMsgsCountObs } from '../../observables';
 
 const InnerMobileTabBarMessageItem = (props) => {
   const navigate = useNavigate();
@@ -20,31 +20,6 @@ const InnerMobileTabBarMessageItem = (props) => {
     navigate('/page/in-app-message');
   };
 
-  useEffect(() => {
-    const disposes: Array<() => void> = [];
-    disposes.push(startMsgSSEStreamWithRetry());
-    const disposeAll = () => {
-      while (disposes.length > 0) {
-        const dispose = disposes.pop();
-        dispose && dispose();
-      }
-    };
-
-    const onVisibilityChange = () => {
-      if (document.visibilityState === 'visible') {
-        disposes.push(startMsgSSEStreamWithRetry());
-      } else {
-        disposeAll();
-      }
-    };
-
-    document.addEventListener('visibilitychange', onVisibilityChange);
-    return () => {
-      disposeAll();
-      document.removeEventListener('visibilitychange', onVisibilityChange);
-    };
-  }, []);
-
   const selected = props.url && location.pathname.startsWith(props.url);
 
   return (

packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/index.ts

@@ -9,6 +9,5 @@
 
 export * from './channel';
 export * from './message';
-export * from './sse';
 export * from './inbox';
 export * from './user';

packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/sse.ts

@@ -1,90 +0,0 @@
-/**
- * This file is part of the NocoBase (R) project.
- * Copyright (c) 2020-2024 NocoBase Co., Ltd.
- * Authors: NocoBase Team.
- *
- * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License.
- * For more information, please refer to: https://www.nocobase.com/agreement.
- */
-
-import { observable, reaction } from '@formily/reactive';
-import { uid } from '@nocobase/utils/client';
-import { SSEData } from '../../types';
-import { getAPIClient } from '../utils';
-import { fetchChannels } from './channel';
-import { messageMapObs, updateUnreadMsgsCount } from './message';
-
-export const liveSSEObs = observable<{ value: SSEData | null }>({ value: null });
-reaction(
-  () => liveSSEObs.value,
-  (sseData) => {
-    if (!sseData) return;
-
-    if (['message:created', 'message:updated'].includes(sseData.type)) {
-      const { data } = sseData;
-      messageMapObs.value[data.id] = data;
-      fetchChannels({ filter: { name: data.channelName } });
-      updateUnreadMsgsCount();
-    }
-  },
-);
-export const startMsgSSEStreamWithRetry: () => () => void = () => {
-  let retryTimes = 0;
-  const controller = new AbortController();
-  let disposed = false;
-  const dispose = () => {
-    disposed = true;
-    controller.abort();
-  };
-
-  const clientId = uid();
-  const createMsgSSEConnection = async (clientId: string) => {
-    await updateUnreadMsgsCount();
-    const apiClient = getAPIClient();
-    const res = await apiClient.silent().request({
-      url: 'myInAppMessages:sse',
-      skipAuth: true,
-      method: 'get',
-      signal: controller.signal,
-      headers: {
-        Accept: 'text/event-stream',
-      },
-      params: {
-        id: clientId,
-      },
-      responseType: 'stream',
-      adapter: 'fetch',
-      skipNotify: (error) => (!error || !error.message ? true : false),
-    });
-    if (!res?.data) return;
-    const stream = res.data;
-    const reader = stream.pipeThrough(new TextDecoderStream()).getReader();
-    retryTimes = 0;
-    // eslint-disable-next-line no-constant-condition
-    while (true) {
-      const { value, done } = await reader.read();
-      if (done) break;
-      const messages = value.split('\n\n').filter(Boolean);
-      for (const message of messages) {
-        const sseData: SSEData = JSON.parse(message.replace(/^data:\s*/, '').trim());
-        liveSSEObs.value = sseData;
-      }
-    }
-  };
-
-  const connectWithRetry = async () => {
-    try {
-      await createMsgSSEConnection(clientId);
-    } catch (error) {
-      console.log(`sse connection for clientId ${clientId} failed, retrying...`, error);
-      const nextDelay = retryTimes < 6 ? 1000 * Math.pow(2, retryTimes) : 60000;
-      retryTimes++;
-      setTimeout(() => {
-        if (!disposed) connectWithRetry();
-      }, nextDelay);
-      return { error };
-    }
-  };
-  connectWithRetry();
-  return dispose;
-};

packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/InAppNotificationChannel.ts

@@ -19,60 +19,29 @@ import defineMyInAppChannels from './defineMyInAppChannels';
 type UserID = string;
 type ClientID = string;
 export default class InAppNotificationChannel extends BaseNotificationChannel {
-  userClientsMap: Record<UserID, Record<ClientID, PassThrough>>;
+  // userClientsMap: Record<UserID, Record<ClientID, PassThrough>>;
 
-  constructor(protected app: Application) {
-    super(app);
-    this.userClientsMap = {};
-  }
+  // constructor(protected app: Application) {
+  //   super(app);
+  //   this.userClientsMap = {};
+  // }
 
   async load() {
-    this.onMessageCreatedOrUpdated();
+    this.app.db.on(`${MessagesDefinition.name}.afterSave`, this.onMessageCreatedOrUpdated);
     this.defineActions();
   }
-  onMessageCreatedOrUpdated = async () => {
-    this.app.db.on(`${MessagesDefinition.name}.afterUpdate`, async (model, options) => {
-      const userId = model.userId;
-      this.sendDataToUser(userId, { type: 'message:updated', data: model.dataValues });
-    });
-    this.app.db.on(`${MessagesDefinition.name}.afterCreate`, async (model, options) => {
-      const userId = model.userId;
-      this.sendDataToUser(userId, { type: 'message:created', data: model.dataValues });
-    });
-  };
 
-  addClient = (userId: UserID, clientId: ClientID, stream: PassThrough) => {
-    if (!this.userClientsMap[userId]) {
-      this.userClientsMap[userId] = {};
-    }
-    this.userClientsMap[userId][clientId] = stream;
+  onMessageCreatedOrUpdated = async (model, options) => {
+    const userId = model.userId;
+    this.app.emit('ws:sendToTag', {
+      tagKey: 'userId',
+      tagValue: userId,
+      message: {
+        type: 'in-app-message:updated',
+        payload: model.toJSON(),
+      },
+    });
   };
-  getClient = (userId: UserID, clientId: ClientID) => {
-    return this.userClientsMap[userId]?.[clientId];
-  };
-  removeClient = (userId: UserID, clientId: ClientID) => {
-    if (this.userClientsMap[userId]) {
-      delete this.userClientsMap[userId][clientId];
-    }
-  };
-  sendDataToUser(userId: UserID, message: { type: string; data: any }) {
-    const clients = this.userClientsMap[userId];
-    if (clients) {
-      for (const clientId in clients) {
-        const stream = clients[clientId];
-        stream.write(
-          `data: ${JSON.stringify({
-            type: message.type,
-            data: {
-              ...message.data,
-              title: message.data.title || '',
-              content: message.data.content || '',
-            },
-          })}\n\n`,
-        );
-      }
-    }
-  }
 
   saveMessageToDB = async ({
     content,
@@ -132,13 +101,8 @@ export default class InAppNotificationChannel extends BaseNotificationChannel {
   };
 
   defineActions() {
-    defineMyInAppMessages({
-      app: this.app,
-      addClient: this.addClient,
-      removeClient: this.removeClient,
-      getClient: this.getClient,
-    });
-    defineMyInAppChannels({ app: this.app });
+    defineMyInAppMessages(this.app);
+    defineMyInAppChannels(this.app);
     this.app.acl.allow('myInAppMessages', '*', 'loggedIn');
     this.app.acl.allow('myInAppChannels', '*', 'loggedIn');
     this.app.acl.allow('notificationInAppMessages', '*', 'loggedIn');

packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/__tests__/server.test.ts

@@ -58,8 +58,8 @@ describe('inapp message channels', () => {
       await messagesRepo.destroy({ truncate: true });
     });
     test('user can get own channels and messages', async () => {
-      defineMyInAppChannels({ app });
-      defineMyInAppMessages({ app, addClient: () => null, removeClient: () => null });
+      defineMyInAppChannels(app);
+      defineMyInAppMessages(app);
       const channelsRes = await channelsRepo.create({
         values: [
           {
@@ -87,8 +87,8 @@ describe('inapp message channels', () => {
     });
 
     test('user can get own channel latestMesageTitle and latestMesageTimestamp', async () => {
-      defineMyInAppChannels({ app });
-      defineMyInAppMessages({ app, addClient: () => null, removeClient: () => null });
+      defineMyInAppChannels(app);
+      defineMyInAppMessages(app);
       const channelsRes = await channelsRepo.create({
         values: [
           {

packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppChannels.ts

@@ -12,7 +12,7 @@ import { Op, Sequelize } from 'sequelize';
 import { ChannelsCollectionDefinition as ChannelsDefinition } from '@nocobase/plugin-notification-manager';
 import { InAppMessagesDefinition as MessagesDefinition } from '../types';
 
-export default function defineMyInAppChannels({ app }: { app: Application }) {
+export default function defineMyInAppChannels(app: Application) {
   app.resourceManager.define({
     name: 'myInAppChannels',
     actions: {

packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppMessages.ts

@@ -12,19 +12,9 @@ import actions from '@nocobase/actions';
 import { ChannelsCollectionDefinition as ChannelsDefinition } from '@nocobase/plugin-notification-manager';
 import { Application } from '@nocobase/server';
 import { Op, Sequelize } from 'sequelize';
-import { PassThrough } from 'stream';
 import { InAppMessagesDefinition as MessagesDefinition } from '../types';
-export default function defineMyInAppMessages({
-  app,
-  addClient,
-  removeClient,
-  getClient,
-}: {
-  app: Application;
-  addClient: any;
-  removeClient: any;
-  getClient?: any;
-}) {
+
+export default function defineMyInAppMessages(app: Application) {
   const countTotalUnreadMessages = async (userId: string) => {
     const messagesRepo = app.db.getRepository(MessagesDefinition.name);
     const channelsCollection = app.db.getCollection(ChannelsDefinition.name);
@@ -50,32 +40,32 @@ export default function defineMyInAppMessages({
   app.resourceManager.define({
     name: 'myInAppMessages',
     actions: {
-      sse: {
-        handler: async (ctx, next) => {
-          const userId = ctx.state.currentUser.id;
-          const clientId = ctx.action?.params?.id;
-          if (!clientId) return;
-          ctx.request.socket.setTimeout(0);
-          ctx.req.socket.setNoDelay(true);
-          ctx.req.socket.setKeepAlive(true);
-          ctx.set({
-            'Content-Type': 'text/event-stream',
-            'Cache-Control': 'no-cache',
-            Connection: 'keep-alive',
-          });
-          const stream = new PassThrough();
-          ctx.status = 200;
-          ctx.body = stream;
-          addClient(userId, clientId, stream);
-          stream.on('close', () => {
-            removeClient(userId, clientId);
-          });
-          stream.on('error', () => {
-            removeClient(userId, clientId);
-          });
-          await next();
-        },
-      },
+      // sse: {
+      //   handler: async (ctx, next) => {
+      //     const userId = ctx.state.currentUser.id;
+      //     const clientId = ctx.action?.params?.id;
+      //     if (!clientId) return;
+      //     ctx.request.socket.setTimeout(0);
+      //     ctx.req.socket.setNoDelay(true);
+      //     ctx.req.socket.setKeepAlive(true);
+      //     ctx.set({
+      //       'Content-Type': 'text/event-stream',
+      //       'Cache-Control': 'no-cache',
+      //       Connection: 'keep-alive',
+      //     });
+      //     const stream = new PassThrough();
+      //     ctx.status = 200;
+      //     ctx.body = stream;
+      //     addClient(userId, clientId, stream);
+      //     stream.on('close', () => {
+      //       removeClient(userId, clientId);
+      //     });
+      //     stream.on('error', () => {
+      //       removeClient(userId, clientId);
+      //     });
+      //     await next();
+      //   },
+      // },
       count: {
         handler: async (ctx) => {
           try {