Support for message retry requests

ts/util/index.ts

@@ -36,6 +36,7 @@ import * as zkgroup from './zkgroup';
 import { StartupQueue } from './StartupQueue';
 import { postLinkExperience } from './postLinkExperience';
 import { sendToGroup, sendContentMessageToGroup } from './sendToGroup';
+import { RetryPlaceholders } from './retryPlaceholders';
 
 export {
   GoogleChrome,
@@ -62,6 +63,7 @@ export {
   parseRemoteClientExpiration,
   postLinkExperience,
   queueUpdateMessage,
+  RetryPlaceholders,
   saveNewMessageBatcher,
   sendContentMessageToGroup,
   sendToGroup,

ts/util/retryPlaceholders.ts

@@ -0,0 +1,196 @@
+// Copyright 2021 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+
+import { z } from 'zod';
+import { groupBy } from 'lodash';
+
+const retryItemSchema = z
+  .object({
+    conversationId: z.string(),
+    sentAt: z.number(),
+    receivedAt: z.number(),
+    receivedAtCounter: z.number(),
+    senderUuid: z.string(),
+  })
+  .passthrough();
+export type RetryItemType = z.infer<typeof retryItemSchema>;
+
+const retryItemListSchema = z.array(retryItemSchema);
+export type RetryItemListType = z.infer<typeof retryItemListSchema>;
+
+export type ByConversationLookupType = {
+  [key: string]: Array<RetryItemType>;
+};
+export type ByMessageLookupType = Map<string, RetryItemType>;
+
+export function getItemId(conversationId: string, sentAt: number): string {
+  return `${conversationId}--${sentAt}`;
+}
+
+const HOUR = 60 * 60 * 1000;
+export const STORAGE_KEY = 'retryPlaceholders';
+
+export function getOneHourAgo(): number {
+  return Date.now() - HOUR;
+}
+
+export class RetryPlaceholders {
+  private items: Array<RetryItemType>;
+
+  private byConversation: ByConversationLookupType;
+
+  private byMessage: ByMessageLookupType;
+
+  constructor() {
+    if (!window.storage) {
+      throw new Error(
+        'RetryPlaceholders.constructor: window.storage not available!'
+      );
+    }
+
+    const parsed = retryItemListSchema.safeParse(
+      window.storage.get(STORAGE_KEY) || []
+    );
+    if (!parsed.success) {
+      window.log.warn(
+        `RetryPlaceholders.constructor: Data fetched from storage did not match schema: ${JSON.stringify(
+          parsed.error.flatten()
+        )}`
+      );
+    }
+
+    this.items = parsed.success ? parsed.data : [];
+    window.log.info(
+      `RetryPlaceholders.constructor: Started with ${this.items.length} items`
+    );
+
+    this.sortByExpiresAtAsc();
+    this.byConversation = this.makeByConversationLookup();
+    this.byMessage = this.makeByMessageLookup();
+  }
+
+  // Arranging local data for efficiency
+
+  sortByExpiresAtAsc(): void {
+    this.items.sort(
+      (left: RetryItemType, right: RetryItemType) =>
+        left.receivedAt - right.receivedAt
+    );
+  }
+
+  makeByConversationLookup(): ByConversationLookupType {
+    return groupBy(this.items, item => item.conversationId);
+  }
+
+  makeByMessageLookup(): ByMessageLookupType {
+    const lookup = new Map<string, RetryItemType>();
+    this.items.forEach(item => {
+      lookup.set(getItemId(item.conversationId, item.sentAt), item);
+    });
+    return lookup;
+  }
+
+  makeLookups(): void {
+    this.byConversation = this.makeByConversationLookup();
+    this.byMessage = this.makeByMessageLookup();
+  }
+
+  // Basic data management
+
+  async add(item: RetryItemType): Promise<void> {
+    const parsed = retryItemSchema.safeParse(item);
+    if (!parsed.success) {
+      throw new Error(
+        `RetryPlaceholders.add: Item did not match schema ${JSON.stringify(
+          parsed.error.flatten()
+        )}`
+      );
+    }
+
+    this.items.push(item);
+    this.sortByExpiresAtAsc();
+    this.makeLookups();
+    await this.save();
+  }
+
+  async save(): Promise<void> {
+    await window.storage.put(STORAGE_KEY, this.items);
+  }
+
+  // Finding items in different ways
+
+  getCount(): number {
+    return this.items.length;
+  }
+
+  getNextToExpire(): RetryItemType | undefined {
+    return this.items[0];
+  }
+
+  async getExpiredAndRemove(): Promise<Array<RetryItemType>> {
+    const expiration = getOneHourAgo();
+    const max = this.items.length;
+    const result: Array<RetryItemType> = [];
+
+    for (let i = 0; i < max; i += 1) {
+      const item = this.items[i];
+      if (item.receivedAt <= expiration) {
+        result.push(item);
+      } else {
+        break;
+      }
+    }
+
+    window.log.info(
+      `RetryPlaceholders.getExpiredAndRemove: Found ${result.length} expired items`
+    );
+
+    this.items.splice(0, result.length);
+    this.makeLookups();
+    await this.save();
+
+    return result;
+  }
+
+  async findByConversationAndRemove(
+    conversationId: string
+  ): Promise<Array<RetryItemType>> {
+    const result = this.byConversation[conversationId];
+    if (!result) {
+      return [];
+    }
+
+    const items = this.items.filter(
+      item => item.conversationId !== conversationId
+    );
+
+    window.log.info(
+      `RetryPlaceholders.findByConversationAndRemove: Found ${result.length} expired items`
+    );
+
+    this.items = items;
+    this.sortByExpiresAtAsc();
+    this.makeLookups();
+    await this.save();
+
+    return result;
+  }
+
+  async findByMessageAndRemove(
+    conversationId: string,
+    sentAt: number
+  ): Promise<RetryItemType | undefined> {
+    const result = this.byMessage.get(getItemId(conversationId, sentAt));
+    if (!result) {
+      return undefined;
+    }
+
+    const index = this.items.findIndex(item => item === result);
+
+    this.items.splice(index, 1);
+    this.makeLookups();
+    await this.save();
+
+    return result;
+  }
+}

ts/util/sendToGroup.ts

@@ -55,6 +55,7 @@ const MAX_RECURSION = 5;
 export async function sendToGroup(
   groupSendOptions: GroupSendOptionsType,
   conversation: ConversationModel,
+  contentHint: number,
   sendOptions?: SendOptionsType,
   isPartialSend?: boolean
 ): Promise<CallbackResultType> {
@@ -75,6 +76,7 @@ export async function sendToGroup(
   );
 
   return sendContentMessageToGroup({
+    contentHint,
     contentMessage,
     conversation,
     isPartialSend,
@@ -85,6 +87,7 @@ export async function sendToGroup(
 }
 
 export async function sendContentMessageToGroup({
+  contentHint,
   contentMessage,
   conversation,
   isPartialSend,
@@ -93,6 +96,7 @@ export async function sendContentMessageToGroup({
   sendOptions,
   timestamp,
 }: {
+  contentHint: number;
   contentMessage: ContentClass;
   conversation: ConversationModel;
   isPartialSend?: boolean;
@@ -110,6 +114,7 @@ export async function sendContentMessageToGroup({
   if (conversation.isGroupV2()) {
     try {
       return await sendToGroupViaSenderKey({
+        contentHint,
         contentMessage,
         conversation,
         isPartialSend,
@@ -127,10 +132,15 @@ export async function sendContentMessageToGroup({
     }
   }
 
+  const groupId = conversation.isGroupV2()
+    ? conversation.get('groupId')
+    : undefined;
   return window.textsecure.messaging.sendGroupProto(
     recipients,
     contentMessage,
     timestamp,
+    contentHint,
+    groupId,
     sendOptions
   );
 }
@@ -138,6 +148,7 @@ export async function sendContentMessageToGroup({
 // The Primary Sender Key workflow
 
 export async function sendToGroupViaSenderKey(options: {
+  contentHint: number;
   contentMessage: ContentClass;
   conversation: ConversationModel;
   isPartialSend?: boolean;
@@ -148,6 +159,7 @@ export async function sendToGroupViaSenderKey(options: {
   timestamp: number;
 }): Promise<CallbackResultType> {
   const {
+    contentHint,
     contentMessage,
     conversation,
     isPartialSend,
@@ -157,6 +169,9 @@ export async function sendToGroupViaSenderKey(options: {
     sendOptions,
     timestamp,
   } = options;
+  const {
+    ContentHint,
+  } = window.textsecure.protobuf.UnidentifiedSenderMessage.Message;
 
   const logId = conversation.idForLogging();
   window.log.info(
@@ -176,6 +191,15 @@ export async function sendToGroupViaSenderKey(options: {
     );
   }
 
+  if (
+    contentHint !== ContentHint.RESENDABLE &&
+    contentHint !== ContentHint.SUPPLEMENTARY
+  ) {
+    throw new Error(
+      `sendToGroupViaSenderKey/${logId}: Invalid contentHint ${contentHint}`
+    );
+  }
+
   assert(
     window.textsecure.messaging,
     'sendToGroupViaSenderKey: textsecure.messaging not available!'
@@ -293,10 +317,15 @@ export async function sendToGroupViaSenderKey(options: {
         newToMemberUuids.length
       } members: ${JSON.stringify(newToMemberUuids)}`
     );
-    await window.textsecure.messaging.sendSenderKeyDistributionMessage({
-      distributionId,
-      identifiers: newToMemberUuids,
-    });
+    await window.textsecure.messaging.sendSenderKeyDistributionMessage(
+      {
+        contentHint: ContentHint.SUPPLEMENTARY,
+        distributionId,
+        groupId,
+        identifiers: newToMemberUuids,
+      },
+      sendOptions
+    );
   }
 
   // 9. Update memberDevices with both adds and the removals which didn't require a reset.
@@ -323,6 +352,7 @@ export async function sendToGroupViaSenderKey(options: {
   // 10. Send the Sender Key message!
   try {
     const messageBuffer = await encryptForSenderKey({
+      contentHint,
       devices: devicesForSenderKey,
       distributionId,
       contentMessage: contentMessage.toArrayBuffer(),
@@ -396,6 +426,8 @@ export async function sendToGroupViaSenderKey(options: {
     normalRecipients,
     contentMessage,
     timestamp,
+    contentHint,
+    groupId,
     sendOptions
   );
 
@@ -594,14 +626,16 @@ function getXorOfAccessKeys(devices: Array<DeviceType>): Buffer {
 }
 
 async function encryptForSenderKey({
+  contentHint,
+  contentMessage,
   devices,
   distributionId,
-  contentMessage,
   groupId,
 }: {
+  contentHint: number;
+  contentMessage: ArrayBuffer;
   devices: Array<DeviceType>;
   distributionId: string;
-  contentMessage: ArrayBuffer;
   groupId: string;
 }): Promise<Buffer> {
   const ourUuid = window.textsecure.storage.user.getUuid();
@@ -625,7 +659,6 @@ async function encryptForSenderKey({
     () => groupEncrypt(sender, distributionId, senderKeyStore, message)
   );
 
-  const contentHint = 1;
   const groupIdBuffer = Buffer.from(groupId, 'base64');
   const senderCertificateObject = await senderCertificateService.get(
     SenderCertificateMode.WithoutE164
@@ -676,8 +709,8 @@ function isValidSenderKeyRecipient(
     return false;
   }
 
-  const { capabilities } = memberConversation.attributes;
-  if (!capabilities.senderKey) {
+  const capabilities = memberConversation.get('capabilities');
+  if (!capabilities?.senderKey) {
     window.log.info(
       `isValidSenderKeyRecipient: Missing senderKey capability for member ${uuid}`
     );