diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/components/Inbox.tsx b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/components/Inbox.tsx index 67580f204bb..766edced495 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/components/Inbox.tsx +++ b/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: ( -
- {data.title} -
- ), - description: data.content.slice(0, 100) + (data.content.length > 100 ? '...' : ''), - onClick: () => { - inboxVisible.value = true; - selectedChannelNameObs.value = data.channelName; - notification.destroy(); - }, - }); - } - } - }, - ); - return dispose; - }, []); return ( { 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 ( diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/index.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/index.ts index 1d8254a68f5..9910c90d553 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/index.ts +++ b/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'; diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/sse.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/sse.ts deleted file mode 100644 index 4d07949ca7f..00000000000 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/client/observables/sse.ts +++ /dev/null @@ -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; -}; diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/InAppNotificationChannel.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/InAppNotificationChannel.ts index 49696f49ee8..4fddc026496 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/InAppNotificationChannel.ts +++ b/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>; + // userClientsMap: Record>; - 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'); diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/__tests__/server.test.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/__tests__/server.test.ts index 375547afbfb..d7678d5c85f 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/__tests__/server.test.ts +++ b/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: [ { diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppChannels.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppChannels.ts index d362ebdecc6..fef962ad5ac 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppChannels.ts +++ b/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: { diff --git a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppMessages.ts b/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppMessages.ts index 2eb9eb4c81d..bd3e2f8716b 100644 --- a/packages/plugins/@nocobase/plugin-notification-in-app-message/src/server/defineMyInAppMessages.ts +++ b/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 {