signal-desktop/ts/jobs/helpers/sendReaction.ts
automated-signal 2585b4e821
Migrate message ids to UUIDv7
Co-authored-by: Fedor Indutny <79877362+indutny-signal@users.noreply.github.com>
2024-10-08 13:45:41 -07:00

468 lines
15 KiB
TypeScript

// Copyright 2021 Signal Messenger, LLC
// SPDX-License-Identifier: AGPL-3.0-only
import { isNumber } from 'lodash';
import * as Errors from '../../types/errors';
import { strictAssert } from '../../util/assert';
import { repeat, zipObject } from '../../util/iterables';
import type { CallbackResultType } from '../../textsecure/Types.d';
import type { MessageModel } from '../../models/messages';
import type { MessageReactionType } from '../../model-types.d';
import type { ConversationModel } from '../../models/conversations';
import { DataWriter } from '../../sql/Client';
import * as reactionUtil from '../../reactions/util';
import { isSent, SendStatus } from '../../messages/MessageSendState';
import { __DEPRECATED$getMessageById } from '../../messages/getMessageById';
import { isIncoming } from '../../messages/helpers';
import {
isMe,
isDirectConversation,
isGroupV2,
} from '../../util/whatTypeOfConversation';
import { getSendOptions } from '../../util/getSendOptions';
import { SignalService as Proto } from '../../protobuf';
import { handleMessageSend } from '../../util/handleMessageSend';
import { ourProfileKeyService } from '../../services/ourProfileKey';
import { canReact, isStory } from '../../state/selectors/message';
import { findAndFormatContact } from '../../util/findAndFormatContact';
import type { AciString, ServiceIdString } from '../../types/ServiceId';
import { isAciString } from '../../util/isAciString';
import { handleMultipleSendErrors } from './handleMultipleSendErrors';
import { incrementMessageCounter } from '../../util/incrementMessageCounter';
import { generateMessageId } from '../../util/generateMessageId';
import type {
ConversationQueueJobBundle,
ReactionJobData,
} from '../conversationJobQueue';
import { isConversationAccepted } from '../../util/isConversationAccepted';
import { isConversationUnregistered } from '../../util/isConversationUnregistered';
import type { LoggerType } from '../../types/Logging';
import { sendToGroup } from '../../util/sendToGroup';
export async function sendReaction(
conversation: ConversationModel,
{
isFinalAttempt,
messaging,
shouldContinue,
timeRemaining,
log,
}: ConversationQueueJobBundle,
data: ReactionJobData
): Promise<void> {
const { messageId, revision } = data;
const ourAci = window.textsecure.storage.user.getCheckedAci();
await window.ConversationController.load();
const ourConversationId =
window.ConversationController.getOurConversationIdOrThrow();
const message = await __DEPRECATED$getMessageById(messageId);
if (!message) {
log.info(
`message ${messageId} was not found, maybe because it was deleted. Giving up on sending its reactions`
);
return;
}
strictAssert(
!isStory(message.attributes),
'Story reactions should be handled by sendStoryReaction'
);
const { pendingReaction, emojiToRemove } =
reactionUtil.getNewestPendingOutgoingReaction(
getReactions(message),
ourConversationId
);
if (!pendingReaction) {
log.info(`no pending reaction for ${messageId}. Doing nothing`);
return;
}
if (!canReact(message.attributes, ourConversationId, findAndFormatContact)) {
log.info(`could not react to ${messageId}. Removing this pending reaction`);
markReactionFailed(message, pendingReaction);
await DataWriter.saveMessage(message.attributes, { ourAci });
return;
}
if (!shouldContinue) {
log.info(
`reacting to message ${messageId} ran out of time. Giving up on sending it`
);
markReactionFailed(message, pendingReaction);
await DataWriter.saveMessage(message.attributes, { ourAci });
return;
}
let sendErrors: Array<Error> = [];
const saveErrors = (errors: Array<Error>): void => {
sendErrors = errors;
};
let originalError: Error | undefined;
try {
const messageConversation = message.getConversation();
if (messageConversation !== conversation) {
log.error(
`message conversation '${messageConversation?.idForLogging()}' does not match job conversation ${conversation.idForLogging()}`
);
return;
}
const expireTimer = messageConversation.get('expireTimer');
const {
allRecipientServiceIds,
recipientServiceIdsWithoutMe,
untrustedServiceIds,
} = getRecipients(log, pendingReaction, conversation);
if (untrustedServiceIds.length) {
window.reduxActions.conversations.conversationStoppedByMissingVerification(
{
conversationId: conversation.id,
untrustedServiceIds,
}
);
throw new Error(
`Reaction for message ${messageId} sending blocked because ${untrustedServiceIds.length} conversation(s) were untrusted. Failing this attempt.`
);
}
const profileKey = conversation.get('profileSharing')
? await ourProfileKeyService.get()
: undefined;
const { emoji, ...restOfPendingReaction } = pendingReaction;
let targetAuthorAci: AciString;
if (isIncoming(message.attributes)) {
strictAssert(
isAciString(message.attributes.sourceServiceId),
'incoming message does not have sender ACI'
);
({ sourceServiceId: targetAuthorAci } = message.attributes);
} else {
targetAuthorAci = ourAci;
}
const reactionForSend = {
...restOfPendingReaction,
emoji: emoji || emojiToRemove,
targetAuthorAci,
remove: !emoji,
};
const ephemeralMessageForReactionSend = new window.Whisper.Message({
...generateMessageId(incrementMessageCounter()),
type: 'outgoing',
conversationId: conversation.get('id'),
sent_at: pendingReaction.timestamp,
received_at_ms: pendingReaction.timestamp,
timestamp: pendingReaction.timestamp,
sendStateByConversationId: zipObject(
Object.keys(pendingReaction.isSentByConversationId || {}),
repeat({
status: SendStatus.Pending,
updatedAt: Date.now(),
})
),
});
// Adds the reaction's attributes to the message cache so that we can
// safely `set` on it later.
window.MessageCache.toMessageAttributes(
ephemeralMessageForReactionSend.attributes
);
ephemeralMessageForReactionSend.doNotSave = true;
let didFullySend: boolean;
const successfulConversationIds = new Set<string>();
if (recipientServiceIdsWithoutMe.length === 0) {
log.info('sending sync reaction message only');
const dataMessage = await messaging.getDataOrEditMessage({
attachments: [],
expireTimer,
expireTimerVersion: conversation.getExpireTimerVersion(),
groupV2: conversation.getGroupV2Info({
members: recipientServiceIdsWithoutMe,
}),
preview: [],
profileKey,
reaction: reactionForSend,
recipients: allRecipientServiceIds,
timestamp: pendingReaction.timestamp,
});
await ephemeralMessageForReactionSend.sendSyncMessageOnly({
dataMessage,
saveErrors,
targetTimestamp: pendingReaction.timestamp,
});
didFullySend = true;
successfulConversationIds.add(ourConversationId);
} else {
const sendOptions = await getSendOptions(conversation.attributes);
const { ContentHint } = Proto.UnidentifiedSenderMessage.Message;
let promise: Promise<CallbackResultType>;
if (isDirectConversation(conversation.attributes)) {
if (!isConversationAccepted(conversation.attributes)) {
log.info(
`conversation ${conversation.idForLogging()} is not accepted; refusing to send`
);
markReactionFailed(message, pendingReaction);
return;
}
if (isConversationUnregistered(conversation.attributes)) {
log.info(
`conversation ${conversation.idForLogging()} is unregistered; refusing to send`
);
markReactionFailed(message, pendingReaction);
return;
}
if (conversation.isBlocked()) {
log.info(
`conversation ${conversation.idForLogging()} is blocked; refusing to send`
);
markReactionFailed(message, pendingReaction);
return;
}
log.info('sending direct reaction message');
promise = messaging.sendMessageToServiceId({
serviceId: recipientServiceIdsWithoutMe[0],
messageText: undefined,
attachments: [],
quote: undefined,
preview: [],
sticker: undefined,
reaction: reactionForSend,
deletedForEveryoneTimestamp: undefined,
timestamp: pendingReaction.timestamp,
expireTimer,
expireTimerVersion: conversation.getExpireTimerVersion(),
contentHint: ContentHint.RESENDABLE,
groupId: undefined,
profileKey,
options: sendOptions,
urgent: true,
includePniSignatureMessage: true,
});
} else {
log.info('sending group reaction message');
promise = conversation.queueJob(
'conversationQueue/sendReaction',
abortSignal => {
// Note: this will happen for all old jobs queued before 5.32.x
if (isGroupV2(conversation.attributes) && !isNumber(revision)) {
log.error('No revision provided, but conversation is GroupV2');
}
const groupV2Info = conversation.getGroupV2Info({
members: recipientServiceIdsWithoutMe,
});
if (groupV2Info && isNumber(revision)) {
groupV2Info.revision = revision;
}
return sendToGroup({
abortSignal,
contentHint: ContentHint.RESENDABLE,
groupSendOptions: {
groupV2: groupV2Info,
reaction: reactionForSend,
timestamp: pendingReaction.timestamp,
expireTimer,
profileKey,
},
messageId,
sendOptions,
sendTarget: conversation.toSenderKeyTarget(),
sendType: 'reaction',
urgent: true,
});
}
);
}
await ephemeralMessageForReactionSend.send({
promise: handleMessageSend(promise, {
messageIds: [messageId],
sendType: 'reaction',
}),
saveErrors,
targetTimestamp: pendingReaction.timestamp,
});
// Because message.send swallows and processes errors, we'll await the inner promise
// to get the SendMessageProtoError, which gives us information upstream
/// processors need to detect certain kinds of errors.
try {
await promise;
} catch (error) {
if (error instanceof Error) {
originalError = error;
} else {
log.error(
`promise threw something other than an error: ${Errors.toLogFormat(
error
)}`
);
}
}
didFullySend = true;
const reactionSendStateByConversationId =
ephemeralMessageForReactionSend.get('sendStateByConversationId') || {};
for (const [conversationId, sendState] of Object.entries(
reactionSendStateByConversationId
)) {
if (isSent(sendState.status)) {
successfulConversationIds.add(conversationId);
} else {
didFullySend = false;
}
}
if (!ephemeralMessageForReactionSend.doNotSave) {
const reactionMessage = ephemeralMessageForReactionSend;
await reactionMessage.hydrateStoryContext(message.attributes, {
shouldSave: false,
});
await DataWriter.saveMessage(reactionMessage.attributes, {
ourAci,
forceSave: true,
});
window.MessageCache.__DEPRECATED$register(
reactionMessage.id,
reactionMessage,
'sendReaction'
);
void conversation.addSingleMessage(reactionMessage.attributes);
}
}
const newReactions = reactionUtil.markOutgoingReactionSent(
getReactions(message),
pendingReaction,
successfulConversationIds
);
setReactions(message, newReactions);
if (!didFullySend) {
throw new Error('reaction did not fully send');
}
} catch (thrownError: unknown) {
await handleMultipleSendErrors({
errors: [thrownError, ...sendErrors],
isFinalAttempt,
log,
markFailed: () => markReactionFailed(message, pendingReaction),
timeRemaining,
// In the case of a failed group send thrownError will not be SentMessageProtoError,
// but we should have been able to harvest the original error. In the Note to Self
// send case, thrownError will be the error we care about, and we won't have an
// originalError.
toThrow: originalError || thrownError,
});
} finally {
await DataWriter.saveMessage(message.attributes, { ourAci });
}
}
const getReactions = (
message: MessageModel
): ReadonlyArray<MessageReactionType> => message.get('reactions') || [];
const setReactions = (
message: MessageModel,
reactions: Array<MessageReactionType>
): void => {
if (reactions.length) {
message.set('reactions', reactions);
} else {
message.set('reactions', undefined);
}
};
function getRecipients(
log: LoggerType,
reaction: Readonly<MessageReactionType>,
conversation: ConversationModel
): {
allRecipientServiceIds: Array<ServiceIdString>;
recipientServiceIdsWithoutMe: Array<ServiceIdString>;
untrustedServiceIds: Array<ServiceIdString>;
} {
const allRecipientServiceIds: Array<ServiceIdString> = [];
const recipientServiceIdsWithoutMe: Array<ServiceIdString> = [];
const untrustedServiceIds: Array<ServiceIdString> = [];
const currentConversationRecipients = conversation.getMemberConversationIds();
for (const id of reactionUtil.getUnsentConversationIds(reaction)) {
const recipient = window.ConversationController.get(id);
if (!recipient) {
continue;
}
const recipientIdentifier = recipient.getSendTarget();
const isRecipientMe = isMe(recipient.attributes);
if (
!recipientIdentifier ||
(!currentConversationRecipients.has(id) && !isRecipientMe)
) {
continue;
}
if (recipient.isUntrusted()) {
const serviceId = recipient.getServiceId();
if (!serviceId) {
log.error(
`sendReaction/getRecipients: Untrusted conversation ${recipient.idForLogging()} missing serviceId.`
);
continue;
}
untrustedServiceIds.push(serviceId);
continue;
}
if (recipient.isUnregistered()) {
continue;
}
if (recipient.isBlocked()) {
continue;
}
allRecipientServiceIds.push(recipientIdentifier);
if (!isRecipientMe) {
recipientServiceIdsWithoutMe.push(recipientIdentifier);
}
}
return {
allRecipientServiceIds,
recipientServiceIdsWithoutMe,
untrustedServiceIds,
};
}
function markReactionFailed(
message: MessageModel,
pendingReaction: MessageReactionType
): void {
const newReactions = reactionUtil.markOutgoingReactionFailed(
getReactions(message),
pendingReaction
);
setReactions(message, newReactions);
}