import { sanitizedRaw } from '@nozbe/watermelondb/RawRecord'; import { InteractionManager } from 'react-native'; import EJSON from 'ejson'; import Model from '@nozbe/watermelondb/Model'; import database from '../../database'; import { merge } from '../helpers/mergeSubscriptionsRooms'; import protectedFunction from '../helpers/protectedFunction'; import log from '../../../utils/log'; import random from '../../../utils/random'; import { store } from '../../auxStore'; import { handlePayloadUserInteraction } from '../actions'; import buildMessage from '../helpers/buildMessage'; import RocketChat from '../../rocketchat'; import EventEmitter from '../../../utils/events'; import { removedRoom } from '../../../actions/room'; import { setUser } from '../../../actions/login'; import { INAPP_NOTIFICATION_EMITTER } from '../../../containers/InAppNotification'; import { Encryption } from '../../encryption'; import { E2E_MESSAGE_TYPE } from '../../encryption/constants'; import updateMessages from '../updateMessages'; import { IMessage, IRoom, ISubscription, TMessageModel, TRoomModel, TThreadMessageModel, TThreadModel } from '../../../definitions'; import sdk from '../../rocketchat/services/sdk'; import { IDDPMessage } from '../../../definitions/IDDPMessage'; import { getSubscriptionByRoomId } from '../../database/services/Subscription'; import { getMessageById } from '../../database/services/Message'; const removeListener = (listener: { stop: () => void }) => listener.stop(); let streamListener: Promise | false; let subServer: string; let queue: { [key: string]: ISubscription } = {}; let subTimer: number | null | false = null; const WINDOW_TIME = 500; const createOrUpdateSubscription = async (subscription: ISubscription, room: IRoom | ISubscription) => { try { const db = database.active; const subCollection = db.get('subscriptions'); const roomsCollection = db.get('rooms'); if (!subscription) { try { const s = await subCollection.find(room._id); // We have to create a plain obj so we can manipulate it on `merge` // Can we do it in a better way? subscription = { _id: s._id, f: s.f, t: s.t, ts: s.ts, ls: s.ls, name: s.name, fname: s.fname, rid: s.rid, open: s.open, alert: s.alert, unread: s.unread, userMentions: s.userMentions, roomUpdatedAt: s.roomUpdatedAt, ro: s.ro, lastOpen: s.lastOpen, description: s.description, announcement: s.announcement, bannerClosed: s.bannerClosed, topic: s.topic, blocked: s.blocked, blocker: s.blocker, reactWhenReadOnly: s.reactWhenReadOnly, archived: s.archived, joinCodeRequired: s.joinCodeRequired, muted: s.muted, ignored: s.ignored, broadcast: s.broadcast, prid: s.prid, draftMessage: s.draftMessage, lastThreadSync: s.lastThreadSync, jitsiTimeout: s.jitsiTimeout, autoTranslate: s.autoTranslate, autoTranslateLanguage: s.autoTranslateLanguage, lastMessage: s.lastMessage, roles: s.roles, usernames: s.usernames, uids: s.uids, visitor: s.visitor, departmentId: s.departmentId, servedBy: s.servedBy, livechatData: s.livechatData, tags: s.tags, encrypted: s.encrypted, e2eKeyId: s.e2eKeyId, E2EKey: s.E2EKey, avatarETag: s.avatarETag } as ISubscription; } catch (error) { try { await db.write(async () => { await roomsCollection.create( protectedFunction((r: TRoomModel) => { r._raw = sanitizedRaw({ id: room._id }, roomsCollection.schema); Object.assign(r, room); }) ); }); } catch (e) { // Do nothing } return; } } if (!room && subscription) { try { const r = await roomsCollection.find(subscription.rid); // We have to create a plain obj so we can manipulate it on `merge` // Can we do it in a better way? room = { v: r.v, ro: r.ro, tags: r.tags, servedBy: r.servedBy, encrypted: r.encrypted, e2eKeyId: r.e2eKeyId, broadcast: r.broadcast, customFields: r.customFields, departmentId: r.departmentId, livechatData: r.livechatData, avatarETag: r.avatarETag } as IRoom; } catch (error) { // Do nothing } } let tmp = merge(subscription, room); tmp = (await Encryption.decryptSubscription(tmp)) as ISubscription; const sub = await getSubscriptionByRoomId(tmp.rid); // If we're receiving a E2EKey of a room if (sub && !sub.E2EKey && subscription?.E2EKey) { // Assing info from database subscription to tmp // It should be a plain object tmp = Object.assign(tmp, { rid: sub.rid, encrypted: sub.encrypted, lastMessage: sub.lastMessage, E2EKey: subscription.E2EKey, e2eKeyId: sub.e2eKeyId }); // Decrypt lastMessage using the received E2EKey tmp = (await Encryption.decryptSubscription(tmp)) as ISubscription; // Decrypt all pending messages of this room in parallel Encryption.decryptPendingMessages(tmp.rid); } const batch: Model[] = []; if (sub) { try { const update = sub.prepareUpdate(s => { Object.assign(s, tmp); if (subscription.announcement) { if (subscription.announcement !== sub.announcement) { s.bannerClosed = false; } } }); batch.push(update); } catch (e) { console.log(e); } } else { try { const create = subCollection.prepareCreate(s => { s._raw = sanitizedRaw({ id: tmp.rid }, subCollection.schema); Object.assign(s, tmp); if (s.roomUpdatedAt) { s.roomUpdatedAt = new Date(); } }); batch.push(create); } catch (e) { console.log(e); } } const { rooms } = store.getState().room; if (tmp.lastMessage && !rooms.includes(tmp.rid)) { const lastMessage = buildMessage(tmp.lastMessage); const messagesCollection = db.get('messages'); let messageRecord = {} as TMessageModel | null; if (lastMessage) { messageRecord = await getMessageById(lastMessage._id); } if (messageRecord) { batch.push( messageRecord.prepareUpdate(() => { Object.assign(messageRecord, lastMessage); }) ); } else { batch.push( messagesCollection.prepareCreate(m => { if (lastMessage) { m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema); if (m.subscription) { m.subscription.id = lastMessage.rid; } } return Object.assign(m, lastMessage); }) ); } } await db.write(async () => { await db.batch(...batch); }); } catch (e) { log(e); } }; const getSubQueueId = (rid: string) => `SUB-${rid}`; const getRoomQueueId = (rid: string) => `ROOM-${rid}`; const debouncedUpdate = (subscription: ISubscription) => { if (!subTimer) { subTimer = setTimeout(() => { const batch = queue; queue = {}; subTimer = null; Object.keys(batch).forEach(key => { InteractionManager.runAfterInteractions(() => { if (batch[key]) { if (/SUB/.test(key)) { const sub = batch[key]; const roomQueueId = getRoomQueueId(sub.rid); const room = batch[roomQueueId]; delete batch[roomQueueId]; createOrUpdateSubscription(sub, room); } else { const room = batch[key]; const subQueueId = getSubQueueId(room._id); const sub = batch[subQueueId]; delete batch[subQueueId]; createOrUpdateSubscription(sub, room); } } }); }); }, WINDOW_TIME); } queue[subscription.rid ? getSubQueueId(subscription.rid) : getRoomQueueId(subscription._id)] = subscription; }; export default function subscribeRooms() { const handleStreamMessageReceived = protectedFunction(async (ddpMessage: IDDPMessage) => { const db = database.active; // check if the server from variable is the same as the js sdk client if (sdk && sdk.current.client && sdk.current.client.host !== subServer) { return; } if (ddpMessage.msg === 'added') { return; } const [type, data] = ddpMessage.fields.args; const [, ev] = ddpMessage.fields.eventName.split('/'); if (/userData/.test(ev)) { const [{ diff }] = ddpMessage.fields.args; if (diff?.statusLivechat) { store.dispatch(setUser({ statusLivechat: diff.statusLivechat })); } if ((['settings.preferences.showMessageInMainThread'] as any) in diff) { store.dispatch(setUser({ showMessageInMainThread: diff['settings.preferences.showMessageInMainThread'] })); } } if (/subscriptions/.test(ev)) { if (type === 'removed') { try { const subCollection = db.get('subscriptions'); const sub = await subCollection.find(data.rid); // TODO - today the Relation type from watermelon just support one to one relations // @ts-ignore const messages = (await sub.messages.fetch()) as TMessageModel[]; // @ts-ignore const threads = (await sub.threads.fetch()) as TThreadModel[]; // @ts-ignore const threadMessages = (await sub.threadMessages.fetch()) as TThreadMessageModel[]; const messagesToDelete = messages?.map((m: TMessageModel) => m.prepareDestroyPermanently()); const threadsToDelete = threads?.map((m: TThreadModel) => m.prepareDestroyPermanently()); const threadMessagesToDelete = threadMessages?.map((m: TThreadMessageModel) => m.prepareDestroyPermanently()); await db.write(async () => { await db.batch(sub.prepareDestroyPermanently(), ...messagesToDelete, ...threadsToDelete, ...threadMessagesToDelete); }); const roomState = store.getState().room; // Delete and remove events come from this stream // Here we identify which one was triggered if (data.rid === roomState.rid && roomState.isDeleting) { store.dispatch(removedRoom()); } else { EventEmitter.emit('ROOM_REMOVED', { rid: data.rid }); } } catch (e) { log(e); } } else { debouncedUpdate(data); } } if (/rooms/.test(ev)) { if (type === 'updated' || type === 'inserted') { debouncedUpdate(data); } } if (/message/.test(ev)) { try { const [args] = ddpMessage.fields.args; const _id = random(17); const message = { // @ts-ignore u: { _id, username: 'rocket.cat', name: 'Rocket Cat' }, ...buildMessage(EJSON.fromJSONValue(args)) } as IMessage; await updateMessages({ rid: args.rid, update: [message] }); } catch (e) { log(e); } } if (/notification/.test(ev)) { const [notification] = ddpMessage.fields.args; try { const { payload: { rid, message, sender } } = notification; const room = await RocketChat.getRoom(rid); notification.title = RocketChat.getRoomTitle(room); notification.avatar = RocketChat.getRoomAvatar(room); // If it's from a encrypted room if (message?.t === E2E_MESSAGE_TYPE) { // Decrypt this message content const { msg } = await Encryption.decryptMessage({ ...message, rid }); // If it's a direct the content is the message decrypted if (room.t === 'd') { notification.text = msg; // If it's a private group we should add the sender name } else { notification.text = `${RocketChat.getSenderName(sender)}: ${msg}`; } } } catch (e) { log(e); } EventEmitter.emit(INAPP_NOTIFICATION_EMITTER, notification); } if (/uiInteraction/.test(ev)) { const { type: eventType, ...args } = type; handlePayloadUserInteraction(eventType, args); } if (/e2ekeyRequest/.test(ev)) { const [roomId, keyId] = ddpMessage.fields.args; try { await Encryption.provideRoomKeyToUser(keyId, roomId); } catch (e) { log(e); } } }); const stop = () => { if (streamListener) { streamListener.then(removeListener); streamListener = false; } queue = {}; if (subTimer) { clearTimeout(subTimer); subTimer = false; } }; streamListener = sdk.onStreamData('stream-notify-user', handleStreamMessageReceived); try { // set the server that started this task subServer = sdk.current.client.host; sdk.current.subscribeNotifyUser().catch((e: unknown) => console.log(e)); return { stop: () => stop() }; } catch (e) { log(e); return Promise.reject(); } }