422 lines
12 KiB
JavaScript
422 lines
12 KiB
JavaScript
import { sanitizedRaw } from '@nozbe/watermelondb/RawRecord';
|
|
import { InteractionManager } from 'react-native';
|
|
|
|
import database from '../../database';
|
|
import { merge } from '../helpers/mergeSubscriptionsRooms';
|
|
import protectedFunction from '../helpers/protectedFunction';
|
|
import messagesStatus from '../../../constants/messagesStatus';
|
|
import log from '../../../utils/log';
|
|
import random from '../../../utils/random';
|
|
import store from '../../createStore';
|
|
import { roomsRequest } from '../../../actions/rooms';
|
|
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';
|
|
|
|
const removeListener = listener => listener.stop();
|
|
|
|
let connectedListener;
|
|
let disconnectedListener;
|
|
let streamListener;
|
|
let subServer;
|
|
let queue = {};
|
|
let subTimer = null;
|
|
const WINDOW_TIME = 500;
|
|
|
|
const createOrUpdateSubscription = async(subscription, room) => {
|
|
try {
|
|
const db = database.active;
|
|
const subCollection = db.collections.get('subscriptions');
|
|
const roomsCollection = db.collections.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,
|
|
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
|
|
};
|
|
} catch (error) {
|
|
try {
|
|
await db.action(async() => {
|
|
await roomsCollection.create(protectedFunction((r) => {
|
|
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
|
|
};
|
|
} catch (error) {
|
|
// Do nothing
|
|
}
|
|
}
|
|
|
|
let tmp = merge(subscription, room);
|
|
tmp = await Encryption.decryptSubscription(tmp);
|
|
let sub;
|
|
try {
|
|
sub = await subCollection.find(tmp.rid);
|
|
} catch (error) {
|
|
// Do nothing
|
|
}
|
|
|
|
// 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);
|
|
// Decrypt all pending messages of this room in parallel
|
|
Encryption.decryptPendingMessages(tmp.rid);
|
|
}
|
|
|
|
const batch = [];
|
|
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.collections.get('messages');
|
|
let messageRecord;
|
|
try {
|
|
messageRecord = await messagesCollection.find(lastMessage._id);
|
|
} catch (error) {
|
|
// Do nothing
|
|
}
|
|
|
|
if (messageRecord) {
|
|
batch.push(
|
|
messageRecord.prepareUpdate(() => {
|
|
Object.assign(messageRecord, lastMessage);
|
|
})
|
|
);
|
|
} else {
|
|
batch.push(
|
|
messagesCollection.prepareCreate((m) => {
|
|
m._raw = sanitizedRaw({ id: lastMessage._id }, messagesCollection.schema);
|
|
m.subscription.id = lastMessage.rid;
|
|
return Object.assign(m, lastMessage);
|
|
})
|
|
);
|
|
}
|
|
}
|
|
|
|
await db.action(async() => {
|
|
await db.batch(...batch);
|
|
});
|
|
} catch (e) {
|
|
log(e);
|
|
}
|
|
};
|
|
|
|
const getSubQueueId = rid => `SUB-${ rid }`;
|
|
|
|
const getRoomQueueId = rid => `ROOM-${ rid }`;
|
|
|
|
const debouncedUpdate = (subscription) => {
|
|
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 handleConnection = () => {
|
|
store.dispatch(roomsRequest());
|
|
};
|
|
|
|
const handleStreamMessageReceived = protectedFunction(async(ddpMessage) => {
|
|
const db = database.active;
|
|
|
|
// check if the server from variable is the same as the js sdk client
|
|
if (this.sdk && this.sdk.client && this.sdk.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 (/subscriptions/.test(ev)) {
|
|
if (type === 'removed') {
|
|
try {
|
|
const subCollection = db.collections.get('subscriptions');
|
|
const sub = await subCollection.find(data.rid);
|
|
const messages = await sub.messages.fetch();
|
|
const threads = await sub.threads.fetch();
|
|
const threadMessages = await sub.threadMessages.fetch();
|
|
const messagesToDelete = messages.map(m => m.prepareDestroyPermanently());
|
|
const threadsToDelete = threads.map(m => m.prepareDestroyPermanently());
|
|
const threadMessagesToDelete = threadMessages.map(m => m.prepareDestroyPermanently());
|
|
await db.action(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)) {
|
|
const [args] = ddpMessage.fields.args;
|
|
const _id = random(17);
|
|
const message = {
|
|
_id,
|
|
rid: args.rid,
|
|
msg: args.msg,
|
|
blocks: args.blocks,
|
|
ts: new Date(),
|
|
_updatedAt: new Date(),
|
|
status: messagesStatus.SENT,
|
|
u: {
|
|
_id,
|
|
username: 'rocket.cat'
|
|
}
|
|
};
|
|
try {
|
|
const msgCollection = db.collections.get('messages');
|
|
await db.action(async() => {
|
|
await msgCollection.create(protectedFunction((m) => {
|
|
m._raw = sanitizedRaw({ id: message._id }, msgCollection.schema);
|
|
m.subscription.id = args.rid;
|
|
Object.assign(m, 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 (connectedListener) {
|
|
connectedListener.then(removeListener);
|
|
connectedListener = false;
|
|
}
|
|
if (disconnectedListener) {
|
|
disconnectedListener.then(removeListener);
|
|
disconnectedListener = false;
|
|
}
|
|
if (streamListener) {
|
|
streamListener.then(removeListener);
|
|
streamListener = false;
|
|
}
|
|
queue = {};
|
|
if (subTimer) {
|
|
clearTimeout(subTimer);
|
|
subTimer = false;
|
|
}
|
|
};
|
|
|
|
connectedListener = this.sdk.onStreamData('connected', handleConnection);
|
|
disconnectedListener = this.sdk.onStreamData('close', handleConnection);
|
|
streamListener = this.sdk.onStreamData('stream-notify-user', handleStreamMessageReceived);
|
|
|
|
try {
|
|
// set the server that started this task
|
|
subServer = this.sdk.client.host;
|
|
this.sdk.subscribeNotifyUser().catch(e => console.log(e));
|
|
|
|
return {
|
|
stop: () => stop()
|
|
};
|
|
} catch (e) {
|
|
log(e);
|
|
return Promise.reject();
|
|
}
|
|
}
|