cleanup creation of opengroup message on sync
Audric Ackermann
3 years ago
parent
5e314e4dcc
commit
5afbd9c19e
@ -0,0 +1,159 @@
|
||||
import { UserUtils } from '../session/utils';
|
||||
import { MessageModel } from './message';
|
||||
import { MessageAttributesOptionals, MessageModelType } from './messageType';
|
||||
|
||||
export type MessageCreationData = {
|
||||
timestamp: number;
|
||||
receivedAt: number;
|
||||
source: string;
|
||||
isPublic: boolean;
|
||||
serverId: number | null;
|
||||
serverTimestamp: number | null;
|
||||
groupId: string | null;
|
||||
|
||||
expirationStartTimestamp?: number;
|
||||
destination: string;
|
||||
messageHash: string;
|
||||
};
|
||||
|
||||
function initIncomingMessage(data: MessageCreationData): MessageModel {
|
||||
const {
|
||||
timestamp,
|
||||
isPublic,
|
||||
receivedAt,
|
||||
source,
|
||||
serverId,
|
||||
serverTimestamp,
|
||||
messageHash,
|
||||
groupId,
|
||||
} = data;
|
||||
|
||||
const messageData: MessageAttributesOptionals = {
|
||||
source,
|
||||
serverId: serverId || undefined,
|
||||
sent_at: timestamp,
|
||||
serverTimestamp: serverTimestamp || undefined,
|
||||
received_at: receivedAt || Date.now(),
|
||||
conversationId: groupId ?? source,
|
||||
type: 'incoming',
|
||||
direction: 'incoming',
|
||||
unread: 1,
|
||||
isPublic,
|
||||
messageHash: messageHash || undefined,
|
||||
};
|
||||
|
||||
return new MessageModel(messageData);
|
||||
}
|
||||
|
||||
/**
|
||||
* This function can be called for either a sync message or a message synced through an opengroup poll.
|
||||
* This does not save it to the db, just in memory
|
||||
*/
|
||||
function createMessageSentFromOurself({
|
||||
timestamp,
|
||||
serverTimestamp,
|
||||
serverId,
|
||||
isPublic,
|
||||
receivedAt,
|
||||
expirationStartTimestamp,
|
||||
destination,
|
||||
groupId,
|
||||
messageHash,
|
||||
}: {
|
||||
timestamp: number;
|
||||
receivedAt: number;
|
||||
isPublic: boolean;
|
||||
serverId: number | null;
|
||||
serverTimestamp: number | null;
|
||||
groupId: string | null;
|
||||
expirationStartTimestamp: number | null;
|
||||
destination: string;
|
||||
messageHash: string;
|
||||
}): MessageModel {
|
||||
// Omit<
|
||||
// MessageAttributesOptionals,
|
||||
// 'conversationId' | 'source' | 'type' | 'direction' | 'received_at'
|
||||
// >
|
||||
const now = Date.now();
|
||||
|
||||
const messageData: MessageAttributesOptionals = {
|
||||
source: UserUtils.getOurPubKeyStrFromCache(),
|
||||
type: 'outgoing' as MessageModelType,
|
||||
serverTimestamp: serverTimestamp || undefined,
|
||||
serverId: serverId || undefined,
|
||||
sent_at: timestamp,
|
||||
received_at: isPublic ? receivedAt : now,
|
||||
isPublic,
|
||||
conversationId: groupId ?? destination,
|
||||
messageHash,
|
||||
unread: 0,
|
||||
sent_to: [],
|
||||
sent: true,
|
||||
expirationStartTimestamp: Math.min(expirationStartTimestamp || data.timestamp || now, now),
|
||||
};
|
||||
|
||||
return new MessageModel(messageData);
|
||||
}
|
||||
|
||||
/**
|
||||
* This function is only called when we get a message from ourself from an opengroup polling event
|
||||
*/
|
||||
export function createPublicMessageSentFromUs({
|
||||
serverTimestamp,
|
||||
serverId,
|
||||
conversationId,
|
||||
}: {
|
||||
serverId: number;
|
||||
serverTimestamp: number;
|
||||
conversationId: string;
|
||||
}): MessageModel {
|
||||
const messageData: MessageAttributesOptionals = {
|
||||
source: UserUtils.getOurPubKeyStrFromCache(),
|
||||
type: 'outgoing' as MessageModelType,
|
||||
serverTimestamp: serverTimestamp || undefined,
|
||||
serverId: serverId || undefined,
|
||||
sent_at: serverTimestamp,
|
||||
received_at: serverTimestamp,
|
||||
isPublic: true,
|
||||
conversationId,
|
||||
messageHash: '', // we do not care of a messageHash for an opengroup message. we have serverId for that
|
||||
unread: 0,
|
||||
sent_to: [],
|
||||
sent: true,
|
||||
expirationStartTimestamp: undefined,
|
||||
};
|
||||
|
||||
return new MessageModel(messageData);
|
||||
}
|
||||
|
||||
/**
|
||||
* This function is only called by the Receiver when we get a message
|
||||
* from someone else than ourself from an opengroup polling event
|
||||
*/
|
||||
export function createPublicMessageSentFromNotUs({
|
||||
serverTimestamp,
|
||||
serverId,
|
||||
conversationId,
|
||||
sender,
|
||||
}: {
|
||||
serverId: number;
|
||||
sender: string;
|
||||
serverTimestamp: number;
|
||||
conversationId: string;
|
||||
}): MessageModel {
|
||||
const messageData: MessageAttributesOptionals = {
|
||||
source: sender,
|
||||
conversationId,
|
||||
type: 'incoming' as MessageModelType,
|
||||
serverTimestamp: serverTimestamp,
|
||||
sent_at: serverTimestamp,
|
||||
received_at: serverTimestamp,
|
||||
serverId,
|
||||
isPublic: true,
|
||||
messageHash: '', // we do not care of a messageHash for an opengroup message. we have serverId for that
|
||||
unread: 1,
|
||||
expirationStartTimestamp: undefined,
|
||||
};
|
||||
|
||||
return new MessageModel(messageData);
|
||||
}
|
||||
@ -1,53 +0,0 @@
|
||||
import { initIncomingMessage } from './dataMessage';
|
||||
import { toNumber } from 'lodash';
|
||||
import { getConversationController } from '../session/conversations';
|
||||
import { ConversationTypeEnum } from '../models/conversation';
|
||||
import { toLogFormat } from '../types/attachments/Errors';
|
||||
import { messagesAdded } from '../state/ducks/conversations';
|
||||
|
||||
export async function onError(ev: any) {
|
||||
const { error } = ev;
|
||||
window?.log?.error('background onError:', toLogFormat(error));
|
||||
|
||||
if (ev.proto) {
|
||||
const envelope = ev.proto;
|
||||
|
||||
const message = initIncomingMessage(envelope);
|
||||
|
||||
await message.saveErrors(error || new Error('Error was null'));
|
||||
const id = message.get('conversationId');
|
||||
const conversation = await getConversationController().getOrCreateAndWait(
|
||||
id,
|
||||
ConversationTypeEnum.PRIVATE
|
||||
);
|
||||
// force conversation unread count to be > 0 so it is highlighted
|
||||
conversation.set({
|
||||
active_at: Date.now(),
|
||||
unreadCount: toNumber(conversation.get('unreadCount')) + 1,
|
||||
});
|
||||
|
||||
const conversationActiveAt = conversation.get('active_at');
|
||||
const messageTimestamp = message.get('timestamp') || 0;
|
||||
if (!conversationActiveAt || messageTimestamp > conversationActiveAt) {
|
||||
conversation.set({ active_at: message.get('sent_at') });
|
||||
}
|
||||
|
||||
conversation.updateLastMessage();
|
||||
await conversation.notify(message);
|
||||
window.inboxStore?.dispatch(
|
||||
messagesAdded([
|
||||
{
|
||||
conversationKey: conversation.id,
|
||||
messageModelProps: message.getMessageModelProps(),
|
||||
},
|
||||
])
|
||||
);
|
||||
|
||||
if (ev.confirm) {
|
||||
ev.confirm();
|
||||
}
|
||||
await conversation.commit();
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
@ -0,0 +1,88 @@
|
||||
import { noop } from 'lodash';
|
||||
import { ConversationTypeEnum } from '../models/conversation';
|
||||
import {
|
||||
createPublicMessageSentFromNotUs,
|
||||
createPublicMessageSentFromUs,
|
||||
} from '../models/messageFactory';
|
||||
import { SignalService } from '../protobuf';
|
||||
import { OpenGroupRequestCommonType } from '../session/apis/open_group_api/opengroupV2/ApiUtil';
|
||||
import { OpenGroupMessageV2 } from '../session/apis/open_group_api/opengroupV2/OpenGroupMessageV2';
|
||||
import { getOpenGroupV2ConversationId } from '../session/apis/open_group_api/utils/OpenGroupUtils';
|
||||
import { getConversationController } from '../session/conversations';
|
||||
import { removeMessagePadding } from '../session/crypto/BufferPadding';
|
||||
import { UserUtils } from '../session/utils';
|
||||
import { fromBase64ToArray } from '../session/utils/String';
|
||||
import { isOpengroupMessageDuplicate } from './dataMessage';
|
||||
import { handleMessageJob } from './queuedJob';
|
||||
|
||||
export async function handleOpenGroupV2Message(
|
||||
message: OpenGroupMessageV2,
|
||||
roomInfos: OpenGroupRequestCommonType
|
||||
) {
|
||||
const { base64EncodedData, sentTimestamp, sender, serverId } = message;
|
||||
const { serverUrl, roomId } = roomInfos;
|
||||
if (!base64EncodedData || !sentTimestamp || !sender || !serverId) {
|
||||
window?.log?.warn('Invalid data passed to handleOpenGroupV2Message.', message);
|
||||
return;
|
||||
}
|
||||
|
||||
// Note: opengroup messages are not padded
|
||||
const dataUint = new Uint8Array(removeMessagePadding(fromBase64ToArray(base64EncodedData)));
|
||||
|
||||
const decoded = SignalService.Content.decode(dataUint);
|
||||
|
||||
const conversationId = getOpenGroupV2ConversationId(serverUrl, roomId);
|
||||
if (!conversationId) {
|
||||
window?.log?.error('We cannot handle a message without a conversationId');
|
||||
return;
|
||||
}
|
||||
const idataMessage = decoded?.dataMessage;
|
||||
if (!idataMessage) {
|
||||
window?.log?.error('Invalid decoded opengroup message: no dataMessage');
|
||||
return;
|
||||
}
|
||||
|
||||
if (!getConversationController().get(conversationId)) {
|
||||
window?.log?.error('Received a message for an unknown convo. Skipping');
|
||||
return;
|
||||
}
|
||||
|
||||
const conversation = await getConversationController().getOrCreateAndWait(
|
||||
conversationId,
|
||||
ConversationTypeEnum.GROUP
|
||||
);
|
||||
|
||||
if (!conversation) {
|
||||
window?.log?.warn('Skipping handleJob for unknown convo: ', conversationId);
|
||||
return;
|
||||
}
|
||||
|
||||
void conversation.queueJob(async () => {
|
||||
const isMe = UserUtils.isUsFromCache(sender);
|
||||
|
||||
const commonAttributes = { serverTimestamp: sentTimestamp, serverId, conversationId };
|
||||
const attributesForNotUs = { ...commonAttributes, sender };
|
||||
// those lines just create an empty message with some basic stuff set.
|
||||
// the whole decoding of data is happening in handleMessageJob()
|
||||
const msgModel = isMe
|
||||
? createPublicMessageSentFromUs(commonAttributes)
|
||||
: createPublicMessageSentFromNotUs(attributesForNotUs);
|
||||
|
||||
// WARNING this is important that the isOpengroupMessageDuplicate is made INSIDE the conversation.queueJob call
|
||||
const isDuplicate = await isOpengroupMessageDuplicate(attributesForNotUs);
|
||||
|
||||
if (isDuplicate) {
|
||||
window?.log?.info('Received duplicate opengroup message. Dropping it.');
|
||||
return;
|
||||
}
|
||||
|
||||
await handleMessageJob(
|
||||
msgModel,
|
||||
conversation,
|
||||
decoded?.dataMessage as SignalService.DataMessage,
|
||||
noop,
|
||||
sender,
|
||||
''
|
||||
);
|
||||
});
|
||||
}
|
||||
Loading…
Reference in New Issue