Fix call history read syncs

ts/sql/Interface.ts

@@ -787,8 +787,8 @@ type WritableInterface = {
   markCallHistoryDeleted: (callId: string) => void;
   cleanupCallHistoryMessages: () => void;
   markCallHistoryRead(callId: string): void;
-  markAllCallHistoryRead(target: CallLogEventTarget): void;
-  markAllCallHistoryReadInConversation(target: CallLogEventTarget): void;
+  markAllCallHistoryRead(target: CallLogEventTarget): number;
+  markAllCallHistoryReadInConversation(target: CallLogEventTarget): number;
   saveCallHistory(callHistory: CallHistoryDetails): void;
   markCallHistoryMissed(callIds: ReadonlyArray<string>): void;
   getRecentStaleRingsAndMarkOlderMissed(): ReadonlyArray<MaybeStaleCallHistory>;

ts/sql/Server.ts

@@ -3491,7 +3491,12 @@ function clearCallHistory(
   target: CallLogEventTarget
 ): ReadonlyArray<string> {
   return db.transaction(() => {
-    const timestamp = getMessageTimestampForCallLogEventTarget(db, target);
+    const callHistory = getCallHistoryForCallLogEventTarget(db, target);
+    if (callHistory == null) {
+      logger.error('clearCallHistory: Target call not found');
+      return [];
+    }
+    const { timestamp } = callHistory;
 
     const [selectCallsQuery, selectCallsParams] = sql`
       SELECT callsHistory.callId
@@ -3650,12 +3655,13 @@ function markCallHistoryRead(db: WritableDB, callId: string): void {
   db.prepare(query).run(params);
 }
 
-function getMessageTimestampForCallLogEventTarget(
+function getCallHistoryForCallLogEventTarget(
   db: ReadableDB,
   target: CallLogEventTarget
-): number {
-  let { callId, peerId } = target;
-  const { timestamp } = target;
+): CallHistoryDetails | null {
+  const { callId, peerId, timestamp } = target;
+
+  let row: unknown;
 
   if (callId == null || peerId == null) {
     const predicate =
@@ -3665,7 +3671,7 @@ function getMessageTimestampForCallLogEventTarget(
 
     // Get the most recent call history timestamp for the target.timestamp
     const [selectQuery, selectParams] = sql`
-      SELECT callsHistory.callId, callsHistory.peerId
+      SELECT *
       FROM callsHistory
       WHERE ${predicate}
         AND callsHistory.timestamp <= ${timestamp}
@@ -3673,54 +3679,130 @@ function getMessageTimestampForCallLogEventTarget(
       LIMIT 1
     `;
 
-    const row = db.prepare(selectQuery).get(selectParams);
-    if (row == null) {
-      logger.warn('getTimestampForCallLogEventTarget: Target call not found');
-      return timestamp;
-    }
-    callId = row.callId as string;
-    peerId = row.peerId as AciString;
+    row = db.prepare(selectQuery).get(selectParams);
+  } else {
+    const [selectQuery, selectParams] = sql`
+      SELECT *
+      FROM callsHistory
+      WHERE callsHistory.peerId IS ${target.peerId}
+        AND callsHistory.callId IS ${target.callId}
+      LIMIT 1
+    `;
+
+    row = db.prepare(selectQuery).get(selectParams);
   }
 
+  if (row == null) {
+    return null;
+  }
+
+  return callHistoryDetailsSchema.parse(row);
+}
+
+function getConversationIdForCallHistory(
+  db: ReadableDB,
+  callHistory: CallHistoryDetails
+): string | null {
+  const { peerId, mode } = callHistory;
+
+  if (mode === CallMode.Adhoc) {
+    throw new Error(
+      'getConversationIdForCallHistory: Adhoc calls do not have conversations'
+    );
+  }
+
+  const predicate =
+    mode === CallMode.Direct
+      ? sqlFragment`serviceId IS ${peerId}`
+      : sqlFragment`groupId IS ${peerId}`;
+
+  const [selectConversationIdQuery, selectConversationIdParams] = sql`
+    SELECT id FROM conversations
+    WHERE ${predicate}
+  `;
+
+  const conversationId = db
+    .prepare(selectConversationIdQuery)
+    .pluck()
+    .get(selectConversationIdParams);
+
+  if (typeof conversationId !== 'string') {
+    logger.warn('getConversationIdForCallHistory: Unknown conversation');
+    return null;
+  }
+
+  return conversationId ?? null;
+}
+
+function getMessageReceivedAtForCall(
+  db: ReadableDB,
+  callId: string,
+  conversationId: string
+): number | null {
   const [selectQuery, selectParams] = sql`
-    SELECT messages.sent_at
+    SELECT messages.received_at
     FROM messages
     WHERE messages.type IS 'call-history'
-      AND messages.conversationId IS ${peerId}
+      AND messages.conversationId IS ${conversationId}
       AND messages.callId IS ${callId}
     LIMIT 1
   `;
 
-  const messageTimestamp = db.prepare(selectQuery).pluck().get(selectParams);
-
-  if (messageTimestamp == null) {
-    logger.warn(
-      'getTimestampForCallLogEventTarget: Target call message not found'
-    );
+  const receivedAt = db.prepare(selectQuery).pluck().get(selectParams);
+  if (receivedAt == null) {
+    logger.warn('getMessageReceivedAtForCall: Target call message not found');
   }
 
-  return messageTimestamp ?? target.timestamp;
+  return receivedAt ?? null;
 }
 
 export function markAllCallHistoryRead(
   db: WritableDB,
   target: CallLogEventTarget,
   inConversation = false
-): void {
+): number {
   if (inConversation) {
     strictAssert(target.peerId, 'peerId is required');
   }
 
-  db.transaction(() => {
+  return db.transaction(() => {
+    const callHistory = getCallHistoryForCallLogEventTarget(db, target);
+    if (callHistory == null) {
+      logger.warn('markAllCallHistoryRead: Target call not found');
+      return 0;
+    }
+
+    const { callId } = callHistory;
+
+    strictAssert(
+      target.callId == null || callId === target.callId,
+      'Call ID must be the same as target if supplied'
+    );
+
+    const conversationId = getConversationIdForCallHistory(db, callHistory);
+    if (conversationId == null) {
+      logger.warn('markAllCallHistoryRead: Conversation not found for call');
+      return 0;
+    }
+    logger.info(`markAllCallHistoryRead: Found conversation ${conversationId}`);
+    const receivedAt = getMessageReceivedAtForCall(db, callId, conversationId);
+
+    if (receivedAt == null) {
+      logger.warn('markAllCallHistoryRead: Message not found for call');
+      return 0;
+    }
+
+    const predicate = inConversation
+      ? sqlFragment`messages.conversationId IS ${conversationId}`
+      : sqlFragment`TRUE`;
+
     const jsonPatch = JSON.stringify({
       seenStatus: SeenStatus.Seen,
     });
 
-    const timestamp = getMessageTimestampForCallLogEventTarget(db, target);
-
-    const predicate = inConversation
-      ? sqlFragment`messages.conversationId IS ${target.peerId}`
-      : sqlFragment`TRUE`;
+    logger.warn(
+      `markAllCallHistoryRead: Marking calls before ${receivedAt} read`
+    );
 
     const [updateQuery, updateParams] = sql`
       UPDATE messages
@@ -3730,19 +3812,20 @@ export function markAllCallHistoryRead(
       WHERE messages.type IS 'call-history'
         AND ${predicate}
         AND messages.seenStatus IS ${SEEN_STATUS_UNSEEN}
-        AND messages.sent_at <= ${timestamp}
+        AND messages.received_at <= ${receivedAt};
     `;
 
-    db.prepare(updateQuery).run(updateParams);
+    const result = db.prepare(updateQuery).run(updateParams);
+    return result.changes;
   })();
 }
 
 function markAllCallHistoryReadInConversation(
   db: WritableDB,
   target: CallLogEventTarget
-): void {
+): number {
   strictAssert(target.peerId, 'peerId is required');
-  markAllCallHistoryRead(db, target, true);
+  return markAllCallHistoryRead(db, target, true);
 }
 
 function getCallHistoryGroupData(

ts/sql/migrations/1170-update-call-history-unread-index.ts

@@ -0,0 +1,29 @@
+// Copyright 2024 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+import type { Database } from '@signalapp/better-sqlite3';
+import type { LoggerType } from '../../types/Logging';
+import { sql } from '../util';
+
+export const version = 1170;
+export function updateToSchemaVersion1170(
+  currentVersion: number,
+  db: Database,
+  logger: LoggerType
+): void {
+  if (currentVersion >= 1170) {
+    return;
+  }
+
+  db.transaction(() => {
+    const [query] = sql`
+      DROP INDEX IF EXISTS messages_callHistory_markReadBefore;
+      CREATE INDEX messages_callHistory_markReadBefore
+        ON messages (type, seenStatus, received_at DESC)
+        WHERE type IS 'call-history';
+    `;
+    db.exec(query);
+
+    db.pragma('user_version = 1170');
+  })();
+  logger.info('updateToSchemaVersion1170: success!');
+}

ts/sql/migrations/index.ts

@@ -92,10 +92,11 @@ import { updateToSchemaVersion1120 } from './1120-messages-foreign-keys-indexes'
 import { updateToSchemaVersion1130 } from './1130-isStory-index';
 import { updateToSchemaVersion1140 } from './1140-call-links-deleted-column';
 import { updateToSchemaVersion1150 } from './1150-expire-timer-version';
+import { updateToSchemaVersion1160 } from './1160-optimize-calls-unread-count';
 import {
-  updateToSchemaVersion1160,
+  updateToSchemaVersion1170,
   version as MAX_VERSION,
-} from './1160-optimize-calls-unread-count';
+} from './1170-update-call-history-unread-index';
 
 function updateToSchemaVersion1(
   currentVersion: number,
@@ -2056,6 +2057,7 @@ export const SCHEMA_VERSIONS = [
   updateToSchemaVersion1140,
   updateToSchemaVersion1150,
   updateToSchemaVersion1160,
+  updateToSchemaVersion1170,
 ];
 
 export class DBVersionFromFutureError extends Error {

ts/test-node/sql/migration_1100_test.ts

@@ -19,7 +19,8 @@ describe('SQL/updateToSchemaVersion1100', () => {
   let db: WritableDB;
   beforeEach(() => {
     db = createDB();
-    updateToVersion(db, 1100);
+    // index updated in 1170
+    updateToVersion(db, 1170);
   });
 
   afterEach(() => {
@@ -29,14 +30,26 @@ describe('SQL/updateToSchemaVersion1100', () => {
   describe('Optimize markAllCallHistoryReadInConversation', () => {
     it('is fast', () => {
       const COUNT = 10_000;
+      const CONVERSATIONS = 30;
+
+      const conversations = Array.from(
+        { length: CONVERSATIONS },
+        (_, index) => {
+          return {
+            id: `test-conversation-${index}`,
+            groupId: `test-conversation-${index}`,
+            serviceId: `test-conversation-${index}`,
+          };
+        }
+      );
 
       const messages = Array.from({ length: COUNT }, (_, index) => {
         return {
           id: `test-message-${index}`,
           type: 'call-history',
           seenStatus: SeenStatus.Unseen,
-          conversationId: `test-conversation-${index % 30}`,
-          sent_at: index,
+          conversationId: `test-conversation-${index % CONVERSATIONS}`,
+          received_at: index,
           json: {
             callId: `test-call-${index}`,
           },
@@ -46,7 +59,7 @@ describe('SQL/updateToSchemaVersion1100', () => {
       const callsHistory = Array.from({ length: COUNT }, (_, index) => {
         return {
           callId: `test-call-${index}`,
-          peerId: `test-conversation-${index % 30}`,
+          peerId: `test-conversation-${index % CONVERSATIONS}`,
           timestamp: index,
           ringerId: null,
           mode: CallMode.Direct,
@@ -56,6 +69,7 @@ describe('SQL/updateToSchemaVersion1100', () => {
         };
       });
 
+      insertData(db, 'conversations', conversations);
       insertData(db, 'messages', messages);
       insertData(db, 'callsHistory', callsHistory);
 
@@ -72,8 +86,9 @@ describe('SQL/updateToSchemaVersion1100', () => {
       };
 
       const start = performance.now();
-      markAllCallHistoryRead(db, target, true);
+      const changes = markAllCallHistoryRead(db, target, true);
       const end = performance.now();
+      assert.equal(changes, Math.ceil(COUNT / CONVERSATIONS));
       assert.isBelow(end - start, 50);
     });
   });

ts/util/callDisposition.ts

@@ -1324,12 +1324,17 @@ export async function markAllCallHistoryReadAndSync(
     log.info(
       `markAllCallHistoryReadAndSync: Marking call history read before (${latestCall.callId}, ${latestCall.timestamp})`
     );
+    let count: number;
     if (inConversation) {
-      await DataWriter.markAllCallHistoryReadInConversation(latestCall);
+      count = await DataWriter.markAllCallHistoryReadInConversation(latestCall);
     } else {
-      await DataWriter.markAllCallHistoryRead(latestCall);
+      count = await DataWriter.markAllCallHistoryRead(latestCall);
     }
 
+    log.info(
+      `markAllCallHistoryReadAndSync: Marked ${count} call history messages read`
+    );
+
     const ourAci = window.textsecure.storage.user.getCheckedAci();
 
     const callLogEvent = new Proto.SyncMessage.CallLogEvent({

ts/util/onCallLogEventSync.ts

@@ -39,7 +39,10 @@ export async function onCallLogEventSync(
   } else if (type === CallLogEvent.MarkedAsRead) {
     log.info('onCallLogEventSync: Marking call history read');
     try {
-      await DataWriter.markAllCallHistoryRead(target);
+      const count = await DataWriter.markAllCallHistoryRead(target);
+      log.info(
+        `onCallLogEventSync: Marked ${count} call history messages read`
+      );
     } finally {
       window.reduxActions.callHistory.updateCallHistoryUnreadCount();
     }
@@ -48,7 +51,11 @@ export async function onCallLogEventSync(
     log.info('onCallLogEventSync: Marking call history read in conversation');
     try {
       strictAssert(peerId, 'Missing peerId');
-      await DataWriter.markAllCallHistoryReadInConversation(target);
+      const count =
+        await DataWriter.markAllCallHistoryReadInConversation(target);
+      log.info(
+        `onCallLogEventSync: Marked ${count} call history messages read`
+      );
     } finally {
       window.reduxActions.callHistory.updateCallHistoryUnreadCount();
     }