New getRecentStoryReplies function to clean up replies in multiple convos

ts/sql/Client.ts

@@ -36,6 +36,7 @@ import type {
   ClientSearchResultMessageType,
   ConversationType,
   GetConversationRangeCenteredOnMessageResultType,
+  GetRecentStoryRepliesOptionsType,
   IdentityKeyIdType,
   IdentityKeyType,
   StoredIdentityKeyType,
@@ -99,6 +100,7 @@ const exclusiveInterface: ClientExclusiveInterface = {
 
   searchMessages,
 
+  getRecentStoryReplies,
   getOlderMessagesByConversation,
   getConversationRangeCenteredOnMessage,
   getNewerMessagesByConversation,
@@ -613,6 +615,15 @@ async function getNewerMessagesByConversation(
   return handleMessageJSON(messages);
 }
 
+async function getRecentStoryReplies(
+  storyId: string,
+  options?: GetRecentStoryRepliesOptionsType
+): Promise<Array<MessageType>> {
+  const messages = await channels.getRecentStoryReplies(storyId, options);
+
+  return handleMessageJSON(messages);
+}
+
 async function getOlderMessagesByConversation(
   options: AdjacentMessagesByConversationOptionsType
 ): Promise<Array<MessageType>> {

ts/sql/Interface.ts

@@ -825,6 +825,11 @@ export type ServerInterface = DataInterface & {
     options?: { limit?: number };
     contactUuidsMatchingQuery?: Array<string>;
   }) => Promise<Array<ServerSearchResultMessageType>>;
+
+  getRecentStoryReplies(
+    storyId: string,
+    options?: GetRecentStoryRepliesOptionsType
+  ): Promise<Array<MessageTypeUnhydrated>>;
   getOlderMessagesByConversation: (
     options: AdjacentMessagesByConversationOptionsType
   ) => Promise<Array<MessageTypeUnhydrated>>;
@@ -895,6 +900,13 @@ export type ServerInterface = DataInterface & {
   getAllBadgeImageFileLocalPaths: () => Promise<Set<string>>;
 };
 
+export type GetRecentStoryRepliesOptionsType = {
+  limit?: number;
+  messageId?: string;
+  receivedAt?: number;
+  sentAt?: number;
+};
+
 // Differing signature on client/server
 export type ClientExclusiveInterface = {
   // Differing signature on client/server
@@ -913,6 +925,11 @@ export type ClientExclusiveInterface = {
     options?: { limit?: number };
     contactUuidsMatchingQuery?: Array<string>;
   }) => Promise<Array<ClientSearchResultMessageType>>;
+
+  getRecentStoryReplies(
+    storyId: string,
+    options?: GetRecentStoryRepliesOptionsType
+  ): Promise<Array<MessageAttributesType>>;
   getOlderMessagesByConversation: (
     options: AdjacentMessagesByConversationOptionsType
   ) => Promise<Array<MessageAttributesType>>;

ts/sql/Server.ts

@@ -93,6 +93,7 @@ import type {
   GetAllStoriesResultType,
   GetConversationRangeCenteredOnMessageResultType,
   GetKnownMessageAttachmentsResultType,
+  GetRecentStoryRepliesOptionsType,
   GetUnreadByConversationAndMarkReadResultType,
   IdentityKeyIdType,
   StoredIdentityKeyType,
@@ -251,6 +252,7 @@ const dataInterface: ServerInterface = {
 
   getMessageCount,
   getStoryCount,
+  getRecentStoryReplies,
   saveMessage,
   saveMessages,
   removeMessage,
@@ -2530,6 +2532,53 @@ enum AdjacentDirection {
   Newer = 'Newer',
 }
 
+async function getRecentStoryReplies(
+  storyId: string,
+  options?: GetRecentStoryRepliesOptionsType
+): Promise<Array<MessageTypeUnhydrated>> {
+  return getRecentStoryRepliesSync(storyId, options);
+}
+
+// This function needs to pull story replies from all conversations, because when we send
+//   a story to one or more distribution lists, each reply to it will be in the sender's
+//   1:1 conversation with us.
+function getRecentStoryRepliesSync(
+  storyId: string,
+  {
+    limit = 100,
+    messageId,
+    receivedAt = Number.MAX_VALUE,
+    sentAt = Number.MAX_VALUE,
+  }: GetRecentStoryRepliesOptionsType = {}
+): Array<MessageTypeUnhydrated> {
+  const db = getInstance();
+  const timeFilters = {
+    first: sqlFragment`received_at = ${receivedAt} AND sent_at < ${sentAt}`,
+    second: sqlFragment`received_at < ${receivedAt}`,
+  };
+
+  const createQuery = (timeFilter: QueryFragment): QueryFragment => sqlFragment`
+    SELECT json FROM messages WHERE
+      (${messageId} IS NULL OR id IS NOT ${messageId}) AND
+      isStory IS 0 AND
+      storyId IS ${storyId} AND
+      (
+        ${timeFilter}
+      )
+      ORDER BY received_at DESC, sent_at DESC
+  `;
+
+  const template = sqlFragment`
+    SELECT first.json FROM (${createQuery(timeFilters.first)}) as first
+    UNION ALL
+    SELECT second.json FROM (${createQuery(timeFilters.second)}) as second
+  `;
+
+  const [query, params] = sql`${template} LIMIT ${limit}`;
+
+  return db.prepare(query).all(params);
+}
+
 function getAdjacentMessagesByConversationSync(
   direction: AdjacentDirection,
   {

ts/sql/migrations/86-story-replies-index.ts

@@ -0,0 +1,32 @@
+// Copyright 2023 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+
+import type { Database } from '@signalapp/better-sqlite3';
+
+import type { LoggerType } from '../../types/Logging';
+
+export default function updateToSchemaVersion86(
+  currentVersion: number,
+  db: Database,
+  logger: LoggerType
+): void {
+  if (currentVersion >= 86) {
+    return;
+  }
+
+  db.transaction(() => {
+    // The key reason for this new schema is that all of our previous schemas start with
+    //   conversationId. This query is meant to find all replies to a given story, no
+    //   matter the conversation.
+    db.exec(
+      `CREATE INDEX messages_story_replies
+        ON messages (storyId, received_at, sent_at)
+        WHERE isStory IS 0;
+      `
+    );
+
+    db.pragma('user_version = 86');
+  })();
+
+  logger.info('updateToSchemaVersion86: success!');
+}

ts/sql/migrations/index.ts

@@ -61,6 +61,7 @@ import updateToSchemaVersion82 from './82-edited-messages-read-index';
 import updateToSchemaVersion83 from './83-mentions';
 import updateToSchemaVersion84 from './84-all-mentions';
 import updateToSchemaVersion85 from './85-add-kyber-keys';
+import updateToSchemaVersion86 from './86-story-replies-index';
 
 function updateToSchemaVersion1(
   currentVersion: number,
@@ -1992,6 +1993,7 @@ export const SCHEMA_VERSIONS = [
   updateToSchemaVersion83,
   updateToSchemaVersion84,
   updateToSchemaVersion85,
+  updateToSchemaVersion86,
 ];
 
 export function updateSchema(db: Database, logger: LoggerType): void {