Reduce thrashing on notification state during message backlog processing.

2026-09-20 00:35:47 +01:00 · 2026-08-26 15:22:20 -04:00
parent eec1b4a987
commit 9f3fc652fc
12 changed files with 399 additions and 93 deletions
@@ -135,6 +135,7 @@ import org.thoughtcrime.securesms.mms.MmsException
 import org.thoughtcrime.securesms.mms.OutgoingMessage
 import org.thoughtcrime.securesms.mms.QuoteModel
 import org.thoughtcrime.securesms.mms.SlideDeck
+import org.thoughtcrime.securesms.notifications.v2.ConversationId
 import org.thoughtcrime.securesms.notifications.v2.DefaultMessageNotifier.StickyThread
 import org.thoughtcrime.securesms.polls.Poll
 import org.thoughtcrime.securesms.polls.PollRecord
@@ -2572,18 +2573,77 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
    }
  }

-  fun markAsNotified(id: Long) {
+  fun markAsNotified(ids: Collection<Long>) {
+    if (ids.isEmpty()) {
+      return
+    }
+
+    val now = System.currentTimeMillis()
+    val idQuery = SqlUtil.buildFastCollectionQuery(ID, ids)
+    val originalQuery = SqlUtil.buildFastCollectionQuery(ORIGINAL_MESSAGE_ID, ids)
+    val revisionQuery = SqlUtil.buildFastCollectionQuery(LATEST_REVISION_ID, ids)
+
    writableDatabase
      .update(TABLE_NAME)
      .values(
        NOTIFIED to 1,
-        REACTIONS_LAST_SEEN to System.currentTimeMillis(),
-        VOTES_LAST_SEEN to System.currentTimeMillis()
+        REACTIONS_LAST_SEEN to now,
+        VOTES_LAST_SEEN to now
+      )
+      .where(
+        "(${idQuery.where}) OR (${originalQuery.where}) OR (${revisionQuery.where})",
+        idQuery.whereArgs + originalQuery.whereArgs + revisionQuery.whereArgs
      )
-      .where("$ID = ? OR $ORIGINAL_MESSAGE_ID = ? OR $LATEST_REVISION_ID = ?", id, id, id)
      .run()
  }

+  /**
+   * Marks everything not yet notified in [conversationIds] as notified, up to and including [maxMessageId].
+   */
+  fun markConversationsAsNotified(conversationIds: Collection<ConversationId>, maxMessageId: Long) {
+    if (conversationIds.isEmpty() || maxMessageId <= 0) {
+      return
+    }
+
+    val now = System.currentTimeMillis()
+    val (storyReplies, chats) = conversationIds.partition { it.groupStoryId != null }
+
+    writableDatabase.withinTransaction { db ->
+      if (chats.isNotEmpty()) {
+        val query = SqlUtil.buildFastCollectionQuery(
+          column = THREAD_ID,
+          values = chats.map { it.threadId }.distinct(),
+          prefix = "$NOTIFIED = 0 AND $STORY_TYPE = 0 AND $PARENT_STORY_ID <= 0 AND $ID <= $maxMessageId AND"
+        )
+
+        db.update(TABLE_NAME)
+          .values(
+            NOTIFIED to 1,
+            REACTIONS_LAST_SEEN to now,
+            VOTES_LAST_SEEN to now
+          )
+          .where(query.where, query.whereArgs)
+          .run()
+      }
+
+      storyReplies.forEach { conversationId ->
+        db.update(TABLE_NAME)
+          .values(
+            NOTIFIED to 1,
+            REACTIONS_LAST_SEEN to now,
+            VOTES_LAST_SEEN to now
+          )
+          .where(
+            "$NOTIFIED = 0 AND $STORY_TYPE = 0 AND $PARENT_STORY_ID = ? AND $THREAD_ID = ? AND $ID <= ?",
+            conversationId.groupStoryId!!,
+            conversationId.threadId,
+            maxMessageId
+          )
+          .run()
+      }
+    }
+  }
+
  fun markAsNotNotified(id: Long) {
    writableDatabase
      .update(TABLE_NAME)
@@ -16,18 +16,16 @@ public class DeleteNotificationReceiver extends BroadcastReceiver {

  public static String DELETE_NOTIFICATION_ACTION = "org.thoughtcrime.securesms.DELETE_NOTIFICATION";

-  public static final String EXTRA_IDS     = "message_ids";
-  public static final String EXTRA_MMS     = "is_mms";
-  public static final String EXTRA_THREADS = "threads";
+  public static final String EXTRA_MAX_MESSAGE_ID = "max_message_id";
+  public static final String EXTRA_THREADS        = "threads";

  @Override
  public void onReceive(final Context context, Intent intent) {
    if (DELETE_NOTIFICATION_ACTION.equals(intent.getAction())) {
      MessageNotifier notifier = AppDependencies.getMessageNotifier();

-      final long[]                        ids     = intent.getLongArrayExtra(EXTRA_IDS);
-      final boolean[]                 mms     = intent.getBooleanArrayExtra(EXTRA_MMS);
-      final ArrayList<ConversationId> threads = intent.getParcelableArrayListExtra(EXTRA_THREADS);
+      final long                      maxMessageId = intent.getLongExtra(EXTRA_MAX_MESSAGE_ID, 0);
+      final ArrayList<ConversationId> threads      = intent.getParcelableArrayListExtra(EXTRA_THREADS);

      if (threads != null) {
        for (ConversationId thread : threads) {
@@ -35,18 +33,12 @@ public class DeleteNotificationReceiver extends BroadcastReceiver {
        }
      }

-      if (ids == null || mms == null || ids.length != mms.length) return;
+      if (threads == null || threads.isEmpty() || maxMessageId <= 0) return;

      PendingResult finisher = goAsync();

      SignalExecutors.BOUNDED.execute(() -> {
-        for (int i = 0; i < ids.length; i++) {
-          if (!mms[i]) {
-            SignalDatabase.messages().markAsNotified(ids[i]);
-          } else {
-            SignalDatabase.messages().markAsNotified(ids[i]);
-          }
-        }
+        SignalDatabase.messages().markConversationsAsNotified(threads, maxMessageId);
        finisher.finish();
      });
    }
@@ -12,10 +12,10 @@ import org.signal.core.util.ExceptionUtil;
 import org.signal.core.util.ThreadUtil;
 import org.signal.core.util.concurrent.SignalExecutors;
 import org.thoughtcrime.securesms.database.SignalDatabase;
+import org.thoughtcrime.securesms.dependencies.AppDependencies;
 import org.thoughtcrime.securesms.notifications.v2.DefaultMessageNotifier;
 import org.thoughtcrime.securesms.notifications.v2.ConversationId;
 import org.thoughtcrime.securesms.recipients.Recipient;
-import org.thoughtcrime.securesms.util.BubbleUtil;
 import org.thoughtcrime.securesms.util.LeakyBucketLimiter;

 import java.util.Optional;
@@ -30,10 +30,12 @@ public class OptimizedMessageNotifier implements MessageNotifier {

  private static final String DEDUPE_KEY_GENERAL        = "MESSAGE_NOTIFIER_DEFAULT";
  private static final String DEDUPE_KEY_CHAT           = "MESSAGE_NOTIFIER_CHAT_";
+  private static final long   DRIP_INTERVAL_DRAINED_MS  = 1_000;
+  private static final long   DRIP_INTERVAL_DRAINING_MS = 10_000;

  @MainThread
  public OptimizedMessageNotifier(@NonNull Application context) {
-    this.limiter                = new LeakyBucketLimiter(3, 1000, new Handler(SignalExecutors.getAndStartHandlerThread("signal-notifier", ThreadUtil.PRIORITY_IMPORTANT_BACKGROUND_THREAD).getLooper()));
+    this.limiter                = new LeakyBucketLimiter(3, OptimizedMessageNotifier::dripInterval, new Handler(SignalExecutors.getAndStartHandlerThread("signal-notifier", ThreadUtil.PRIORITY_IMPORTANT_BACKGROUND_THREAD).getLooper()));
    this.defaultMessageNotifier = new DefaultMessageNotifier(context);
  }

@@ -110,7 +112,7 @@ public class OptimizedMessageNotifier implements MessageNotifier {
  @Override
  public void forceBubbleNotification(@NonNull Context context, @NonNull ConversationId conversationId) {
    SignalDatabase.runPostSuccessfulTransaction(() -> {
-      runOnLimiter(() -> getNotifier().forceBubbleNotification(context, conversationId));
+      runWithoutLimit(() -> getNotifier().forceBubbleNotification(context, conversationId));
    });
  }

@@ -129,14 +131,26 @@ public class OptimizedMessageNotifier implements MessageNotifier {
  }

  private void runOnLimiter(@NonNull Runnable runnable) {
+    limiter.run(withCallerStackTrace(runnable));
+  }
+
+  private void runWithoutLimit(@NonNull Runnable runnable) {
+    limiter.runWithoutLimit(withCallerStackTrace(runnable));
+  }
+
+  private static Runnable withCallerStackTrace(@NonNull Runnable runnable) {
    Throwable prettyException = new Throwable();
-    limiter.run(() -> {
+    return () -> {
      try {
        runnable.run();
      } catch (RuntimeException e) {
        throw ExceptionUtil.joinStackTrace(e, prettyException);
      }
-    });
+    };
+  }
+
+  private static long dripInterval() {
+    return AppDependencies.getIncomingMessageObserver().getDecryptionDrained() ? DRIP_INTERVAL_DRAINED_MS : DRIP_INTERVAL_DRAINING_MS;
  }

  private MessageNotifier getNotifier() {
@@ -60,15 +60,15 @@ class DefaultMessageNotifier(context: Application) : MessageNotifier {
  @Volatile private var previousState: NotificationState = NotificationState.EMPTY

  private val threadReminders: MutableMap<ConversationId, Reminder> = ConcurrentHashMap()
-  private val stickyThreads: MutableMap<ConversationId, StickyThread> = mutableMapOf()
+  private val stickyThreads: MutableMap<ConversationId, StickyThread> = ConcurrentHashMap()
  private val lastThreadNotification: MutableMap<ConversationId, Long> = ConcurrentHashMap()

  private val executor = CancelableExecutor()

  override fun setVisibleThread(conversationId: ConversationId?) {
    visibleThread.set(conversationId)
-    stickyThreads.remove(conversationId)
    if (conversationId != null) {
+      stickyThreads.remove(conversationId)
      lastThreadNotification.remove(conversationId)
    }
  }
@@ -155,37 +155,26 @@ class DefaultMessageNotifier(context: Application) : MessageNotifier {
    val notificationProfile: NotificationProfile? = NotificationProfiles.getActiveProfile(SignalDatabase.notificationProfiles.getProfiles())

    Log.internal().i(TAG, "sticky thread: $stickyThreads active profile: ${notificationProfile?.id ?: "none" }")
-    var state: NotificationState = NotificationStateProvider.constructNotificationState(stickyThreads, notificationProfile)
-    Log.internal().i(TAG, "state: $state")
+    val state: NotificationState = NotificationStateProvider.constructNotificationState(stickyThreads, notificationProfile)

    if (state.muteFilteredMessages.isNotEmpty()) {
      Log.i(TAG, "Marking ${state.muteFilteredMessages.size} muted messages as notified to skip notification")
-      state.muteFilteredMessages.forEach { item ->
-        SignalDatabase.messages.markAsNotified(item.id)
-      }
+      SignalDatabase.messages.markAsNotified(state.muteFilteredMessages.map { it.id })
    }

    if (state.profileFilteredMessages.isNotEmpty()) {
      Log.i(TAG, "Marking ${state.profileFilteredMessages.size} profile filtered messages as notified to skip notification")
-      state.profileFilteredMessages.forEach { item ->
-        SignalDatabase.messages.markAsNotified(item.id)
-      }
+      SignalDatabase.messages.markAsNotified(state.profileFilteredMessages.map { it.id })
    }

    if (state.reactionsDisabledFilteredMessages.isNotEmpty()) {
      Log.i(TAG, "Marking ${state.reactionsDisabledFilteredMessages.size} reactions as notified to skip notification")
-      state.reactionsDisabledFilteredMessages.forEach { item ->
-        SignalDatabase.messages.markAsNotified(item.id)
-      }
+      SignalDatabase.messages.markAsNotified(state.reactionsDisabledFilteredMessages.map { it.id })
    }

    if (!SignalStore.settings.isMessageNotificationsEnabled) {
      Log.i(TAG, "Marking ${state.conversations.size} conversations as notified to skip notification")
-      state.conversations.forEach { conversation ->
-        conversation.notificationItems.forEach { item ->
-          SignalDatabase.messages.markAsNotified(item.id)
-        }
-      }
+      SignalDatabase.messages.markConversationsAsNotified(state.conversations.map { it.thread }, state.notificationItems.maxOfOrNull { it.id } ?: 0L)
      return
    }

@@ -37,8 +37,6 @@ import java.util.Optional
 import androidx.core.app.Person as PersonCompat
 import org.signal.core.ui.R as CoreUiR

-private const val BIG_PICTURE_DIMEN = 500
-
 /**
 * Wraps the compat and OS versions of the Notification builders so we can more easily access native
 * features in newer versions. Also provides some domain-specific helpers.
@@ -199,6 +197,13 @@ sealed class NotificationBuilder(protected val context: Context) {
   * Notification builder using solely androidx/compat libraries.
   */
  private class NotificationBuilderCompat(context: Context) : NotificationBuilder(context) {
+    companion object {
+      private const val BIG_PICTURE_DIMEN = 500
+
+      /** Cap on messages rendered into a single notification. */
+      private const val MAX_DISPLAYED_MESSAGES = 25
+    }
+
    val builder: NotificationCompat.Builder = NotificationCompat.Builder(context, NotificationChannels.getInstance().messagesChannel)

    override fun addActions(replyMethod: ReplyMethod, conversation: NotificationConversation) {
@@ -299,7 +304,7 @@ sealed class NotificationBuilder(protected val context: Context) {
      messagingStyle.conversationTitle = conversation.getConversationTitle(context)
      messagingStyle.isGroupConversation = conversation.isGroup

-      conversation.notificationItems.forEach { notificationItem ->
+      conversation.notificationItems.takeLast(MAX_DISPLAYED_MESSAGES).forEach { notificationItem ->
        var person: PersonCompat? = null
        val isNoteToSelf = notificationItem.isPersonSelf && conversation.recipient.isSelf

@@ -332,7 +337,7 @@ sealed class NotificationBuilder(protected val context: Context) {

      val style: NotificationCompat.InboxStyle = NotificationCompat.InboxStyle()

-      for (notificationItem: NotificationItem in state.notificationItems) {
+      for (notificationItem: NotificationItem in state.notificationItems.takeLast(MAX_DISPLAYED_MESSAGES)) {
        val line: CharSequence? = notificationItem.getInboxLine(context)
        if (line != null) {
          style.addLine(line)
@@ -152,17 +152,9 @@ data class NotificationConversation(
  }

  fun getDeleteIntent(context: Context): PendingIntent? {
-    val ids = LongArray(notificationItems.size)
-    val mms = BooleanArray(ids.size)
-    notificationItems.forEachIndexed { index, notificationItem ->
-      ids[index] = notificationItem.id
-      mms[index] = notificationItem.isMms
-    }
-
    val intent = Intent(context, DeleteNotificationReceiver::class.java)
      .setAction(DeleteNotificationReceiver.DELETE_NOTIFICATION_ACTION)
-      .putExtra(DeleteNotificationReceiver.EXTRA_IDS, ids)
-      .putExtra(DeleteNotificationReceiver.EXTRA_MMS, mms)
+      .putExtra(DeleteNotificationReceiver.EXTRA_MAX_MESSAGE_ID, notificationItems.maxOfOrNull { it.id } ?: 0L)
      .putParcelableArrayListExtra(DeleteNotificationReceiver.EXTRA_THREADS, arrayListOf(thread))
      .makeUniqueToPreventMerging()

@@ -48,8 +48,9 @@ object NotificationFactory {

  val TAG: String = Log.tag(NotificationFactory::class.java)

-  private val STILL_DECRYPTING_INDIVIDUAL_THROTTLE: Duration = 5.seconds
+  private val INDIVIDUAL_THROTTLE: Duration = 5.seconds
  private val GROUP_THROTTLE: Duration = 20.seconds
+  private val STILL_DECRYPTING_THROTTLE: Duration = 30.seconds

  @WorkerThread
  fun notify(
@@ -208,9 +209,9 @@ object NotificationFactory {

  private fun shouldAlert(conversation: NotificationConversation, lastNotificationTimestamp: Long, alertOverride: Boolean): Boolean {
    val throttle: Duration = when {
+      !AppDependencies.incomingMessageObserver.decryptionDrained -> STILL_DECRYPTING_THROTTLE
      conversation.recipient.isGroup && (conversation.mostRecentNotification as? MessageNotification)?.hasSelfMention == false -> GROUP_THROTTLE
-      AppDependencies.incomingMessageObserver.decryptionDrained -> STILL_DECRYPTING_INDIVIDUAL_THROTTLE
-      else -> 0.seconds
+      else -> INDIVIDUAL_THROTTLE
    }
    val canAlertBasedOnTime: Boolean = lastNotificationTimestamp < System.currentTimeMillis() - throttle.inWholeMilliseconds || lastNotificationTimestamp > System.currentTimeMillis()
    val isUnreadNoteToSelf: Boolean = conversation.recipient.isSelf && (conversation.mostRecentNotification as? MessageNotification)?.isUnread == true
@@ -54,23 +54,10 @@ data class NotificationState(
  }

  fun getDeleteIntent(context: Context): PendingIntent? {
-    val ids = LongArray(messageCount)
-    val mms = BooleanArray(ids.size)
-    val threads: MutableList<ConversationId> = mutableListOf()
-
-    conversations.forEach { conversation ->
-      threads += conversation.thread
-      conversation.notificationItems.forEachIndexed { index, notificationItem ->
-        ids[index] = notificationItem.id
-        mms[index] = notificationItem.isMms
-      }
-    }
-
    val intent = Intent(context, DeleteNotificationReceiver::class.java)
      .setAction(DeleteNotificationReceiver.DELETE_NOTIFICATION_ACTION)
-      .putExtra(DeleteNotificationReceiver.EXTRA_IDS, ids)
-      .putExtra(DeleteNotificationReceiver.EXTRA_MMS, mms)
-      .putParcelableArrayListExtra(DeleteNotificationReceiver.EXTRA_THREADS, ArrayList(threads))
+      .putExtra(DeleteNotificationReceiver.EXTRA_MAX_MESSAGE_ID, notificationItems.maxOfOrNull { it.id } ?: 0L)
+      .putParcelableArrayListExtra(DeleteNotificationReceiver.EXTRA_THREADS, ArrayList(conversations.map { it.thread }))
      .makeUniqueToPreventMerging()

    return NotificationPendingIntentHelper.getBroadcast(context, 0, intent, PendingIntentFlags.updateCurrent())
@@ -243,28 +243,14 @@ public final class AvatarUtil {
      }
    }

-    @Override
-    public void onDestroy() {
-      Log.d(TAG, "AvatarTarget: onDestroy");
-      super.onDestroy();
-    }
-
-    @Override
-    public void onLoadStarted(@Nullable Drawable placeholder) {
-      Log.d(TAG, "AvatarTarget: onLoadStarted");
-      super.onLoadStarted(placeholder);
-    }
-
    @Override
    public void onResourceReady(@NonNull Bitmap resource, @Nullable Transition<? super Bitmap> transition) {
-      Log.d(TAG, "AvatarTarget: onResourceReady");
      bitmap.set(resource);
      countDownLatch.countDown();
    }

    @Override
    public void onLoadFailed(@Nullable Drawable errorDrawable) {
-      Log.d(TAG, "AvatarTarget: onLoadFailed");
      if (errorDrawable == null) {
        throw new AssertionError("Expected an error drawable.");
      }
@@ -276,7 +262,6 @@ public final class AvatarUtil {

    @Override
    public void onLoadCleared(@Nullable Drawable placeholder) {
-      Log.d(TAG, "AvatarTarget: onLoadCleared");
      bitmap.set(null);
      countDownLatch.countDown();
    }
@@ -8,6 +8,8 @@ import androidx.core.os.HandlerCompat;

 import org.signal.core.util.logging.Log;

+import java.util.function.LongSupplier;
+
 /**
 * Imagine a bucket. Now imagine your tasks as little droplets. As your tasks are thrown into the
 * bucket, the tasks are executed, and the bucket fills up. If the bucket is full, the tasks
@@ -22,6 +24,9 @@ import org.signal.core.util.logging.Log;
 * waited 10 seconds, the bucket would be fully drained, and you'd be able to execute 10 tasks in
 * rapid succession again.
 *
+ * The drip interval is read each time a drip is scheduled, so a caller can vary the rate at runtime.
+ * A change applies to the next drip scheduled, not to one already posted.
+ *
 * This class also does something a little extra -- it keeps track of the most-recently-overflowed
 * task, and will run it the next time it 'drips' instead of leaking. This lets you have a sort of
 * "throw tasks at the bucket and forget about it" attitude, because you know the task will
@@ -37,16 +42,16 @@ public final class LeakyBucketLimiter {

  private static final String TAG = Log.tag(LeakyBucketLimiter.class);

-  private final int     bucketCapacity;
-  private final long    dripInterval;
-  private final Handler handler;
+  private final int          bucketCapacity;
+  private final LongSupplier dripInterval;
+  private final Handler      handler;

  private int      bucketLevel;
  private Runnable lastOverflowedRunnable;

  private final Object RUNNABLE_TOKEN = new Object();

-  public LeakyBucketLimiter(int bucketCapacity, long dripInterval, @NonNull Handler handler) {
+  public LeakyBucketLimiter(int bucketCapacity, @NonNull LongSupplier dripInterval, @NonNull Handler handler) {
    this.bucketCapacity = bucketCapacity;
    this.dripInterval   = dripInterval;
    this.handler        = handler;
@@ -76,10 +81,15 @@ public final class LeakyBucketLimiter {
    }

    if (scheduleDrip) {
-      handler.postDelayed(this::drip, dripInterval);
+      handler.postDelayed(this::drip, dripInterval.getAsLong());
    }
  }

+  @AnyThread
+  public void runWithoutLimit(@NonNull Runnable runnable) {
+    handler.post(runnable);
+  }
+
  private void drip() {
    Runnable runnable  = null;
    boolean  needsDrip = false;
@@ -101,7 +111,7 @@ public final class LeakyBucketLimiter {
    }

    if (needsDrip) {
-      handler.postDelayed(this::drip, dripInterval);
+      handler.postDelayed(this::drip, dripInterval.getAsLong());
    }
  }
 }
@@ -34,6 +34,7 @@ import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.mms.IncomingMessage
 import org.thoughtcrime.securesms.mms.OutgoingMessage
+import org.thoughtcrime.securesms.notifications.v2.ConversationId
 import org.thoughtcrime.securesms.profiles.ProfileName
 import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.recipients.RecipientId
@@ -100,6 +101,7 @@ class ApiPlugin : Plugin {
      )
    ),
    "createThread" to ApiSpec(::createThread, listOf(Param("recipientId", "1"))),
+    "updateNotification" to ApiSpec(::updateNotification, listOf(Param("threadId", "", placeholder = "blank = all threads"))),
    "createMessage" to ApiSpec(
      ::createMessage,
      listOf(
@@ -513,6 +515,30 @@ class ApiPlugin : Plugin {
    }
  }

+  /**
+   * Drives a notification pass on demand.
+   */
+  private fun updateNotification(parameters: Map<String, List<String>>): PluginResult {
+    val threadIdParam = parameters["threadId"]?.firstOrNull()?.takeIf { it.isNotBlank() }
+    val threadId = if (threadIdParam != null) {
+      threadIdParam.toLongOrNull() ?: return PluginResult.ErrorResult(message = "Invalid 'threadId' parameter")
+    } else {
+      null
+    }
+
+    return try {
+      if (threadId != null) {
+        AppDependencies.messageNotifier.updateNotification(AppDependencies.application, ConversationId.forConversation(threadId))
+      } else {
+        AppDependencies.messageNotifier.updateNotification(AppDependencies.application)
+      }
+      UpdateNotificationResponse(threadId ?: -1).toJsonResult()
+    } catch (e: Exception) {
+      Log.w(TAG, "Failed to update notification", e)
+      PluginResult.ErrorResult(message = "Failed: ${e.message}")
+    }
+  }
+
  private fun createMessage(parameters: Map<String, List<String>>): PluginResult {
    val threadId = parameters["threadId"]?.firstOrNull()?.toLongOrNull()
      ?: return PluginResult.ErrorResult(message = "Missing or invalid 'threadId' parameter")
@@ -685,6 +711,10 @@ class ApiPlugin : Plugin {
    @field:JsonProperty val groupId: String
  )

+  data class UpdateNotificationResponse @JsonCreator constructor(
+    @JsonProperty("threadId") val threadId: Long
+  )
+
  data class CreateThreadResponse @JsonCreator constructor(
    @field:JsonProperty val threadId: Long
  )
@@ -0,0 +1,241 @@
+/*
+ * Copyright 2026 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.database
+
+import android.app.Application
+import assertk.assertThat
+import assertk.assertions.isFalse
+import assertk.assertions.isTrue
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+import org.signal.core.util.readToSingleBoolean
+import org.signal.core.util.select
+import org.thoughtcrime.securesms.database.model.MmsMessageRecord
+import org.thoughtcrime.securesms.database.model.ParentStoryId
+import org.thoughtcrime.securesms.mms.IncomingMessage
+import org.thoughtcrime.securesms.notifications.v2.ConversationId
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.testutil.RecipientTestRule
+
+@Suppress("ClassName")
+@RunWith(RobolectricTestRunner::class)
+@Config(manifest = Config.NONE, application = Application::class)
+class MessageTableTest_markNotified {
+
+  @get:Rule
+  val recipients = RecipientTestRule()
+
+  private val messages: MessageTable
+    get() = SignalDatabase.messages
+
+  @Test
+  fun `markAsNotified marks only the given messages`() {
+    val sender = recipients.createRecipient("Alice Bulk")
+    val threadId = threadFor(sender)
+
+    val first = insertIncoming(threadId, sender, time = 1000)
+    val second = insertIncoming(threadId, sender, time = 1001)
+    val untouched = insertIncoming(threadId, sender, time = 1002)
+
+    messages.markAsNotified(listOf(first, second))
+
+    assertThat(isNotified(first)).isTrue()
+    assertThat(isNotified(second)).isTrue()
+    assertThat(isNotified(untouched)).isFalse()
+  }
+
+  @Test
+  fun `markAsNotified with no ids is a no-op`() {
+    val sender = recipients.createRecipient("Bob Empty")
+    val threadId = threadFor(sender)
+    val message = insertIncoming(threadId, sender, time = 1000)
+
+    messages.markAsNotified(emptyList())
+
+    assertThat(isNotified(message)).isFalse()
+  }
+
+  @Test
+  fun `markAsNotified marks every revision of an edited message`() {
+    val sender = recipients.createRecipient("Carol Editor")
+    val threadId = threadFor(sender)
+
+    val original = insertIncoming(threadId, sender, time = 1000)
+    val edit = insertEdit(sender, originalSentTimestamp = 1000, editSentTimeMillis = 1001)
+
+    messages.markAsNotified(listOf(original))
+
+    assertThat(isNotified(original)).isTrue()
+    assertThat(isNotified(edit)).isTrue()
+  }
+
+  @Test
+  fun `markConversationsAsNotified marks unnotified messages up to the bound`() {
+    val sender = recipients.createRecipient("Dave Bound")
+    val threadId = threadFor(sender)
+
+    val first = insertIncoming(threadId, sender, time = 1000)
+    val second = insertIncoming(threadId, sender, time = 1001)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(threadId)), second)
+
+    assertThat(isNotified(first)).isTrue()
+    assertThat(isNotified(second)).isTrue()
+  }
+
+  @Test
+  fun `markConversationsAsNotified leaves messages above the bound alone`() {
+    val sender = recipients.createRecipient("Erin Snapshot")
+    val threadId = threadFor(sender)
+
+    val inSnapshot = insertIncoming(threadId, sender, time = 1000)
+    val arrivedAfter = insertIncoming(threadId, sender, time = 1001)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(threadId)), inSnapshot)
+
+    assertThat(isNotified(inSnapshot)).isTrue()
+    assertThat(isNotified(arrivedAfter)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified bounds by id and not by received time`() {
+    val sender = recipients.createRecipient("Frank Backdated")
+    val threadId = threadFor(sender)
+
+    val inSnapshot = insertIncoming(threadId, sender, time = 5000)
+    val backdated = insertIncoming(threadId, sender, time = 1000)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(threadId)), inSnapshot)
+
+    assertThat(isNotified(inSnapshot)).isTrue()
+    assertThat(isNotified(backdated)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified only touches the given threads`() {
+    val included = recipients.createRecipient("Grace Included")
+    val excluded = recipients.createRecipient("Heidi Excluded")
+    val includedThreadId = threadFor(included)
+    val excludedThreadId = threadFor(excluded)
+
+    val includedMessage = insertIncoming(includedThreadId, included, time = 1000)
+    val excludedMessage = insertIncoming(excludedThreadId, excluded, time = 1001)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(includedThreadId)), excludedMessage)
+
+    assertThat(isNotified(includedMessage)).isTrue()
+    assertThat(isNotified(excludedMessage)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified with no conversations is a no-op`() {
+    val sender = recipients.createRecipient("Ivan None")
+    val threadId = threadFor(sender)
+    val message = insertIncoming(threadId, sender, time = 1000)
+
+    messages.markConversationsAsNotified(emptyList(), message)
+
+    assertThat(isNotified(message)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified with a non-positive bound is a no-op`() {
+    val sender = recipients.createRecipient("Judy Zero")
+    val threadId = threadFor(sender)
+    val message = insertIncoming(threadId, sender, time = 1000)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(threadId)), 0)
+
+    assertThat(isNotified(message)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified for a group story reply leaves the main chat alone`() {
+    val sender = recipients.createRecipient("Karl Story")
+    val group = recipients.createGroup(sender)
+    val threadId = threadFor(group.recipientId)
+
+    val storyId = insertIncoming(threadId, sender, time = 1000)
+    val chatMessage = insertIncoming(threadId, sender, time = 1001)
+    val reply = insertGroupReply(threadId, sender, time = 1002, parentStoryId = storyId)
+
+    messages.markConversationsAsNotified(listOf(ConversationId(threadId, storyId)), reply)
+
+    assertThat(isNotified(reply)).isTrue()
+    assertThat(isNotified(chatMessage)).isFalse()
+  }
+
+  @Test
+  fun `markConversationsAsNotified for a chat leaves group story replies alone`() {
+    val sender = recipients.createRecipient("Lena Chat")
+    val group = recipients.createGroup(sender)
+    val threadId = threadFor(group.recipientId)
+
+    val storyId = insertIncoming(threadId, sender, time = 1000)
+    val chatMessage = insertIncoming(threadId, sender, time = 1001)
+    val reply = insertGroupReply(threadId, sender, time = 1002, parentStoryId = storyId)
+
+    messages.markConversationsAsNotified(listOf(ConversationId.forConversation(threadId)), reply)
+
+    assertThat(isNotified(chatMessage)).isTrue()
+    assertThat(isNotified(reply)).isFalse()
+  }
+
+  private fun threadFor(recipientId: RecipientId): Long {
+    return SignalDatabase.threads.getOrCreateThreadIdFor(Recipient.resolved(recipientId))
+  }
+
+  private fun insertIncoming(threadId: Long, from: RecipientId, time: Long): Long {
+    val message = IncomingMessage(
+      type = MessageType.NORMAL,
+      from = from,
+      sentTimeMillis = time,
+      serverTimeMillis = time,
+      receivedTimeMillis = time,
+      body = "msg $time"
+    )
+    return messages.insertMessageInbox(message, threadId).get().messageId
+  }
+
+  private fun insertGroupReply(threadId: Long, from: RecipientId, time: Long, parentStoryId: Long): Long {
+    val message = IncomingMessage(
+      type = MessageType.NORMAL,
+      from = from,
+      sentTimeMillis = time,
+      serverTimeMillis = time,
+      receivedTimeMillis = time,
+      body = "reply $time",
+      parentStoryId = ParentStoryId.GroupReply(parentStoryId)
+    )
+    return messages.insertMessageInbox(message, threadId).get().messageId
+  }
+
+  private fun insertEdit(from: RecipientId, originalSentTimestamp: Long, editSentTimeMillis: Long): Long {
+    val target = messages.getMessageFor(originalSentTimestamp, from) as MmsMessageRecord
+    val edit = IncomingMessage(
+      type = MessageType.NORMAL,
+      from = from,
+      sentTimeMillis = editSentTimeMillis,
+      serverTimeMillis = editSentTimeMillis,
+      receivedTimeMillis = editSentTimeMillis,
+      body = "edited at $editSentTimeMillis"
+    )
+    return messages.insertEditMessageInbox(edit, target).get().messageId
+  }
+
+  private fun isNotified(messageId: Long): Boolean {
+    return SignalDatabase.writableDatabase
+      .select(MessageTable.NOTIFIED)
+      .from(MessageTable.TABLE_NAME)
+      .where("${MessageTable.ID} = ?", messageId)
+      .run()
+      .readToSingleBoolean()
+  }
+}